Client transports

Every Client talks to its server over a transport: the thing that actually carries the messages.

You never configure one separately. Client takes a single positional argument and works the transport out from its type.

The server side of each (what mcp.run() does and what you deploy) is Running your server.

Streamable HTTP

Pass a URL string and you get Streamable HTTP, the transport you deploy behind and the one to reach for first:

from mcp import Client


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        result = await client.list_tools()
        print([tool.name for tool in result.tools])

That is the whole production client. Client wraps the URL in streamable_http_client(...) for you, on top of an httpx2.AsyncClient configured the way MCP needs: a 30-second timeout for connect/write/pool, and a 300-second read timeout because the server may hold a response stream open.

RuntimeError: Client must be used within an async context manager

Bring your own httpx2.AsyncClient

The moment you need an Authorization header, a cookie, a proxy, mTLS, or a different timeout, build the httpx2.AsyncClient yourself and hand it to streamable_http_client:

import httpx2

from mcp import Client
from mcp.client.streamable_http import streamable_http_client


async def main() -> None:
    async with httpx2.AsyncClient(
        headers={"Authorization": "Bearer ..."},
        timeout=httpx2.Timeout(30.0, read=300.0),
    ) as http_client:
        transport = streamable_http_client("http://localhost:8000/mcp", http_client=http_client)
        async with Client(transport) as client:
            result = await client.list_tools()
            print([tool.name for tool in result.tools])

Two things to notice:

• You own the httpx2.AsyncClient, so you enter and exit it. The SDK never closes a client it didn't create.
• streamable_http_client(url, http_client=...) returns a transport, and Client(transport) accepts it like anything else.

One TLS note: httpx2 verifies certificates against the operating system trust store (via truststore), not a bundled CA list. In an environment with no usable system CA store (some minimal containers), set the standard SSL_CERT_FILE/SSL_CERT_DIR environment variables or pass an explicit verify=ssl_context to your httpx2.AsyncClient (background in httpx and httpx-sse replaced by httpx2).

TypeError: streamable_http_client() got an unexpected keyword argument 'headers'

Redirects

The transport connects to the URL you gave it, and only that origin.

• A 307/308 redirect that stays on the same scheme, host and port is followed, and so is http:// → https:// on the same host. That covers the usual /mcp → /mcp/ trailing-slash redirect.

• A redirect anywhere else is not followed. The call fails with:

  MCPError: Redirect to https://other.example.com/mcp not followed; use that URL as the endpoint if it is the intended server
  

  If that URL is the server you meant, put it in your config. If it isn't, the server or a proxy in front of it is misconfigured.

This holds for any httpx2.AsyncClient you pass in: its follow_redirects setting is not consulted for MCP requests, in either direction. The SDK's OAuth providers apply the same rule to their own requests.

stdio

A stdio server is a subprocess. The client launches it, writes JSON-RPC to its stdin and reads JSON-RPC from its stdout. It is how a desktop host runs a server on your machine: a host is this code plus a UI, and Connect to a real host is the same relationship seen from the host's side, as a config file.

Describe the process with StdioServerParameters and hand it to Client:

from mcp import Client, StdioServerParameters

server = StdioServerParameters(
    command="uv",
    args=["run", "server.py"],
    env={"BOOKSHOP_API_KEY": "secret"},
)


async def main() -> None:
    async with Client(server) as client:
        result = await client.list_tools()
        print([tool.name for tool in result.tools])

Entering the block spawns the process. Leaving it shuts the subprocess down: close stdin, wait, kill if it lingers. You never clean it up yourself.

The child's stderr goes to yours. To send it somewhere else, build the transport yourself with stdio_client (from mcp) and pass that instead: Client(stdio_client(server, errlog=log_file)).

In memory

In a test there is nothing to deploy and nothing to launch. Pass the server object itself:

from mcp import Client
from mcp.server import MCPServer

mcp = MCPServer("Bookshop")


@mcp.tool()
def search_books(query: str) -> str:
    """Search the catalog by title or author."""
    return f"Found 3 books matching {query!r}."


async def main() -> None:
    async with Client(mcp) as client:
        result = await client.call_tool("search_books", {"query": "dune"})
        print(result.structured_content)

No subprocess, no port, no bytes on a wire. The client and the server are two objects in the same process, and the call still goes through the real protocol layer: search_books is listed, validated and invoked exactly as it would be over HTTP. Testing builds the whole pattern around it.

The same form doubles as an embedding API: an application that constructs the server itself can call its tools without a network hop.

Closing the client cancels active in-process requests and waits for their handler cleanup before leaving application lifespan. A caller interrupted by connection closure receives MCPError with code CONNECTION_CLOSED. Handlers and callbacks must cooperate with cancellation; shielded cleanup keeps the application's resources alive until it finishes.

SSE

sse_client(url), from mcp.client.sse, is the HTTP transport that Streamable HTTP superseded. Wrap it the same way, Client(sse_client("http://localhost:8000/sse")), to talk to a server that still speaks it, and don't build anything new on it.

The Transport protocol

To Client, all of the above are the same thing.

A transport is any async context manager that yields a (read, write) pair of message streams: formally, the Transport protocol in mcp.client. Client resolves its argument by type: a str becomes streamable_http_client(url), a StdioServerParameters becomes stdio_client(params), a server object connects in-process, and anything else is entered as a transport directly. That last rule is why stdio_client(...), streamable_http_client(...) and sse_client(...) all drop into the same slot, and why you can write your own.

Implement a message transport

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from typing import Any

import anyio

from mcp import Client
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.runtime import ServerRuntime
from mcp.shared.memory import create_client_server_memory_streams
from mcp.shared.transport import MessageMetadata, TransportContext, TransportStreams


@dataclass(kw_only=True, frozen=True)
class PeerContext(TransportContext):
    peer: str


