Add a native gRPC transport example

.github/workflows/shared.yml

@@ -96,6 +96,40 @@ jobs:
         if: runner.os != 'Windows'
         run: uv run --frozen --no-sync strict-no-cover
 
+  transport-examples:
+    name: transport examples (${{ matrix.python-version }})
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    strategy:
+      fail-fast: false
+      matrix:
+        python-version: ["3.10", "3.14"]
+    env:
+      UV_PROJECT_ENVIRONMENT: examples/transports/.venv
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+        with:
+          persist-credentials: false
+      - uses: astral-sh/setup-uv@20cfd1bf945f4377ade1205e4dbc17946fc9a30d # v10.0.1
+        with:
+          enable-cache: true
+          version: 0.9.5
+      - name: Install transport dependencies
+        run: uv sync --frozen --package mcp-transport-examples --group dev --python ${{ matrix.python-version }}
+      - name: Run all adapter regressions
+        run: >-
+          uv run --frozen --no-sync --package mcp-transport-examples --group dev
+          pytest -c examples/transports/pyproject.toml examples/transports/tests --record-mode=none
+          --junitxml=transport-results.xml
+      - name: Retain adapter test results
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: transport-results-${{ matrix.python-version }}
+          path: transport-results.xml
+      - name: Check adapter types
+        run: uv run --frozen --no-sync --package mcp-transport-examples --group dev pyright --project examples/transports
+
   readme-snippets:
     runs-on: ubuntu-latest
     steps:

examples/transports/README.md

