Add an MQTT 5 transport example

.github/workflows/shared.yml

@@ -121,6 +121,16 @@ jobs:
           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: Start isolated broker fixtures
+        run: docker compose -p mcp-sdk-transport-check -f examples/transports/compose.yaml up --wait
+      - name: Check live MQTT routing and peer loss
+        run: |
+          for script in demo_mqtt.py demo_mqtt_disconnect.py; do
+            uv run --frozen --no-sync --package mcp-transport-examples --group dev python "examples/transports/$script"
+          done
+      - name: Stop broker fixtures
+        if: always()
+        run: docker compose -p mcp-sdk-transport-check -f examples/transports/compose.yaml down --volumes
       - name: Retain adapter test results
         if: always()
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1

examples/transports/README.md

@@ -63,3 +63,45 @@ The gRPC cassette tests record real calls with `cassetter` and replay with `--re
 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.
+
+## MQTT 5
+
+```bash
+docker compose -p mcp-sdk-transport-check -f examples/transports/compose.yaml up --wait
+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_mqtt.py
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv run --frozen --package mcp-transport-examples python examples/transports/demo_mqtt_disconnect.py
+docker compose -p mcp-sdk-transport-check -f examples/transports/compose.yaml down --volumes
+```
+
+The programs check concurrent calls for two peers, both server APIs, and `legacy`, `auto`, and pinned `2026-07-28` clients. The disconnect check uses a broker-forced session takeover and requires a pending call to receive `CONNECTION_CLOSED` without relying on its request timeout. CI runs these live checks separately from cassette replay.
+
+`demo_mqtt.py` contains complete connection setup. You own and enter the MQTT client before entering `mqtt_transport()`. The adapter unsubscribes on exit but does not close the borrowed client. Each logical peer gets a dedicated client and messages iterator; session identifiers are agreed out of band. Discovery and multiplexing are not implemented.
+
+### MQTT wire binding
+
+| Property | Value |
+| --- | --- |
+| Requests | `mcp/<principal>/<session>/requests` |
+| Replies | `mcp/<principal>/<session>/responses` |
+| Framing | One JSON-RPC message per publish |
+| Delivery | QoS 2 only |
+| Close | Empty payload |
+| Retention | Never retain commands; reject retained deliveries |
+| Expiry | MQTT message expiry, default 60 seconds |
+| Message limit | 4 MiB by default |
+| Reconnect | Fail old calls and establish fresh topics; never replay requests |
+
+Configure a QoS-2, non-retained Last Will with an empty payload on the outgoing topic before CONNECT. The broker publishes it on unexpected disconnection; the example uses a 15-second keepalive to bound detection of a silent network loss. An already-connected client cannot acquire a Last Will through this adapter. Use a client request timeout for startup failures before the peer subscribes: a non-retained will is not replayed to a later subscriber.
+
+QoS 2 handles protocol retransmissions within a session, not exactly-once tool execution. Republishing a request can repeat its side effects. Malformed messages become recoverable stream exceptions; connection loss ends the read stream.
+
+### MQTT authorization and limits
+
+The local fixture uses per-user topic ACLs and `use_username_as_clientid true`, so another authenticated user cannot evict a peer by claiming its client ID. It supports one connection per credential; server routes use separate `server-alice` and `server-bob` users. Deployments needing multiple sessions per credential need broker authorization of client-ID namespaces instead.
+
+The fixture uses public test credentials, binds only to localhost, and disables persistence. Do not deploy it. Use TLS and your broker's authorization policy in production, and bind request state to a verified, authority-qualified principal through `RequestStateSecurity.bind_principal`. A successful SUBACK is not proof that Mosquitto's ACL permits message delivery.
+
+The example bounds aiomqtt's incoming queue at 256 messages, but aiomqtt can drop messages when it fills. Its public publish callback also discards negative broker reason codes. These are unresolved reliability blockers, not successful execution acknowledgments. Request timeouts and monitoring are required; neither the queue bound nor MQTT QoS bounds concurrently executing tool handlers.
+
+`cassetter` has no MQTT interceptor. Full broker branch coverage, saturation and failure validation, and production TLS authorization remain open gates. The provider uses asyncio and requires a selector event loop on Windows; Trio and Windows support have not been validated. Change the local port with `MQTT_TEST_PORT` (default 13883).

examples/transports/brokers/mosquitto.acl

@@ -0,0 +1,18 @@
+user server-alice
+topic read mcp/alice/+/requests
+topic write mcp/alice/+/responses
+
+user server-bob
+topic read mcp/bob/+/requests
+topic write mcp/bob/+/responses
+
+user health
+topic write health
+
+user alice
+topic write mcp/alice/+/requests
+topic read mcp/alice/+/responses
+
+user bob
+topic write mcp/bob/+/requests
+topic read mcp/bob/+/responses

examples/transports/brokers/mosquitto.conf

@@ -0,0 +1,11 @@
+listener 1883
+allow_anonymous false
+use_username_as_clientid true
+password_file /tmp/passwords
+acl_file /mosquitto/config/mosquitto.acl
+persistence false
+log_dest stdout
+log_type all
+max_packet_size 4195328
+max_inflight_messages 32
+max_queued_messages 256

examples/transports/compose.yaml

@@ -0,0 +1,39 @@
+services:
+  mqtt:
+    image: eclipse-mosquitto:2.0.22@sha256:212f89e1eaeb2c322d6441b64396e3346026674db8fa9c27beac293405c32b3c
+    ports:
+      - "127.0.0.1:${MQTT_TEST_PORT:-13883}:1883"
+    volumes:
+      - ./brokers/mosquitto.conf:/mosquitto/config/mosquitto.conf:ro
+      - ./brokers/mosquitto.acl:/mosquitto/config/mosquitto.acl:ro
+    entrypoint: ["/bin/sh", "-ec"]
+    command:
+      - |
+        mosquitto_passwd -b -c /tmp/passwords server-alice test-server-alice-password
+        mosquitto_passwd -b /tmp/passwords server-bob test-server-bob-password
+        mosquitto_passwd -b /tmp/passwords health test-health-password
+        mosquitto_passwd -b /tmp/passwords alice test-alice-password
+        mosquitto_passwd -b /tmp/passwords bob test-bob-password
+        chmod 644 /tmp/passwords
+        exec mosquitto -c /mosquitto/config/mosquitto.conf
+    healthcheck:
+      test:
+        [
+          "CMD",
+          "mosquitto_pub",
+          "-h",
+          "127.0.0.1",
+          "-u",
+          "health",
+          "-P",
+          "test-health-password",
+          "-t",
+          "health",
+          "-m",
+          "",
+          "-q",
+          "2",
+        ]
+      interval: 1s
+      timeout: 3s
+      retries: 20

examples/transports/demo_common.py

@@ -0,0 +1,97 @@
+"""Shared live-broker checks for the reference adapters."""
+
+from collections.abc import AsyncIterator, Awaitable, Callable
+from contextlib import AsyncExitStack, asynccontextmanager
+from dataclasses import dataclass
+from functools import partial
+from typing import Any, TypeAlias
+from uuid import uuid4
+
+import anyio
+from mcp import Client
+from mcp.server import Server, ServerRequestContext
+from mcp.server.mcpserver import Context, MCPServer
+from mcp.shared.transport import MessageMetadata, Transport, TransportContext
+from mcp.types import CallToolRequestParams, CallToolResult, ListToolsResult, PaginatedRequestParams, TextContent, Tool
+
+TransportFactory: TypeAlias = Callable[[AsyncExitStack, str, str, bool], Awaitable[Transport]]
+
+
+@dataclass(kw_only=True, frozen=True)
+class BrokerContext(TransportContext):
+    principal: str
+
+
+def peer_context(metadata: MessageMetadata, *, principal: str, kind: str) -> BrokerContext:
+    return BrokerContext(kind=kind, can_send_request=True, principal=principal)
+
+
+async def verify(factory: TransportFactory, *, kind: str, highlevel: bool, mode: str) -> None:
+    """Check real concurrent calls, peer metadata, both server APIs, and shared lifespan."""
+    entered = {"alice": anyio.Event(), "bob": anyio.Event()}
+    sessions = {principal: uuid4().hex for principal in entered}
+    lifecycle: list[str] = []
+
+    async def identity(transport: TransportContext | None) -> str:
+        assert isinstance(transport, BrokerContext)
+        principal = transport.principal
+        entered[principal].set()
+        await entered["bob" if principal == "alice" else "alice"].wait()
+        return principal
+
+    @asynccontextmanager
+    async def lifespan(server: Server[Any] | MCPServer[Any]) -> AsyncIterator[None]:
+        lifecycle.append("start")
+        try:
+            yield None
+        finally:
+            lifecycle.append("stop")
+
+    if highlevel:
+        server = MCPServer("Broker", lifespan=lifespan)
+
+        @server.tool()
+        async def identify(ctx: Context) -> str:
+            return await identity(ctx.transport)
+
+    else:
+
+        async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
+            return ListToolsResult(tools=[Tool(name="identify", input_schema={"type": "object"})])
+
+        async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
+            assert params.name == "identify"
+            return CallToolResult(content=[TextContent(text=await identity(ctx.transport))])
+
+        server = Server("Broker", lifespan=lifespan, on_list_tools=list_tools, on_call_tool=call_tool)
+
+    results: dict[str, str] = {}
+
+    async def call(client: Client, principal: str) -> None:
+        result = await client.call_tool("identify")
+        content = result.content[0]
+        assert isinstance(content, TextContent)
+        results[principal] = content.text
+
+    async with AsyncExitStack() as stack:
+        transports = {
+            principal: await factory(stack, principal, session, True) for principal, session in sessions.items()
+        }
+        runtime = await stack.enter_async_context(server.serve())
+        clients: dict[str, Client] = {}
+        for principal, transport in transports.items():
+            await runtime.connect(
+                transport,
+                session_id=sessions[principal],
+                transport_builder=partial(peer_context, principal=principal, kind=kind),
+            )
+            client_transport = await factory(stack, principal, sessions[principal], False)
+            clients[principal] = await stack.enter_async_context(
+                Client(client_transport, mode=mode, read_timeout_seconds=5)
+            )
+        async with anyio.create_task_group() as tg:
+            for principal, client in clients.items():
+                tg.start_soon(call, client, principal)
+        assert results == {"alice": "alice", "bob": "bob"}
+        assert lifecycle == ["start"]
+    assert lifecycle == ["start", "stop"]

examples/transports/demo_mqtt.py

@@ -0,0 +1,42 @@
+"""Exercise the MQTT 5 adapter against the local Mosquitto broker."""
+
+import os
+from contextlib import AsyncExitStack
+
+import aiomqtt
+import anyio
+from mcp.shared.transport import Transport
+
+from demo_common import verify
+from mcp_transport_examples.mqtt import mqtt_transport
+
+
+async def open_transport(stack: AsyncExitStack, principal: str, session: str, server_side: bool) -> Transport:
+    user = f"server-{principal}" if server_side else principal
+    topic = f"mcp/{principal}/{session}"
+    incoming, outgoing = ("requests", "responses") if server_side else ("responses", "requests")
+    client = await stack.enter_async_context(
+        aiomqtt.Client(
+            "127.0.0.1",
+            int(os.environ.get("MQTT_TEST_PORT", "13883")),
+            username=user,
+            password=f"test-{user}-password",
+            identifier=f"mcp-{principal}-{session}-{'server' if server_side else 'client'}",
+            protocol=aiomqtt.ProtocolVersion.V5,
+            keepalive=15,
+            will=aiomqtt.Will(f"{topic}/{outgoing}", payload=b"", qos=2, retain=False),
+            max_queued_incoming_messages=256,
+        )
+    )
+    return mqtt_transport(client, incoming_topic=f"{topic}/{incoming}", outgoing_topic=f"{topic}/{outgoing}")
+
+
+async def main() -> None:
+    for highlevel in (False, True):
+        for mode in ("legacy", "auto", "2026-07-28"):
+            with anyio.fail_after(5):
+                await verify(open_transport, kind="mqtt", highlevel=highlevel, mode=mode)
+
+
+if __name__ == "__main__":
+    anyio.run(main)

examples/transports/demo_mqtt_disconnect.py

@@ -0,0 +1,71 @@
+"""Check remote-peer loss using Mosquitto's session-takeover behavior."""
+
+import os
+from contextlib import AsyncExitStack
+from uuid import uuid4
+
+import aiomqtt
+import anyio
+from mcp import Client, MCPError
+from mcp.server.mcpserver import MCPServer
+from mcp.types import CONNECTION_CLOSED
+
+from demo_mqtt import open_transport
+
+
+async def main() -> None:
+    """A broker-forced disconnect publishes the configured Last Will and settles a waiting MCP call."""
+    entered = anyio.Event()
+    finished = anyio.Event()
+    server = MCPServer("peer loss")
+    session = uuid4().hex
+
+    @server.tool()
+    async def hold() -> str:
+        entered.set()
+        await anyio.sleep_forever()
+        raise NotImplementedError
+
+    with anyio.fail_after(5):
+        async with AsyncExitStack() as stack:
+            server_transport = await open_transport(stack, "alice", session, True)
+            client_transport = await open_transport(stack, "alice", session, False)
+            runtime = await stack.enter_async_context(server.serve())
+            await runtime.connect(server_transport)
+            client = await stack.enter_async_context(Client(client_transport, read_timeout_seconds=None))
+
+            async def call() -> None:
+                try:
+                    await client.call_tool("hold")
+                except MCPError as exc:
+                    assert exc.code == CONNECTION_CLOSED
+                else:
+                    raise AssertionError("Remote peer loss must fail the pending call")
+                finished.set()
+
+            async with anyio.create_task_group() as tg:
+                tg.start_soon(call)
+                await entered.wait()
+                async with aiomqtt.Client(
+                    "127.0.0.1",
+                    int(os.environ.get("MQTT_TEST_PORT", "13883")),
+                    username="bob",
+                    password="test-bob-password",
+                    identifier="server-alice",
+                    protocol=aiomqtt.ProtocolVersion.V5,
+                ):
+                    assert [tool.name for tool in (await client.list_tools()).tools] == ["hold"]
+                    assert not finished.is_set()
+                async with aiomqtt.Client(
+                    "127.0.0.1",
+                    int(os.environ.get("MQTT_TEST_PORT", "13883")),
+                    username="server-alice",
+                    password="test-server-alice-password",
+                    identifier="server-alice",
+                    protocol=aiomqtt.ProtocolVersion.V5,
+                ):
+                    await finished.wait()
+
+
+if __name__ == "__main__":
+    anyio.run(main)