LangChain on Swarmd

swarmd-langchain wraps LangChain and LangGraph so a compiled graph can live on Swarmd: authenticated, discovering its sub-agents and MCP tools, served over A2A, and rebuilding itself when grants change. The helper trio deliberately mirrors Google ADK — same names, same order, same arguments. If you know one you know the other.

Helper	Returns	In one line
`create_runtime()`	`SwarmDRuntime`	Reads `SWARMD_*` env vars and builds your authenticated clients.
`create_llm_agent(...)`	`CompiledStateGraph`	Picks the model, discovers sub-agents and MCP servers, compiles the graph.
`serve(agent, runtime)`	runs forever	A2A server, agent card, `/admin`, correlation, live rebuild.

Assumes you’ve finished Setup. Ten minutes on Core concepts will make the rest of this page obvious rather than magical — but it isn’t required.

Step 1 — Install

pip install swarmd-langchain

That pulls in swarmd-sdk, langchain, langgraph, langchain-openai, langchain-mcp-adapters, mcp, and a2a-sdk. Use a virtualenv — the version bounds are deliberately tight.

Step 2 — Add your LLM to `.env`

On top of the credentials from Setup:

.env

OPENAI_API_KEY=sk-...
OPENAI_MODEL=gpt-4o   # optional, default gpt-4o

swarmd-langchain does not route LLM calls through the Swarmd gateway. Unlike the ADK wrapper, create_llm_agent() builds a ChatOpenAI directly and raises ValueError without OPENAI_API_KEY. SWARMD_LLM_GATEWAY_ID is ignored here.If you need gateway routing — audited completions, no provider key in the agent — build the model yourself and skip create_llm_agent(); see Going off the rails.

The model is built with temperature=0. To change that, or to use a different provider, assemble the graph yourself.

Step 3 — Write the agent

main.py

from dotenv import load_dotenv

load_dotenv()

from langchain_core.tools import tool

from swarmd_langchain import create_llm_agent, create_runtime, serve


@tool
def get_time(city: str) -> str:
    """Get the current time for any city."""
    return f"The time in {city} is 18:00"


runtime = create_runtime()

agent = create_llm_agent(
    runtime,
    name="time_agent",
    description="A simple agent that provides time information",
    instruction=(
        "You are a helpful time agent. Use the get_time tool to tell users "
        "the time in any city they ask about. You can also collaborate with "
        "other agents and call MCP tools for additional capabilities."
    ),
    tools=[get_time],
)


if __name__ == "__main__":
    serve(agent, runtime)

load_dotenv() must run before the swarmd_langchain import. Module import reads environment variables; a .env loaded afterwards is too late. That’s why the imports sit below the call, against every linter’s advice.

Local tools must be LangChain BaseTools. The @tool decorator is the easy path — it turns a plain function into one, using the docstring as the description the model sees. A bare function passed in tools= will fail when the graph compiles.

Step 4 — Run it

python main.py

==================================================
  time_agent
  http://0.0.0.0:8080
  Agent Card: http://0.0.0.0:8080/.well-known/agent-card.json
==================================================

Discovery reports through the standard logging module at INFO, so you need logging configured to see it:

import logging
logging.basicConfig(level=logging.INFO)

Then you’ll get Found N subscribed agents and Loaded N MCP tool(s) from M server(s).

Seeing 'SwarmD not configured - standalone mode'?

create_runtime() only configures when all four of SWARMD_AGENT_ID, SWARMD_CLIENT_SECRET, SWARMD_BASE_URL and SWARMD_TOKEN_URL are set. Miss one and the agent runs as a plain LangChain agent against direct OpenAI, discovering nothing.Deliberate escape hatch for offline development; also the single most common cause of “why is my catalogue empty?”. Check all four.

Step 5 — Talk to it

curl http://localhost:8080/.well-known/agent-card.json

curl -X POST http://localhost:8080/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0", "id": 1, "method": "message/send",
    "params": {
      "message": {
        "role": "user",
        "parts": [{"kind": "text", "text": "What time is it in Tokyo?"}],
        "messageId": "11111111-1111-1111-1111-111111111111"
      }
    }
  }'

curl http://localhost:8080/admin/status

Dissected: `create_runtime()`

runtime = create_runtime()

Constructs a SwarmDRuntime and, when the four platform env vars are present, calls runtime.configure(...). That builds a SwarmDClient, an McpClient, and the two token manager proxies — swarmd:api and mcp:call.

Dissected: `create_llm_agent()`