@@ -0,0 +1,65 @@
+# Reference custom transports
+
+This package contains experimental adapters for the public MCP transport API. Installing `mcp` does not install their dependencies. The adapters are not production-ready transports or official MCP wire bindings.
+
+## Native gRPC
+
+```bash
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv sync --frozen --package mcp-transport-examples --group dev
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv run --frozen --package mcp-transport-examples python examples/transports/demo_grpc.py
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv run --frozen --package mcp-transport-examples python examples/transports/demo_grpc_features.py
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv run --frozen --package mcp-transport-examples --group dev pytest -c examples/transports/pyproject.toml examples/transports/tests --record-mode=none
+uv run --frozen pyright --project examples/transports
+```
+
+Run these commands from the repository root. The programs start their own loopback gRPC listener. No broker is needed. `grpc_client(channel)` and `grpc_server(listener)` return `DispatcherTransport` objects for the existing client and server runtime APIs. You own the channel and listener; runtime shutdown stops MCP handlers without taking ownership of other gRPC services on the listener.
+
+Register the binding through `runtime.connect(grpc_server(listener))` before starting the listener. Use one MCP binding per gRPC server. The server adapter rejects excess work at `max_requests`, which defaults to 64, rather than queuing unlimited waiting handlers.
+
+The binding serves modern per-request MCP envelopes. It has no legacy initialize handshake. Each MCP request is a native server-streaming RPC on `/mcp.transport.example.MCP/Call`; gRPC correlates calls and provides deadlines and cancellation. The auxiliary request ID supports MCP subscription correlation and stays local to each client's calls. Native progress uses the protobuf `report_progress` opt-in from `CallOptions["on_progress"]`; `_meta.progressToken` alone does not enable it. Progress carries the auxiliary ID for notification observers, while the originating RPC selects the callback without token-based demultiplexing.
+
+`mcp_transport_examples/rpc.proto` defines protobuf envelopes with JSON-encoded parameters, results, and error data. There is no JSON-RPC envelope. JSON payloads preserve arbitrary extension fields and integer precision, which protobuf `Struct` would otherwise lose through its floating-point number representation. This is an example binding, not compatibility with another project's gRPC schema.
+
+Notifications precede one terminal result or error, followed by end-of-stream. Progress, subscription acknowledgments, and change events use the originating RPC's response stream. Unsolicited notifications without a request channel are unsupported. Ordinary MCP errors preserve their code, message, and data. Native deadline failures become `REQUEST_TIMEOUT`; other gRPC failures become `CONNECTION_CLOSED` with the original status exception as their cause.
+
+The examples and regression tests check both server APIs, progress, subscriptions, multi-round-trip results, concurrent clients with colliding request IDs, caller cancellation, deadlines, runtime shutdown, borrowed-channel closure, and client shutdown during a blocked callback. Cancellation is signalled before handler cleanup begins. Active handlers and callbacks are joined before their owning resources close, including shielded cleanup that takes longer than five seconds. Code that ignores cancellation indefinitely can therefore hold shutdown indefinitely; enforce a hard process deadline in your supervisor rather than closing resources under running code. The SDK's five-second transport and application cleanup deadlines do not replace this join.
+
+### Generate the protobuf bindings
+
+```bash
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv run --frozen --package mcp-transport-examples --group dev python -m grpc_tools.protoc --proto_path=examples/transports --python_out=examples/transports --pyi_out=examples/transports examples/transports/mcp_transport_examples/rpc.proto
+```
+
+Use the pinned compiler. Generated implementation code is excluded from adapter coverage; regeneration checks its provenance.
+
+### TLS and peer identity
+
+```bash
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv run --frozen --package mcp-transport-examples --group dev pytest -c examples/transports/pyproject.toml examples/transports/tests/test_grpc_tls.py --record-mode=none
+```
+
+These tests create temporary certificate authorities and real local TLS endpoints. They check mutual TLS, server-only TLS, and plaintext connections. Missing or untrusted client certificates cannot reach MCP middleware when the listener requires client authentication. Forged MCP client information and gRPC invocation metadata do not change the verified identity.
+
+Configure TLS through your gRPC channel and listener credentials. Set `require_client_auth=True` on `grpc.ssl_server_credentials()` when clients must present a certificate. Handlers receive `GRPCContext.peer_identity_key` and `peer_identities` from gRPC's native authentication context. Without client authentication, these are `None` and an empty tuple, including on encrypted server-only TLS connections.
+
+Certificate validation is not application authorization. The identity values do not identify their issuing authority. If you trust independent authorities that can issue the same common name or subject alternative name, do not treat that name as a globally unique principal. Choose a trust-domain namespace and certificate-issuance policy before using these values with `RequestStateSecurity.bind_principal`. An issuer-name string or untrusted invocation metadata cannot supply that trust boundary.
+
+### Event-loop lifetime
+
+```bash
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv run --frozen --package mcp-transport-examples python examples/transports/reproduce_grpc_loop_shutdown.py
+```
+
+This diagnostic intentionally fails when a native completion targets a closed loop. It reproduces the limitation without importing MCP. The failure was observed with `grpcio==1.84.0` on macOS and Python 3.14.6; the assertion includes the runtime versions.
+
+Keep one long-lived asyncio event loop per process or test worker. Create, use, and close all gRPC resources on that loop. Repeated `anyio.run()` or `asyncio.run()` lifetimes are outside this adapter's current support. gRPC's process-wide completion queue can deliver cancelled connectivity-watch callbacks after `channel.close()` returns. Joining MCP handlers does not drain those native callbacks. The adapter tests keep one AnyIO runner alive for the session; they do not suppress loop errors or claim an upstream correction.
+
+### Validation boundaries
+
+The dedicated CI job runs the adapter suite on Python 3.10 and 3.14 and retains JUnit results. Final compatibility review and cross-platform validation remain open gates.
+
+The gRPC cassette tests record real calls with `cassetter` and replay with `--record-mode=none`. They check payload fidelity, progress, and application errors. They also compare serialized requests with the recording: the current matcher matches only the RPC method, which is insufficient for a generic MCP binding. Each cassette contains one RPC to avoid replaying a newly recorded call as the response to a different request during recording.
+
+The lifecycle, capacity, and malformed-frame regression tests own a gRPC server inside the test process. That server is the software under test, not an external service; replaying its outputs would bypass the behavior being checked. Cassette tests separately compare recorded native results with the current in-process MCP handler. `cassetter` lacks parts of the streaming-call cancellation interface, so it is not used to stand in for live lifecycle checks.
+
+Binary protobuf payloads are not pattern-scrubbed. Inspect new cassettes before committing them; the checked-in recordings contain only public test data.

examples/transports/demo_grpc.py

