> ## Documentation Index
> Fetch the complete documentation index at: https://docs.swarmd.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Core SDK

> Use swarmd-sdk directly — authenticate, discover subscriptions, call MCP servers, and serve an agent, in any Python framework.

# The core SDK

`swarmd-sdk` is the foundation both framework wrappers are built on. It has no
opinion about how you write your agent — it only handles the parts that touch
Swarmd.

Use it directly when:

* You're using a framework we don't wrap yet (CrewAI, Autogen, raw OpenAI, your
  own loop).
* You want a plain API client — a script, a CI job, a backend service that
  lists agents or calls an MCP tool.
* You're using ADK or LangChain but need to deviate from what the wrapper does.

<Note>
  Using Google ADK or LangChain? The wrapper is strictly less work — start at
  [Google ADK](/sdks/python/google-adk) or [LangChain](/sdks/python/langchain)
  and come back here when you need to drop a level.
</Note>

This page builds up from a one-line client to a fully-served agent. Each part
stands on its own; take what you need.

<Info>
  Assumes you've finished [Setup](/sdks/python/setup) — a registered agent and a
  `.env` with your three secrets.
</Info>

```bash theme={null}
pip install swarmd-sdk
```

***

## Part 1 — Authenticate and discover

The smallest useful program: who am I subscribed to?

```python discover.py theme={null}
import os
from uuid import UUID

from dotenv import load_dotenv

from swarmd_sdk import SwarmDClient

load_dotenv()

me = UUID(os.environ["SWARMD_AGENT_ID"])

with SwarmDClient.from_env() as client:
    for agent in client.get_agent_subscriptions(me):
        print(f"{agent.name}")
        print(f"  {agent.description}")
        print(f"  card: {agent.agent_card_url}")
```

### Dissecting it

<AccordionGroup>
  <Accordion title="SwarmDClient.from_env() — where config comes from">
    Reads `SWARMD_AGENT_ID` and `SWARMD_CLIENT_SECRET` (both required, raises
    `ValueError` if missing), plus four optional overrides:

    | Variable             | Default                                                                  |
    | -------------------- | ------------------------------------------------------------------------ |
    | `SWARMD_BASE_URL`    | `https://api.dev.swarmd.ai`                                              |
    | `SWARMD_TOKEN_URL`   | `https://auth.dev.swarmd.ai/realms/swarmd/protocol/openid-connect/token` |
    | `SWARMD_TIMEOUT`     | `30`                                                                     |
    | `SWARMD_MAX_RETRIES` | `3`                                                                      |

    Running two agents in one process? Pass a prefix:
    `SwarmDClient.from_env("WEATHER_AGENT")` reads `WEATHER_AGENT_AGENT_ID` and so
    on.

    Prefer explicit config? `SwarmDClient(agent_id=..., client_secret=...)` or
    `SwarmDClient.from_config(SwarmDConfig(...))`.
  </Accordion>

  <Accordion title="The `with` block — why it matters">
    `SwarmDClient` holds a pooled `httpx.Client`. The context manager closes it on
    exit. In a long-lived service you'd build the client once at startup and close
    it at shutdown instead — just don't create one per request; you'd throw away
    the connection pool and the cached token every time.
  </Accordion>

  <Accordion title="What happened on the wire">
    You wrote one line. The SDK did five things:

    1. `POST` to the token endpoint with `grant_type=client_credentials`,
       `scope=swarmd:api`, `resource={base_url}` — the RFC 8707 parameter that
       binds the token's audience to the gateway.
    2. Cached the token in memory, marked expired 60s before its real expiry.
    3. `GET /registry/v1/agents/{id}/subscriptions?page=0&size=100` with
       `Authorization: Bearer …`.
    4. Followed pagination until `totalPages` was exhausted (capped at 100 pages).
    5. Validated each row into an `AgentResponse` pydantic model.

    On a `401` it would have cleared the token and retried once. On a `5xx` or a
    network error, retried with exponential backoff up to `max_retries`. On any
    other `4xx`, failed immediately — retrying a `403` doesn't help anyone.
  </Accordion>
</AccordionGroup>

### Handling failure

```python theme={null}
from swarmd_sdk import APIError, AuthenticationError, SwarmDClient

try:
    with SwarmDClient.from_env() as client:
        agents = client.get_agent_subscriptions(me)
except AuthenticationError as e:
    # Credentials rejected by the token endpoint. Not retryable.
    print(f"auth failed: {e}")
except APIError as e:
    # The API answered, unhappily. status_code and response_body tell you why.
    print(f"api error {e.status_code}: {e.response_body}")
```