create_llm_agent(
    runtime,
    name: str,
    description: str,
    instruction: str,
    tools: Sequence[BaseTool],
) -> CompiledStateGraph

Returns a compiled LangGraph you can invoke / ainvoke like any other.

Picks the model

ChatOpenAI(model=OPENAI_MODEL or "gpt-4o", api_key=OPENAI_API_KEY, temperature=0). No key, no agent — ValueError.

Discovers sub-agents

fetch_remote_agents(runtime) pulls GET /registry/v1/agents/{id}/subscriptions and wraps each subscription in a PollingA2aTool — a LangChain BaseTool your model calls with a natural-language request.The tool owns the whole A2A exchange: it resolves the agent card, sends message/send, and if the task comes back non-terminal it polls tasks/get through the relay (every 5s, up to 600s) until it’s terminal, then returns the final text. Your model never sees a {"state": "working"} payload it can’t act on.Tool names are sanitised to ^[a-zA-Z0-9_-]+$ — anything else is rejected by OpenAI with 400.

Discovers MCP servers

fetch_mcp_tools(runtime) lists your grants, then hands langchain-mcp-adapters’ MultiServerMCPClient one connection per server, pointed at the relay’s proxy (/relay/v1/mcp-servers/{id}/mcp).Auth is an httpx.Auth implementation rather than static headers, so every request gets a freshly-minted mcp:call token and the current correlation ID — without the LangChain client needing to know anything about Swarmd.Connection keys are mcp_tool_namespace(server), which is also langchain-mcp-adapters’ tool-name prefix, so a server called "GitLab - swarmd.ai" can’t produce an illegal tool name.

The sync/async bridge

create_llm_agent() is synchronous; the MCP client is async-only. The helper bridges with asyncio.run — or, if it’s already inside a running loop, by running the coroutine on a side thread with a copied context, so the correlation ID propagates across the thread boundary. ContextVars aren’t inherited by threads otherwise.

Compiles the graph

local_tools + remote_tools + mcp_tools go to langchain.agents.create_agent() with your instruction as the system prompt and your name as the graph name.Discovery failures are logged and swallowed — an agent that can’t reach the registry at boot still compiles with its local tools.

Stashes the build recipe on the graph

swarmd_name, swarmd_description, swarmd_instruction and swarmd_local_tools are attached as attributes on the returned graph.That’s how serve() can rebuild it later without you threading the arguments through again — and how your local tools survive a rebuild. If you build the graph yourself and want auto-refresh, set these attributes too.

Dissected: `serve()`

serve(
    agent: CompiledStateGraph,
    runtime: SwarmDRuntime,
    *,
    skills: Optional[List[AgentSkill]] = None,
    on_refresh: Optional[Callable[[], None]] = None,
) -> None

Wraps the graph in a LangChainA2aExecutor

The adapter that turns an inbound message/send into a LangGraph ainvoke, publishing the task as submitted → working → completed with the final text as a single message. tasks/cancel routes to the executor’s cancel; tasks/get is served by the A2A request handler out of the task store, without touching the graph.

Builds the agent card

From your skills=[AgentSkill(...)] list, or — if you didn’t pass one — a single catch-all skill synthesised from the agent’s description.

Pass real skills. The card is how other agents and the marketplace decide whether to call you; one generic catch-all skill makes your agent look interchangeable with every other agent that skipped this argument.

from a2a.types import AgentSkill

serve(agent, runtime, skills=[
    AgentSkill(
        id="get_time",
        name="Get local time",
        description="Returns the current local time for a named city",
        examples=["What time is it in Tokyo?"],
        tags=["time", "timezone"],
    ),
])

Persists tasks

SQLite at /tmp/{name}_tasks.db.

/tmp doesn’t survive a pod restart. For durable task state, copy serve() and point DatabaseTaskStore at a real database — it takes any SQLAlchemy async URL.

Adds correlation propagation

CorrelationIdMiddleware captures inbound X-Correlation-Id; every outbound client forwards it. One user request, one trace id, across every hop.

Mounts /admin

POST /admin/configure, POST /admin/refresh, POST /admin/webhook, GET /admin/status — see the webhook-secret gap above for what currently works.

Rebuilds on refresh

The default refresh callback recompiles the graph from the recipe stashed by create_llm_agent() — re-discovering sub-agents and MCP tools — and swaps it into the executor. Your local tools survive.If the rebuild throws, the previous graph is kept and the failure is logged. A registry blip can’t take your agent down.Pass on_refresh= to replace this entirely.

Starts uvicorn