@@ -0,0 +1,71 @@
+"""Exercise the native gRPC binding over a real loopback connection."""
+
+from contextlib import AsyncExitStack
+
+import anyio
+import grpc.aio
+from mcp import Client
+from mcp.server import Server, ServerRequestContext
+from mcp.server.mcpserver import Context, MCPServer
+from mcp.types import CallToolRequestParams, CallToolResult, ListToolsResult, PaginatedRequestParams, TextContent, Tool
+
+from mcp_transport_examples.grpc import grpc_client, grpc_server
+from mcp_transport_examples.grpc_context import GRPCContext
+
+
+async def verify(highlevel: bool, mode: str) -> None:
+    if highlevel:
+        server = MCPServer("Native gRPC")
+
+        @server.tool()
+        async def echo(value: str, ctx: Context) -> str:
+            assert isinstance(ctx.transport, GRPCContext)
+            assert not ctx.transport.can_send_request
+            await ctx.report_progress(1, 2, "halfway")
+            return value
+
+    else:
+
+        async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
+            return ListToolsResult(tools=[Tool(name="echo", input_schema={"type": "object"})])
+
+        async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
+            assert params.name == "echo"
+            assert isinstance(ctx.transport, GRPCContext)
+            assert params.arguments is not None
+            await ctx.session.report_progress(1, 2, "halfway")
+            return CallToolResult(content=[TextContent(text=str(params.arguments["value"]))])
+
+        server = Server("Native gRPC", on_list_tools=list_tools, on_call_tool=call_tool)
+
+    updates: list[tuple[float, float | None, str | None]] = []
+
+    async def progress(progress: float, total: float | None, message: str | None) -> None:
+        updates.append((progress, total, message))
+
+    async with AsyncExitStack() as stack:
+        listener = grpc.aio.server()
+        port = listener.add_insecure_port("127.0.0.1:0")
+        stack.push_async_callback(listener.stop, 0)
+        runtime = await stack.enter_async_context(server.serve())
+        await runtime.connect(grpc_server(listener))
+        await listener.start()
+        channel = await stack.enter_async_context(grpc.aio.insecure_channel(f"127.0.0.1:{port}"))
+        client = await stack.enter_async_context(Client(grpc_client(channel), mode=mode))
+        value = "MCP without a JSON-RPC envelope"
+        result = await client.call_tool("echo", {"value": value}, progress_callback=progress)
+        content = result.content[0]
+        assert isinstance(content, TextContent)
+        assert content.text == value
+        assert updates == [(1, 2, "halfway")]
+
+
+async def main() -> None:
+    for highlevel in (False, True):
+        for mode in ("auto", "2026-07-28"):
+            with anyio.fail_after(5):
+                await verify(highlevel, mode)
+
+
+if __name__ == "__main__":
+    anyio.run(main)

examples/transports/demo_grpc_features.py

@@ -0,0 +1,91 @@
+"""Live checks for subscriptions and multi-round-trip results over native gRPC."""
+
+from contextlib import AsyncExitStack
+
+import anyio
+import grpc.aio
+from mcp import Client
+from mcp.client import ClientRequestContext
+from mcp.client.subscriptions import ToolsListChanged
+from mcp.server.mcpserver import Context, MCPServer
+from mcp.types import ElicitRequest, ElicitRequestFormParams, ElicitRequestParams, ElicitResult, InputRequiredResult
+
+from mcp_transport_examples.grpc import grpc_client, grpc_server
+
+
+async def verify() -> None:
+    server = MCPServer("native features")
+    entered = {"alice": anyio.Event(), "bob": anyio.Event()}
+
+    @server.tool()
+    async def overlap(label: str, ctx: Context) -> str:
+        assert ctx.request_context.request_id == 0
+        entered[label].set()
+        await entered["bob" if label == "alice" else "alice"].wait()
+        return label
+
+    @server.tool()
+    async def announce(ctx: Context) -> str:
+        await ctx.notify_tools_changed()
+        return "announced"
+
+    @server.tool()
+    async def confirm(ctx: Context) -> str | InputRequiredResult:
+        if ctx.input_responses is not None:
+            assert ctx.request_state == "awaiting confirmation"
+            answer = ctx.input_responses["confirm"]
+            assert isinstance(answer, ElicitResult)
+            assert answer.action == "accept"
+            return "confirmed"
+        return InputRequiredResult(
+            input_requests={
+                "confirm": ElicitRequest(
+                    params=ElicitRequestFormParams(
+                        message="Confirm?", requested_schema={"type": "object", "properties": {}}
+                    )
+                )
+            },
+            request_state="awaiting confirmation",
+        )
+
+    async def elicit(context: ClientRequestContext, params: ElicitRequestParams) -> ElicitResult:
+        return ElicitResult(action="accept", content={})
+
+    async with AsyncExitStack() as stack:
+        listener = grpc.aio.server()
+        port = listener.add_insecure_port("127.0.0.1:0")
+        stack.push_async_callback(listener.stop, 0)
+        runtime = await stack.enter_async_context(server.serve())
+        await runtime.connect(grpc_server(listener))
+        await listener.start()
+        channel = await stack.enter_async_context(grpc.aio.insecure_channel(f"127.0.0.1:{port}"))
+        client = await stack.enter_async_context(Client(grpc_client(channel), elicitation_callback=elicit))
+        async with client.listen(tools_list_changed=True) as subscription:
+            result = await client.call_tool("announce")
+            assert result.structured_content == {"result": "announced"}
+            event = await anext(subscription)
+            assert isinstance(event, ToolsListChanged)
+        result = await client.call_tool("confirm")
+        assert result.structured_content == {"result": "confirmed"}
+
+        clients: dict[str, Client] = {}
+        for label in entered:
+            peer_channel = await stack.enter_async_context(grpc.aio.insecure_channel(f"127.0.0.1:{port}"))
+            clients[label] = await stack.enter_async_context(Client(grpc_client(peer_channel), mode="2026-07-28"))
+
+        async def call(label: str, client: Client) -> None:
+            result = await client.call_tool("overlap", {"label": label})
+            assert result.structured_content == {"result": label}
+
+        async with anyio.create_task_group() as tg:
+            for label, peer in clients.items():
+                tg.start_soon(call, label, peer)
+
+
+async def main() -> None:
+    with anyio.fail_after(5):
+        await verify()
+
+
+if __name__ == "__main__":
+    anyio.run(main)

