Add extensible client and server transport APIs

docs/advanced/low-level-server.md

@@ -65,6 +65,12 @@ The same text the `@mcp.tool()` version produced. Two honest differences:
 
 In a test you skip uvicorn and the port: `Client(server)` takes a low-level `Server` in-process exactly like it takes an `MCPServer`, and **[Testing](../get-started/testing.md)** is that pattern.
 
+## Custom transports
+
+`Server.serve()` shares one application lifespan across multiple custom transport connections, just like `MCPServer.serve()`. Use the complete adapter example under [Running your server](../run/index.md#custom-transports).
+
+For a single connection, `Server.run(read_stream, write_stream, initialization_options, *, transport_builder=...)` remains available. The optional builder converts inbound message metadata into the `TransportContext` exposed as `ctx.transport`. Without it, stream dispatch uses the context supplied by the framing transport, falling back to generic JSON-RPC metadata. Built-in HTTP transports supply their kind and the current request's headers. Both paths retain the existing protocol-version handling; custom transport capabilities cannot enable features that the negotiated version forbids.
+
 ## Nothing is checked for you
 
 `MCPServer` rejects a bad argument before your function ever runs, validating the call against the schema it generated (**[Tools](../servers/tools.md)**).

docs/client/transports.md

@@ -118,6 +118,8 @@ No subprocess, no port, no bytes on a wire. The client and the server are two ob
 
 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.
@@ -128,6 +130,40 @@ 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
+
+```python title="custom_transport.py"
+--8<-- "docs_src/client_transports/tutorial005.py"
+```
+
+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.
+
+!!! warning "Delivery is not execution"
+    MQTT or AMQP delivery guarantees do not make a tool execute exactly once. A redelivered request can repeat a side effect. Define expiry, duplicate handling, and reconnect behavior in the adapter; do not silently replay unfinished calls.
+
+The server side of this example uses `server.serve()`. Its lifecycle and connection limits are covered under [Custom transports](../run/index.md#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
+
+```python title="dispatcher_transport.py"
+--8<-- "docs_src/client_transports/tutorial006.py"
+```
+
+`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=`.
+
+!!! warning "Native bindings remain experimental"
+    The custom `Dispatcher` lifecycle is still provisional pending validation against native network adapters. This wrapper is not an official gRPC wire binding. Define and test framing, cancellation, error mapping, notifications, and extension payloads in your adapter before claiming interoperability.
+
 ## Recap
 
 * `Client("http://.../mcp")` (a URL) connects over Streamable HTTP, the production transport.

docs/handlers/context.md

@@ -63,6 +63,7 @@ The injected object is small. Besides `request_id`:
 * `await ctx.report_progress(progress, total, message)`: stream progress back to the caller during a long call. The whole story is in **[Progress](progress.md)**.
 * `await ctx.elicit(message, schema)` and `await ctx.elicit_url(...)`: pause the tool and ask the user a question. That's **[Elicitation](elicitation.md)**.
 * `ctx.session`: the server's side of the conversation with this client. Notifications you send to the client live here; the last section uses it.
+* `ctx.transport`: transport metadata supplied by the dispatcher. Custom adapters can attach a `TransportContext` subclass; see [Custom transports](../run/index.md#custom-transports). The SDK populates it for dispatched requests; manually constructed request contexts may leave it `None`. It does not change the existing `ctx.headers` behavior.
 * `ctx.headers`: the request headers the transport carried, or `None` on stdio. Read a custom header with `(ctx.headers or {}).get("x-...")`. Headers are client-supplied input - fine for a locale or a feature flag, never an identity.
 * `ctx.request_context`: the raw per-request record. The field you'll reach for is `lifespan_context`, the object your startup code yielded (see **[Lifespan](lifespan.md)**).
 

docs/run/authorization.md

@@ -42,7 +42,7 @@ The SDK has no opinion about what a valid token looks like. You tell it, by impl
 
 ## What you get over HTTP
 
-Authorization lives in HTTP headers, so it exists only on the HTTP transports. Run it on the one you deploy: `mcp.run(transport="streamable-http")` puts it on `http://127.0.0.1:8000/mcp`, and **[Running your server](index.md)** has the rest. The app now has two routes:
+The SDK's built-in OAuth integration uses HTTP headers, so it applies only to HTTP transports. Run it on the one you deploy: `mcp.run(transport="streamable-http")` puts it on `http://127.0.0.1:8000/mcp`, and **[Running your server](index.md)** has the rest. The app now has two routes:
 
 ```text
 /mcp
@@ -104,6 +104,21 @@ Call `whoami` with `Authorization: Bearer alice-token` and the model reads:
 alice (scopes: notes:read)
 ```
 
+## Custom transport identities
+
+```python title="server.py"
+--8<-- "docs_src/authorization/tutorial003.py"
+```
+
+Have your adapter attach `VerifiedPeer` only after authenticating the caller. `runtime.connect(transport_builder=...)` passes that metadata to handlers as `ctx.transport`. Use a stable, namespaced principal that distinguishes the issuing authority and user, not a display name or a client-supplied `_meta` field.
+
+The existing `RequestStateSecurity.bind_principal` hook binds sealed request state to this identity. Another principal cannot replay it. Raising when verified metadata is absent prevents state from silently becoming anonymous. This hook protects multi-round-trip state; it does not authenticate connections or authorize ordinary tool calls. Those checks still belong at the adapter boundary and in your application policy.
+
+The generated key suits a single process. Share keys across workers when retries can reach another instance, as described in [Protecting request state](../handlers/multi-round-trip.md#protecting-requeststate).
+
+!!! warning "Broker credentials are not publisher identity"
+    A service's broker credentials authenticate the service, not every publisher. Bind peers through broker-enforced topic or queue permissions, or verify an end-user credential yourself. Validate reply destinations before sending data. The SDK's `get_access_token()` remains an HTTP OAuth helper; custom transport metadata does not populate it automatically.
+
 ## The half the SDK doesn't do
 
 The SDK gives you the resource-server half: verify, advertise, refuse. It does not give you a login page, a consent screen, or a token.

docs/run/index.md

@@ -88,6 +88,31 @@ Each transport has its own keyword arguments, all on `run()`:
 
 `run()` is the short road. The moment you need more (your server mounted inside an existing app, two servers in one process, CORS for browser clients), you build the ASGI app yourself and hand it to any ASGI host. That is **[Add to an existing app](asgi.md)**.
 
+## Custom transports
+
+```python title="custom_transport.py"
+--8<-- "docs_src/client_transports/tutorial005.py"
+```
+
+`server.serve()` returns a context manager yielding a `ServerRuntime`. It starts application lifespan once and shares that state across the connections you supply. Both `MCPServer` and the low-level `Server` expose this API. It does not open a network listener or connect to a broker.
+
+Call `await runtime.connect(transport)` for each logical peer. The runtime opens the transport and serves it in the background. For message streams, the call returns when the transport is open, before MCP negotiation. For a dispatcher transport, it also waits for the dispatcher to signal readiness. Each peer has its own request-ID state; message streams negotiate their protocol era independently.
+
+| Option | Behavior |
+| --- | --- |
+| `server.serve(max_connections=100)` | Limits active connections. `connect()` waits for capacity before opening another transport. |
+| `runtime.connect(..., transport_builder=...)` | Builds each inbound message's `TransportContext`, available as `ctx.transport` in handlers. |
+| `runtime.connect(..., session_id=...)` | Supplies an optional identifier for a handshake-era connection. It is not authentication. |
+
+The default connection limit prevents an adapter from opening unlimited peers. Await admission in your listener instead of spawning unbounded tasks that wait for a slot. Message-size limits, broker queue limits, and per-peer request limits remain the adapter's responsibility.
+
+An error opening a transport reaches the caller of `connect()`. A later connection failure is logged and closes that peer without cancelling other peers. Exiting `server.serve()` stops admission, cancels active work, closes transports, and then exits application lifespan. Transport cleanup and lifespan cleanup each have a five-second cancellation deadline; cleanup code must cooperate with cancellation. Cleanup timeouts do not suppress an earlier listener or dispatcher startup failure. Dispatchers must join their handlers before returning; these deadlines do not permit closing application resources while a handler still uses them. Code that ignores cancellation can delay that join, so enforce hard process deadlines outside the SDK. Do not retain a runtime after its context exits.
+
+!!! warning "A peer label is not an identity"
+    The example attaches a label for demonstration. A real adapter must authenticate and authorize callers before binding identity to a request. Authenticating your server's broker connection does not authenticate every publisher. Validate reply destinations instead of forwarding messages to arbitrary client-supplied topics or queues.
+
+The [client transport contract](../client/transports.md#implement-a-message-transport) describes message types, resource ownership, and connection loss. For a native RPC binding, `runtime.connect()` also accepts an explicit [dispatcher transport](../client/transports.md#integrate-a-native-dispatcher). That entry serves modern per-request envelopes, not legacy handshakes. The built-in `run()` forms remain unchanged.
+
 ## Server settings
 
 A couple of things about running are not about the transport. They are constructor arguments:

docs_src/authorization/tutorial003.py

@@ -0,0 +1,27 @@
+import secrets
+from dataclasses import dataclass
+
+from mcp.server import ServerRequestContext
+from mcp.server.mcpserver import MCPServer
+from mcp.server.request_state import RequestStateSecurity
+from mcp.shared.transport import TransportContext
+
+
+@dataclass(kw_only=True, frozen=True)
+class VerifiedPeer(TransportContext):
+    principal: str
+
+
+def principal(ctx: ServerRequestContext) -> str:
+    if not isinstance(ctx.transport, VerifiedPeer):
+        raise ValueError("Verified transport identity is required")
+    return ctx.transport.principal
+
+
+mcp = MCPServer(
+    "broker-service",
+    request_state_security=RequestStateSecurity(
+        keys=[secrets.token_bytes(32)],
+        bind_principal=principal,
+    ),
+)

docs_src/client_transports/tutorial005.py

@@ -0,0 +1,57 @@
+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)

docs_src/client_transports/tutorial006.py

@@ -0,0 +1,54 @@
+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)

src/mcp/client/_transport.py

@@ -1,21 +1,5 @@
 """Transport protocol for MCP clients."""
 
-from __future__ import annotations
-
-from contextlib import AbstractAsyncContextManager
-from typing import Protocol
-
-from mcp.shared._stream_protocols import ReadStream, WriteStream
-from mcp.shared.message import SessionMessage
+from mcp.shared.transport import ReadStream, Transport, TransportStreams, WriteStream
 
 __all__ = ["ReadStream", "WriteStream", "Transport", "TransportStreams"]
-
-TransportStreams = tuple[ReadStream[SessionMessage | Exception], WriteStream[SessionMessage]]
-
-
-class Transport(AbstractAsyncContextManager[TransportStreams], Protocol):
-    """Protocol for MCP transports.
-
-    A transport is an async context manager that yields read and write streams
-    for bidirectional communication with an MCP server.
-    """