On HOST:PORT (default 0.0.0.0:8080) at LOG_LEVEL (default info).

Going off the rails

Everything is exported:

from swarmd_langchain import (
    create_runtime,
    fetch_remote_agents,     # -> List[PollingA2aTool]
    fetch_mcp_tools,         # -> List[BaseTool]
    PollingA2aTool,
    LangChainA2aExecutor,
    create_a2a_client_factory,
)
from swarmd_sdk import SwarmDRuntime, create_admin_app

Building the runtime yourself

import os
from swarmd_sdk import SwarmDRuntime

runtime = SwarmDRuntime()
runtime.configure(
    agent_id=os.environ["SWARMD_AGENT_ID"],
    client_secret=os.environ["SWARMD_CLIENT_SECRET"],
    base_url=os.environ["SWARMD_BASE_URL"],
    token_url=os.environ["SWARMD_TOKEN_URL"],
    webhook_secret=os.environ["SWARMD_WEBHOOK_SECRET"],
)

Pass that runtime to create_llm_agent() and serve() as normal. Useful when the values come from somewhere other than environment variables — a secret manager, or a config object you already have.

A different model, or the Swarmd LLM gateway

Skip create_llm_agent() and compile the graph yourself:

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

gateway_id = os.environ["SWARMD_LLM_GATEWAY_ID"]
llm = ChatOpenAI(
    model=os.getenv("OPENAI_MODEL", "gpt-4o"),
    base_url=f"{runtime.config.base_url}/llm/v1/{gateway_id}",
    api_key=runtime.token_manager.get_access_token(),
    temperature=0.7,
)

graph = create_agent(
    model=llm,
    tools=[my_tool] + fetch_remote_agents(runtime) + fetch_mcp_tools(runtime),
    system_prompt="...",
    name="my_agent",
)

# So serve()'s auto-refresh can rebuild it later:
graph.swarmd_name = "my_agent"
graph.swarmd_description = "..."
graph.swarmd_instruction = "..."
graph.swarmd_local_tools = [my_tool]

api_key= is read once at construction, so this pins a single bearer token for the model’s lifetime and it will expire. For anything longer-lived than a demo, pass a custom http_client with an auth hook that calls runtime.token_manager.get_access_token() per request — the pattern swarmd-google-adk uses in its _make_llm_httpx_client.

Custom polling interval or wait cap

from swarmd_langchain import PollingA2aTool

tool = PollingA2aTool(
    name="slow_agent",
    description="Takes its time",
    agent_card_url=subscription.agent_card_url,
    poll_interval=15.0,   # default 5.0
    max_wait=3600.0,      # default 600.0
)

Your own server, keeping the admin surface

from swarmd_langchain import LangChainA2aExecutor
from swarmd_sdk import create_admin_app

executor = LangChainA2aExecutor(graph)
# ... bolt executor onto your own Starlette / FastAPI app ...
app.mount("/admin", create_admin_app(runtime, on_refresh=my_refresh))

executor.set_agent(new_graph) swaps the graph in place — that’s what the default refresh callback uses.

Migrating an existing LangChain agent

Diff-shaped guide: what to delete from your current main.py.

Reference

Every helper, argument, and environment variable.

Google ADK

The same three helpers, for ADK.

Troubleshooting

Empty catalogues, 403s on MCP, webhooks that never fire.

Python

Build your agent

Migrating an existing agent

Python reference

TypeScript

LangChain

LangChain on Swarmd

Step 1 — Install

Step 2 — Add your LLM to `.env`

Step 3 — Write the agent

Step 4 — Run it

Step 5 — Talk to it

Dissected: `create_runtime()`

Dissected: `create_llm_agent()`

Dissected: `serve()`

Going off the rails

Next

Migrating an existing LangChain agent

Reference

Google ADK

Troubleshooting

​LangChain on Swarmd

​Step 1 — Install

​Step 2 — Add your LLM to .env

​Step 3 — Write the agent

​Step 4 — Run it

​Step 5 — Talk to it

​Dissected: create_runtime()

​Dissected: create_llm_agent()

​Dissected: serve()

​Going off the rails

​Next

Migrating an existing LangChain agent

Reference

Google ADK

Troubleshooting

LangChain on Swarmd

Step 1 — Install

Step 2 — Add your LLM to `.env`

Step 3 — Write the agent

Step 4 — Run it

Step 5 — Talk to it

Dissected: `create_runtime()`

Dissected: `create_llm_agent()`

Dissected: `serve()`

Going off the rails

Next