examples/transports/mcp_transport_examples/__init__.py

@@ -0,0 +1 @@
+__all__ = []

examples/transports/mcp_transport_examples/_grpc_codec.py

@@ -0,0 +1,46 @@
+"""JSON payloads inside the experimental protobuf binding."""
+
+from __future__ import annotations
+
+import json
+import math
+from typing import Any, NoReturn, cast
+
+MAX_PAYLOAD_SIZE = 4 * 1024 * 1024
+RPC_METHOD = "/mcp.transport.example.MCP/Call"
+
+
+def encode_json(value: Any) -> bytes:
+    payload = json.dumps(value, allow_nan=False, ensure_ascii=True, separators=(",", ":")).encode("utf-8")
+    if len(payload) > MAX_PAYLOAD_SIZE:
+        raise ValueError("Payload exceeds the gRPC binding's size limit")
+    return payload
+
+
+def decode_json(payload: bytes) -> Any:
+    if len(payload) > MAX_PAYLOAD_SIZE:
+        raise ValueError("Payload exceeds the gRPC binding's size limit")
+
+    def finite_float(value: str) -> float:
+        number = float(value)
+        if not math.isfinite(number):
+            raise ValueError("Non-finite JSON number")
+        return number
+
+    try:
+        return json.loads(payload, parse_constant=reject_constant, parse_float=finite_float)
+    except RecursionError as exc:
+        raise ValueError("JSON payload is too deeply nested") from exc
+
+
+def decode_object(payload: bytes, *, nullable: bool = False) -> dict[str, Any] | None:
+    value = decode_json(payload)
+    if isinstance(value, dict):
+        return cast("dict[str, Any]", value)
+    if nullable and value is None:
+        return None
+    raise ValueError("Expected a JSON object")
+
+
+def reject_constant(value: str) -> NoReturn:
+    raise ValueError(f"Invalid JSON constant: {value}")

examples/transports/mcp_transport_examples/grpc.py

@@ -0,0 +1,38 @@
+"""Factories for the experimental native protobuf transport."""
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+import grpc.aio
+from mcp.shared.dispatcher import Dispatcher
+from mcp.shared.transport import DispatcherTransport, TransportContext
+
+from mcp_transport_examples.grpc_client import GRPCClientDispatcher
+from mcp_transport_examples.grpc_server import GRPCServerDispatcher
+
+
+def grpc_client(channel: grpc.aio.Channel) -> DispatcherTransport:
+    """Use a borrowed channel with `Client`; the caller owns TLS, credentials, and channel closure."""
+
+    @asynccontextmanager
+    async def connection() -> AsyncIterator[Dispatcher[TransportContext]]:
+        yield GRPCClientDispatcher(channel)
+
+    return DispatcherTransport(connection())
+
+
+def grpc_server(server: grpc.aio.Server, *, max_requests: int = 64) -> DispatcherTransport:
+    """Attach one MCP binding to a borrowed server before starting its listener.
+
+    Connect this transport to `ServerRuntime`, then start the gRPC server.
+    The caller owns the listener and other registered gRPC services. Runtime
+    shutdown cancels MCP handlers without stopping unrelated services.
+    """
+
+    @asynccontextmanager
+    async def connection() -> AsyncIterator[Dispatcher[TransportContext]]:
+        yield GRPCServerDispatcher(server, max_requests=max_requests)
+
+    return DispatcherTransport(connection())