server = MCPServer("Custom transport")


@server.tool()
async def identify(ctx: Context) -> str:
    transport = ctx.transport
    assert isinstance(transport, PeerContext)
    return transport.peer


@asynccontextmanager
async def memory_client(runtime: ServerRuntime[Any], peer: str) -> AsyncIterator[TransportStreams]:
    async with create_client_server_memory_streams() as (client_streams, server_streams):

        @asynccontextmanager
        async def server_transport() -> AsyncIterator[TransportStreams]:
            async with server_streams[0], server_streams[1]:
                yield server_streams

        def build_context(metadata: MessageMetadata) -> PeerContext:
            return PeerContext(kind="memory", can_send_request=True, peer=peer)

        await runtime.connect(server_transport(), transport_builder=build_context)
        yield client_streams


async def main() -> None:
    async with server.serve(max_connections=10) as runtime:
        async with Client(memory_client(runtime, "alice")) as alice:
            async with Client(memory_client(runtime, "bob")) as bob:
                alice_result = await alice.call_tool("identify")
                bob_result = await bob.call_tool("identify")
                assert alice_result.structured_content == {"result": "alice"}
                assert bob_result.structured_content == {"result": "bob"}


if __name__ == "__main__":
    anyio.run(main)

This example implements an in-memory adapter with two independent clients. A network adapter uses the same TransportStreams contract and replaces the memory channels with message readers and writers. You import the contract and its supporting types from mcp.shared.transport; the existing mcp.client.Transport import still works.

Each stream pair represents one logical peer, not an entire broker. The adapter owns framing, routing, and its network resources. The SDK owns negotiation, request correlation, and MCP validation.

Entering a transport opens its channel. Exiting stops its background tasks and closes resources it owns. The SDK also closes streams during connection shutdown, so their aclose() methods must be safe to call more than once. A network client supplied by the application remains owned by the application.

An inbound item is a decoded SessionMessage or an exception describing a recoverable message error. An exception item alone does not disconnect the peer. End the read stream on connection loss so pending calls fail instead of waiting indefinitely. Make writes cancellable and apply backpressure rather than buffering without a bound.

The server side of this example uses server.serve(). Its lifecycle and connection limits are covered under Custom transports. The repository's examples/transports/README.md contains live MQTT 5 and AMQP 0.9.1 examples, their binding rules, and the validation still needed before production use.

Integrate a native dispatcher

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Any

import anyio

from mcp import Client
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.runtime import ServerRuntime
from mcp.shared.direct_dispatcher import create_direct_dispatcher_pair
from mcp.shared.dispatcher import Dispatcher
from mcp.shared.transport import DispatcherTransport, TransportContext

server = MCPServer("Dispatcher transport")


@server.tool()
async def greet(name: str, ctx: Context) -> str:
    assert ctx.transport is not None
    assert not ctx.transport.can_send_request
    return f"Hello, {name}!"


def direct_client(runtime: ServerRuntime[Any]) -> DispatcherTransport:
    @asynccontextmanager
    async def connection() -> AsyncIterator[Dispatcher[TransportContext]]:
        client_dispatcher, server_dispatcher = create_direct_dispatcher_pair()

        @asynccontextmanager
        async def server_connection() -> AsyncIterator[Dispatcher[TransportContext]]:
            try:
                yield server_dispatcher
            finally:
                server_dispatcher.close()

        try:
            await runtime.connect(DispatcherTransport(server_connection()))
            yield client_dispatcher
        finally:
            client_dispatcher.close()
            server_dispatcher.close()

    return DispatcherTransport(connection())


async def main() -> None:
    async with server.serve() as runtime:
        async with Client(direct_client(runtime)) as client:
            result = await client.call_tool("greet", {"name": "Alice"})
            assert result.structured_content == {"result": "Hello, Alice!"}


if __name__ == "__main__":
    anyio.run(main)

DispatcherTransport explicitly wraps an async context manager yielding a Dispatcher. Client enters that context, starts the dispatcher, and uses its ordinary MCP negotiation, callbacks, caching, and validation. It stops the dispatcher before exiting the connection context. You configure the client through the same constructor; there is no separate native client-session API.

The example uses the SDK's DirectDispatcher. The native gRPC reference adapter is developed in a separate follow-up to this SDK API change. Native network bindings implement this dispatcher boundary instead of creating SessionMessage streams. The connection context acquires the transport resources; it must yield an unstarted dispatcher because the SDK owns run().

On the server, runtime.connect(DispatcherTransport(...)) serves the modern per-request-envelope protocol. It rejects the legacy initialize handshake. Use mode="auto" or a supported modern version on the client. Message transports still support both eras. Native dispatchers supply their own contexts, so this server path rejects session_id= and transport_builder=.

Recap

• Client("http://.../mcp") (a URL) connects over Streamable HTTP, the production transport.
• Headers, auth, proxies and timeouts belong on an httpx2.AsyncClient you pass to streamable_http_client(url, http_client=...). There is no headers= keyword.
• Redirects are followed only within the URL's own origin (a trailing-slash 307/308), plus http→https on the same host. Anything else fails with Redirect to … not followed; configure the final URL.
• stdio is Client(StdioServerParameters(...)). Wrap it in stdio_client(...) yourself only to redirect the child's stderr.
• The subprocess gets an allow-listed environment, not yours; env= adds to it.
• Client(mcp) (the server object) connects in memory. Use it in tests, or to embed a server in the application that built it.
• A transport is anything you can async with x as (read, write). Client hands anything that isn't a server object, a URL or StdioServerParameters straight to that protocol.
• Constructing a Client picks the transport. async with opens it.

Once the transport is open the two sides have to agree on a protocol version. You normally never think about it; when you do, Protocol versions is the page.