Model Context Protocol (MCP)

Padwan AI ships two MCP client transports — McpStreamable (HTTP) and McpStdio (subprocess) — both implementing the MCP 2025-11-25 spec. They expose the same tools interface, but their internal concurrency models differ in important ways.

Quick start

from padwan_ai import McpStreamable, McpStdio

# Streamable HTTP — remote MCP server (with optional bearer token)
async with McpStreamable(url="https://mcp.example.com/mcp", token="sk-...") as mcp:
    for tool in mcp.tools:
        print(tool.name, tool.description)
    result = await mcp.tools[0].handler({"query": "test"})

# Stdio — local subprocess
async with McpStdio(command="uvx", args=["my-mcp-server"]) as mcp:
    tool = next(t for t in mcp.tools if t.name == "search")
    result = await tool.handler({"query": "hello"})

Feature comparison

Constructor parameters

McpStreamable

McpStdio

McpStreamable — dual-channel architecture

The streamable HTTP transport runs two concurrent communication channels against the same endpoint:

flowchart LR
    subgraph client["McpStreamable client"]
        Caller["User code<br/>tool.handler"]
        RPC["_rpc<br/>foreground POST"]
        Listen["_listen<br/>background GET task"]
        Tools["tools list<br/>(_tools)"]
    end

    subgraph server["MCP server"]
        Endpoint["/mcp"]
    end

    Caller -->|1 await| RPC
    RPC -->|2 POST JSON-RPC| Endpoint
    Endpoint -->|3 response JSON or SSE| RPC
    RPC -->|4 result| Caller

    Listen -.->|GET SSE persistent| Endpoint
    Endpoint -.->|server-pushed notifications| Listen
    Listen -.->|tools list_changed, refresh| Tools
    Listen -.->|progress, on_progress callback| Caller

The foreground channel (_rpc) handles request/response: you await tool.handler(...), it POSTs a JSON-RPC request, gets back either a single JSON response or an SSE stream containing the response. This is what user code interacts with directly.