The hierarchy: `SwarmDSDKError` → `AuthenticationError` → `TokenRefreshError`,
and `SwarmDSDKError` → `APIError`. Catch `SwarmDSDKError` to catch everything
the SDK raises.

***

## Part 2 — Call an MCP server

MCP servers are granted to your agent the same way sub-agents are. `McpClient`
discovers them and proxies JSON-RPC through the relay.

```python mcp_demo.py theme={null}
from dotenv import load_dotenv

from swarmd_sdk import McpClient

load_dotenv()

with McpClient.from_env() as mcp:
    servers = mcp.list_available()
    if not servers:
        raise SystemExit("no MCP servers granted to this agent yet")

    server = servers[0]
    print(f"{server.name} ({server.mcp_server_id})")

    for tool in mcp.list_tools(server.mcp_server_id):
        print(f"  - {tool['name']}: {tool.get('description', '')}")

    result = mcp.call_tool(
        server.mcp_server_id,
        "search",
        {"query": "weather in Tokyo"},
    )
    print(result)
```

### Dissecting it

<AccordionGroup>
  <Accordion title="Why a separate client from SwarmDClient?">
    Because it needs a different token. `McpClient` mints `mcp:call`-scoped tokens
    bound to `{base_url}/relay`; the relay's MCP path rejects the `swarmd:api`
    token `SwarmDClient` carries. Same credentials, different audience — see
    [two tokens, not one](/sdks/python/concepts#2-two-tokens-not-one).

    Everything else mirrors `SwarmDClient`: same constructor arguments, same
    `from_env` / `from_config`, same retry semantics, same context manager.
  </Accordion>

  <Accordion title="list_available() vs. the relay">
    Discovery hits the **registry**
    (`GET /registry/v1/agents/{id}/mcp-subscriptions`), because the registry is the
    source of truth for grants. The relay only enforces grants on the call path.

    An agent principal is constrained to its own id server-side, so you can't
    enumerate another agent's grants with your token.
  </Accordion>

  <Accordion title="call() — dropping to raw JSON-RPC">
    `list_tools()` and `call_tool()` are conveniences over `call()`. For anything
    else the MCP spec defines — `resources/list`, `prompts/get`, `initialize` —
    use `call()` directly:

    ```python theme={null}
    result = mcp.call(
        server.mcp_server_id,
        "resources/list",
        {},
        session_id="optional-Mcp-Session-Id",
        protocol_version="2025-06-18",
    )
    ```

    It builds the JSON-RPC 2.0 envelope, generates a `uuid4` request id if you
    don't supply one, and returns the `result` field. A JSON-RPC `error` object in
    the response is raised as `McpJsonRpcError` with `.code`, `.message` and
    `.data`.
  </Accordion>

  <Accordion title="McpNotSubscribedError — the useful 403">
    A `403` from the relay's MCP proxy path specifically means *you don't have a
    grant for this server*, and is raised as `McpNotSubscribedError` (a subclass of
    `APIError`) so you can distinguish it from generic failure:

    ```python theme={null}
    from swarmd_sdk import McpNotSubscribedError

    try:
        mcp.call_tool(server_id, "search", {"query": "x"})
    except McpNotSubscribedError:
        print("ask an operator to grant this server")
    ```

    `403`s from other paths stay as plain `APIError` — the SDK deliberately doesn't
    paper over an RBAC gap on the registry side by calling it a subscription
    problem.
  </Accordion>
</AccordionGroup>

***

## Part 3 — Keep discovery fresh

Grants change while your agent runs. `AgentDirectory` polls in the background
and hands you an atomically-swapped snapshot.

```python theme={null}
from swarmd_sdk import AgentDirectory, SwarmDClient


def on_change(added, removed):
    for agent_id in added:
        print(f"+ {agent_id}")
    for agent_id in removed:
        print(f"- {agent_id}")


client = SwarmDClient.from_env()

with AgentDirectory(client, me, refresh_interval=30, on_change=on_change) as directory:
    weather = directory.get("weather_agent")     # by name, or None
    everything = directory.agents                # immutable tuple
```

`start()` refreshes once synchronously before spawning the thread, so
`directory.agents` is populated the moment the block opens. Refresh failures
are logged and swallowed — a registry blip leaves the previous snapshot in
place rather than emptying your catalogue.

***

## Part 4 — Serve an agent

Here's the part the wrappers otherwise hide: a complete, self-serving agent
using nothing but `swarmd-sdk` and Starlette.

```python main.py theme={null}
import os
from uuid import UUID

import uvicorn
from dotenv import load_dotenv
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route

from swarmd_sdk import CorrelationIdMiddleware, SwarmDRuntime, create_admin_app

load_dotenv()

# 1. Build and configure the runtime.
runtime = SwarmDRuntime()
runtime.configure(
    agent_id=os.environ["SWARMD_AGENT_ID"],
    client_secret=os.environ["SWARMD_CLIENT_SECRET"],
    base_url=os.getenv("SWARMD_BASE_URL", "https://api.dev.swarmd.ai"),
    token_url=os.environ["SWARMD_TOKEN_URL"],
    webhook_secret=os.getenv("SWARMD_WEBHOOK_SECRET"),
)

# 2. Your catalogue, discovered at boot and rebuilt on refresh.
CATALOGUE = {"agents": (), "mcp_servers": ()}


def refresh() -> None:
    """Called at boot, on POST /admin/refresh, and on every signed webhook."""
    me = UUID(runtime.config.agent_id)
    CATALOGUE["agents"] = tuple(runtime.client.get_agent_subscriptions(me))
    CATALOGUE["mcp_servers"] = tuple(runtime.mcp.list_available())
    print(
        f"[refresh] {len(CATALOGUE['agents'])} agent(s), "
        f"{len(CATALOGUE['mcp_servers'])} MCP server(s)"
    )


refresh()


# 3. Your agent's own endpoints. Replace with a real A2A server in production.
async def agent_card(request):
    return JSONResponse({
        "name": "my_agent",
        "description": "An agent built on the core Swarmd SDK",
        "url": f"http://localhost:{os.getenv('PORT', '8080')}/",
        "version": "0.1.0",
        "capabilities": {"streaming": False},
        "defaultInputModes": ["text/plain"],
        "defaultOutputModes": ["text/plain"],
        "skills": [],
    })


async def health(request):
    return JSONResponse({"status": "ok"})


app = Starlette(routes=[
    Route("/.well-known/agent-card.json", agent_card),
    Route("/health", health),
])

# 4. Correlation propagation, then the admin surface.
app.add_middleware(CorrelationIdMiddleware)
app.mount("/admin", create_admin_app(runtime, on_refresh=refresh))

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", "8080")))
```

Run it and check the admin surface:

```bash theme={null}
python main.py
```

```bash theme={null}
curl http://localhost:8080/admin/status
# {"configured":true,"agent_id":"0c6eab9a-..."}

curl -X POST http://localhost:8080/admin/refresh
# {"status":"refreshed"}
```

### Dissecting it

<AccordionGroup>
  <Accordion title="Why configure() rather than passing config to a constructor">
    So credentials can arrive after boot. If you deploy an agent before its
    service account exists, leave the runtime unconfigured — `runtime.configured`
    is `False`, the agent still serves its card and health check, and an operator
    later does:

    ```bash theme={null}
    curl -X POST http://localhost:8080/admin/configure \
      -H 'Content-Type: application/json' \
      -d '{"agent_id":"...","client_secret":"...","base_url":"https://api.dev.swarmd.ai","token_url":"https://auth.dev.swarmd.ai/realms/swarmd/protocol/openid-connect/token","webhook_secret":"..."}'
    ```

    `configure()` builds fresh clients, swaps them in under a lock, and closes the
    old ones. `base_url` and `token_url` are required on the *first* configure and
    optional on subsequent ones, so a credential rotation only needs the two
    secrets.
  </Accordion>

  <Accordion title="create_admin_app(runtime, on_refresh=refresh)">
    Returns a FastAPI app with the four `/admin` endpoints. `on_refresh` is the
    only thing you supply — it's called by both `POST /admin/refresh` and by
    `POST /admin/webhook` after the HMAC verifies.

    Your callback runs while holding the runtime's lock, so it can't race a
    concurrent `configure()`. It should be idempotent: the same webhook can arrive
    twice, and the SDK acks duplicates rather than trying to dedupe them.

    Omit `on_refresh` and `/admin/refresh` returns `501`, while a verified webhook
    returns `{"status": "ignored"}` — acked, so the platform's dispatcher doesn't
    dead-letter a legitimate delivery.
  </Accordion>

  <Accordion title="Mount order: middleware before mount">
    `CorrelationIdMiddleware` is added to the outer Starlette app, so it wraps
    everything including the mounted admin app. It reads `X-Correlation-Id` off the
    inbound request into a `ContextVar`, and resets it in a `finally` so requests
    don't leak ids into each other.

    Every outbound client the SDK builds reads that same var and forwards the
    header — which is what makes one user request show up as one trace across
    five agents in the audit log.
  </Accordion>

  <Accordion title="Push your card to the registry on boot">
    The registry caches your agent card from registration time. If your
    description or skills have drifted, ask it to re-read:

    ```python theme={null}
    result = runtime.client.refresh_agent_card(UUID(runtime.config.agent_id))
    # {"status": "unchanged", "changedFields": []}
    # {"status": "updated",   "changedFields": ["description", "skills"]}
    ```

    Call it with your **own** agent id — the registry compares it against the JWT
    subject and rejects a cross-agent refresh with `403`. Treat failure as
    non-fatal: a registry blip at boot must not stop your agent serving.

    `swarmd-google-adk` does this for you inside `create_llm_agent()`.
    `swarmd-langchain` does not — call it yourself if you want it.
  </Accordion>
</AccordionGroup>

***

## Part 5 — Call a sub-agent over A2A

The wrappers turn each subscription into a framework-native tool. Without a
wrapper, you drive the A2A protocol yourself. Here's the minimum, so you can
see exactly what's being automated for you.

```python call_subagent.py theme={null}
import asyncio
import os
from uuid import UUID, uuid4

import httpx
from dotenv import load_dotenv

from swarmd_sdk import SwarmDRuntime

load_dotenv()

runtime = SwarmDRuntime()
runtime.configure(
    agent_id=os.environ["SWARMD_AGENT_ID"],
    client_secret=os.environ["SWARMD_CLIENT_SECRET"],
    base_url=os.getenv("SWARMD_BASE_URL", "https://api.dev.swarmd.ai"),
    token_url=os.environ["SWARMD_TOKEN_URL"],
)

TERMINAL = {"completed", "failed", "canceled", "rejected", "unknown"}


async def ask(agent_name: str, prompt: str) -> dict:
    me = UUID(runtime.config.agent_id)
    target = next(
        a for a in runtime.client.get_agent_subscriptions(me) if a.name == agent_name
    )

    def auth() -> dict:
        return {"Authorization": f"Bearer {runtime.token_manager.get_access_token()}"}

    async with httpx.AsyncClient(timeout=300.0) as http:
        # The subscription gives you a *relay* card URL. Fetch it and read the
        # JSON-RPC endpoint out of the card rather than guessing a path.
        card = (await http.get(str(target.agent_card_url), headers=auth())).json()
        rpc_url = card["url"]

        send = await http.post(rpc_url, headers=auth(), json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "message/send",
            "params": {"message": {
                "role": "user",
                "parts": [{"kind": "text", "text": prompt}],
                "messageId": str(uuid4()),
            }},
        })
        task = send.json()["result"]

        # This is the bit the wrappers exist for: A2A tasks come back
        # non-terminal, and an LLM handed {"state": "working"} is useless.
        while task.get("status", {}).get("state") not in TERMINAL:
            await asyncio.sleep(5)
            poll = await http.post(rpc_url, headers=auth(), json={
                "jsonrpc": "2.0",
                "id": 2,
                "method": "tasks/get",
                "params": {"id": task["id"]},
            })
            task = poll.json()["result"]

        return task


print(asyncio.run(ask("weather_agent", "What's the weather in Tokyo?")))
```

That's roughly 40 lines to call **one** sub-agent, and it's still missing
correlation-ID propagation, `401` retry, a wait cap, and conversion into
whatever tool shape your LLM expects. Multiply by every subscription and
that's the argument for the wrappers.

<Tip>
  If you're on a framework we don't wrap, you don't have to write this from
  scratch either — use the [`a2a-sdk`](https://pypi.org/project/a2a-sdk/) client
  and inject `runtime.token_manager` in an interceptor. That's exactly what
  `swarmd-google-adk` and `swarmd-langchain` do; their `helpers.py` is a fine
  template.
</Tip>

***

## What you're doing by hand

Everything on this page, side by side with what a wrapper would have done:

|                                | Core SDK         | Wrapper           |
| ------------------------------ | ---------------- | ----------------- |
| Tokens, refresh, `401` retry   | ✓ automatic      | ✓ automatic       |
| Discovery calls                | ✓ automatic      | ✓ automatic       |
| MCP JSON-RPC                   | ✓ automatic      | ✓ automatic       |
| Webhook HMAC + `/admin`        | ✓ automatic      | ✓ automatic       |
| Correlation middleware         | you add one line | ✓ automatic       |
| Refresh callback               | you write it     | ✓ automatic       |
| Subscriptions → LLM tools      | you write it     | ✓ automatic       |
| A2A send + poll                | you write it     | ✓ automatic       |
| A2A server + agent card        | you write it     | ✓ automatic       |
| LLM routed through the gateway | you write it     | ✓ automatic (ADK) |

***

## Next

<CardGroup cols={2}>
  <Card title="Reference" icon="book" href="/sdks/python/reference">
    Every class, method, and environment variable in `swarmd-sdk`.
  </Card>

  <Card title="Troubleshooting" icon="bug" href="/sdks/python/troubleshooting">
    The failure modes you'll actually hit, and what they mean.
  </Card>
</CardGroup>