The background channel (_listen) is a persistent GET against the same /mcp endpoint, opened during __aenter__ and torn down in __aexit__. It receives unsolicited server pushes — things the server wants to tell you without being asked, like notifications/tools/list_changed (server's tool list changed, we should re-fetch) or out-of-band notifications/progress for long-running operations. It is opened without a read timeout: a quiet stream is normal, and timing it out would count as a listener failure.

This separation means the listener can deliver notifications even while you have no in-flight RPC. Without it, a server that wanted to tell you "my tools changed" would have to wait until you happened to ask for something.

Notifications aren't limited to the background channel: if a POST's response is itself an SSE stream, _rpc also handles tools/list_changed and progress events that arrive inline before the response.

Setting on_progress makes tools/call send _meta.progressToken with the request, so a spec-compliant server knows to emit notifications/progress for that call. Without on_progress, no token is sent and no progress notifications arrive.

McpStreamable — full lifecycle

sequenceDiagram
    participant App as User code
    participant Client as McpStreamable
    participant Listener as _listen task
    participant Server as MCP server

    Note over App,Server: aenter
    App->>Client: async with McpStreamable
    Client->>Server: POST initialize
    Server-->>Client: result plus MCP-Session-Id header
    Client->>Server: POST notifications/initialized
    Client->>Server: POST tools/list
    Server-->>Client: tool list
    Client->>Listener: spawn background task
    Listener->>Server: GET /mcp (Accept text/event-stream)
    Server-->>Listener: SSE stream open

    Note over App,Server: Operation
    App->>Client: await tool.handler(args)
    Client->>Server: POST tools/call
    Server-->>Client: result (JSON or SSE)
    Client-->>App: return result

    Server-->>Listener: notifications/progress
    Listener->>App: on_progress(event)

    Server-->>Listener: notifications/tools/list_changed
    Listener->>Server: POST tools/list (refresh)
    Server-->>Listener: updated tool list

    Note over App,Server: aexit
    App->>Client: exit context
    Client->>Listener: cancel task
    Client->>Server: close session

The key thing to notice: the background _listen task is spawned during __aenter__ and lives for the entire duration of the async with block. It runs in parallel with all your RPC calls and can deliver notifications at any moment.

Session recovery and cleanup

If a request gets HTTP 404 with an active session (the server forgot it), the client re-initializes once and retries the request. On __aexit__, the client sends DELETE to end the session server-side.

Token refresh on 401

Set on_auth to recover from expired bearer tokens. This applies to request/response RPCs only (initialize, tools/list, tools/call, ping) — when one returns HTTP 401, the client invokes the callback (sync or async), stores the returned token, and retries the request exactly once. The notifications/initialized POST, cancel(), and the background listener's GET don't call on_auth; a 401 on the listener GET instead burns one of its reconnect attempts.

async def refresh(transport: McpStreamable) -> str:
    return await oauth.fetch_token(...)


async with McpStreamable(url="...", token=initial_token, on_auth=refresh) as mcp:
    ...

Without on_auth, a 401 raises RuntimeError.

Reconnection with Last-Event-ID

If the SSE stream drops mid-session, _listen automatically reconnects and tells the server to resume from the last event it successfully processed. This uses the standard SSE Last-Event-ID header.

sequenceDiagram
    participant Listener as _listen task
    participant Server as MCP server

    Listener->>Server: GET /mcp (no Last-Event-ID)
    Server-->>Listener: event id="evt-1"
    Note over Listener: _last_event_id = "evt-1"
    Server-->>Listener: event id="evt-2"
    Note over Listener: _last_event_id = "evt-2"
    Server--xListener: connection drops

    Note over Listener: retries < max_retries<br/>sleep(_retry_ms)

    Listener->>Server: GET /mcp<br/>Last-Event-ID: evt-2
    Server-->>Listener: event id="evt-3"<br/>(resumes after evt-2)

The retry budget allows up to 5 consecutive failed reconnects (not configurable) before the listener gives up; a clean stream end counts as one retry too, and a successful reconnect resets the counter back to 0. The inter-attempt delay is controlled by the server via the SSE retry: field — if the server sends retry: 5000, the next reconnect waits 5 seconds. The client falls back to _DEFAULT_RETRY_MS = 3000 ms otherwise.

Once the 5 consecutive failures are exhausted, the listener logs a warning and stops. is_open stays True — the transport is still usable for RPCs — but push notifications (progress events, tools/list_changed) no longer arrive.

If the server returns 405 Method Not Allowed on the GET, the listener exits cleanly — that signals "this server doesn't support push notifications," which is allowed by the spec.

McpStdio — single-pipe architecture

The stdio transport launches the MCP server as a child process and communicates over its stdin/stdout. There's only one bidirectional pipe (NDJSON-framed), so the concurrency model is different from streamable HTTP:

flowchart LR
    subgraph client["McpStdio parent process"]
        Caller["User code<br/>tool.handler"]
        RPC["_rpc<br/>writes to stdin"]
        Reader["_reader task<br/>reads stdout"]
        Pending["_pending<br/>id to Future map"]
        OnProg["on_progress callback"]
    end

    subgraph child["MCP server subprocess"]
        Stdin["stdin"]
        Stdout["stdout"]
    end

    Caller -->|await| RPC
    RPC -->|register Future by id| Pending
    RPC -->|write JSON line| Stdin
    Stdout -->|read line| Reader
    Reader -->|match id, resolve Future| Pending
    Pending -->|future result| Caller
    Reader -->|notification without id| OnProg

Because both directions share a single pipe, there's a single background reader task parsing NDJSON lines as they arrive. Each outgoing request gets a unique JSON-RPC id, and the reader uses that id to look up the right asyncio.Future and resolve it with the response. Notifications (messages without an id) are dispatched directly to the appropriate handler — tools/list_changed triggers a refresh, progress invokes the user's on_progress callback. Server-initiated requests (e.g. a server ping) and messages with an unrecognized id fall into the same branch as notifications and are silently ignored — the client never answers them.

Compared to McpStreamable:

• No session management — the connection is the process lifetime
• No reconnection logic — if the subprocess dies, the connection is dead, all pending futures get an exception
• No HTTP semantics — no headers, no auth, no Last-Event-ID. The protocol is just newline-delimited JSON-RPC over pipes
• One reader, many waiters — the reader is a fan-out point that resolves multiple pending futures from a single byte stream

Cancellation

Both transports support cancel(request_id), which sends a notifications/cancelled message to the server. The server is expected to abort the in-flight operation and stop sending progress events for it. This is best-effort — the spec doesn't require the server to actually stop, just to acknowledge the request.

cancel(request_id) is low-level: request ids are generated internally (uuid4().hex for McpStreamable, an incrementing counter for McpStdio) and never exposed to callers, so there's no supported way to obtain a request_id to pass it.

The user-facing way to cancel is to cancel the asyncio.Task that's awaiting tool.handler(...). The transport catches CancelledError, sends the cancellation notification, and re-raises.

See also

• MCP transports spec
• MCP lifecycle spec
• MDN — Server-sent events: Last-Event-ID