Add an AMQP 0.9.1 transport example

.github/workflows/shared.yml

@@ -123,9 +123,9 @@ jobs:
           --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
+      - name: Check live broker routing and peer loss
         run: |
-          for script in demo_mqtt.py demo_mqtt_disconnect.py; do
+          for script in demo_amqp.py demo_amqp_permissions.py demo_amqp_lifecycle.py 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

examples/transports/README.md

@@ -105,3 +105,63 @@ The fixture uses public test credentials, binds only to localhost, and disables
 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).
+
+## AMQP 0.9.1
+
+```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_amqp.py
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv run --frozen --package mcp-transport-examples python examples/transports/demo_amqp_permissions.py
+UV_PROJECT_ENVIRONMENT=examples/transports/.venv uv run --frozen --package mcp-transport-examples python examples/transports/demo_amqp_lifecycle.py
+docker compose -p mcp-sdk-transport-check -f examples/transports/compose.yaml down --volumes
+```
+
+The programs reuse the same two-peer application checks as MQTT for both server APIs and all three client modes. The permission check requires RabbitMQ to reject client declarations, response bindings, and publication through response, default, or foreign exchanges. It also tests an exchange named exactly like a response queue. The lifecycle check covers either close order, requests queued before the peer consumes, unroutable publication, and the remote-loss deadline limitation below. CI runs all three programs against the pinned RabbitMQ fixture.
+
+`demo_amqp.py` contains complete setup. The trusted server account declares and binds both directions before either transport enters. You own the connection and publisher-confirm channel; `amqp_transport()` only gets handles to existing resources and cancels its consumer without closing that borrowed channel. Keep the non-auto-delete exchanges alive until both peer transports have stopped; normal teardown deletes them, but cleanup after a process crash is manual. Queue expiry does not delete these exchanges; see [AMQP crash cleanup](#amqp-crash-cleanup). Deleting an exchange while a peer still publishes can make RabbitMQ close that peer's channel. Missing queues or exchanges fail at consumption or publication. Use a fresh queue pair for each logical connection and do not load-balance handshake-era traffic across independent sessions.
+
+### AMQP wire binding
+
+| Property | Value |
+| --- | --- |
+| Requests | `mcp.<principal>.<session>.requests` |
+| Replies | `mcp.<principal>.<session>.responses` |
+| Routing | One direct exchange named `<queue>.exchange` per receiving queue |
+| Framing | One JSON-RPC message per delivery; `application/json` content type |
+| Delivery | Mandatory publication with checked confirmations; acknowledge before SDK handoff |
+| Close | Empty JSON-typed message body; never reply to a received close |
+| Retention | Nondurable, auto-delete queues |
+| Expiry | Outgoing message TTL defaults to 60 seconds; the demo separately provisions a 60-second unused-queue expiry |
+| Message limit | 4 MiB by default |
+| Redelivery | Reject without requeue; never replay requests automatically |
+
+Acknowledging before SDK handoff avoids automatically rerunning uncertain work, but a process failure in that window can lose it. Publisher confirmations describe broker delivery, not tool completion or exactly-once execution. Applications still own idempotency.
+
+The adapter checks the publish confirmation and treats unroutable returns as a write failure, ending the logical connection without replay. The example also enables `on_return_raises=True` on its publisher-confirm channels. A return from an existing exchange does not close the borrowed channel.
+
+Queues hold at most 256 ready messages and reject publication on overflow. Consumer prefetch bounds unacknowledged deliveries, not concurrently executing tool handlers. Malformed messages become recoverable stream exceptions; channel closure ends the read stream.
+
+### AMQP crash cleanup
+
+```bash
+docker compose -p mcp-sdk-transport-check -f examples/transports/compose.yaml down --volumes
+```
+
+After stopping all demo peers, run this command to remove the local broker fixture and its disposable state. It resets both brokers in this Compose project.
+
+The provisioner registers exchange deletion for normal `AsyncExitStack` teardown only. A process crash skips those callbacks, and the demo has no startup orphan cleanup. On a shared broker, an administrator must identify abandoned sessions and manually delete their `mcp.<principal>.<session>.requests.exchange` and `mcp.<principal>.<session>.responses.exchange` resources. Confirm that neither peer is active before deleting them; do not sweep exchanges belonging to active sessions.
+
+### AMQP remote-peer loss
+
+The broker does not notify your response consumer when a remote request consumer disappears. An idle remote failure therefore does not automatically close this adapter's read stream. Set `Client(..., read_timeout_seconds=...)` to bound pending calls; the lifecycle program demonstrates `REQUEST_TIMEOUT` after the server connection disappears, not automatic `CONNECTION_CLOSED` detection. A subsequent unroutable write fails the connection, but is not an idle liveness monitor.
+
+Servers also need an application-level session lease or liveness monitor to release idle sessions. Automatic AMQP peer-loss detection remains an open limitation; do not use an unlimited request timeout while relying on broker connection heartbeats, which monitor only your own connection.
+
+### AMQP authorization and limits
+
+The fixture grants each client only publication rights on its request exchanges and consumption rights on its response queues. Clients cannot configure topology or bind queues. RabbitMQ write permissions apply to resource names, not resource types: granting write access to a response queue would also permit publication to an exchange with that name. Server-side provisioning removes the need for that grant. Client credentials also cannot publish through `amq.default` or another principal's exchange; messages cannot choose an arbitrary reply destination.
+
+The fixture uses public test credentials, listens only on localhost, and disables durable storage. Do not deploy it. Use TLS and broker authorization in production, and bind request state to verified, authority-qualified identity. Change the local port with `AMQP_TEST_PORT` (default 15673, avoiding RabbitMQ's standard management port).
+
+`cassetter` has no AMQP interceptor. Full broker branch coverage, broader delivery/failure validation, and production TLS checks remain open gates. The provider uses asyncio; Trio and Windows validation have not been completed.

examples/transports/brokers/rabbitmq-definitions.json

@@ -0,0 +1,13 @@
+{
+  "users": [
+    {"name": "server", "password_hash": "dGVzdISQMYLFRDkVmIyH/2iPR3elfgPHcZO7uFXGqTE0UeU9", "hashing_algorithm": "rabbit_password_hashing_sha256", "tags": []},
+    {"name": "alice", "password_hash": "dGVzdMhyAjoCEMSETc4cmiL+/OInoknje5+9BVEZ8SaVlj+Z", "hashing_algorithm": "rabbit_password_hashing_sha256", "tags": []},
+    {"name": "bob", "password_hash": "dGVzdFu3/YOufljrGKLUVvAqIh2R9WeQ87Ql039pWvRL9HZO", "hashing_algorithm": "rabbit_password_hashing_sha256", "tags": []}
+  ],
+  "vhosts": [{"name": "/"}],
+  "permissions": [
+    {"user": "server", "vhost": "/", "configure": ".*", "write": ".*", "read": ".*"},
+    {"user": "alice", "vhost": "/", "configure": "^$", "write": "^mcp\\.alice\\.[^.]+\\.requests\\.exchange$", "read": "^mcp\\.alice\\.[^.]+\\.responses$"},
+    {"user": "bob", "vhost": "/", "configure": "^$", "write": "^mcp\\.bob\\.[^.]+\\.requests\\.exchange$", "read": "^mcp\\.bob\\.[^.]+\\.responses$"}
+  ]
+}

examples/transports/brokers/rabbitmq.conf

@@ -0,0 +1,2 @@
+definitions.import_backend = local_filesystem
+definitions.local.path = /etc/rabbitmq/definitions.json

examples/transports/compose.yaml

@@ -1,4 +1,16 @@
 services:
+  amqp:
+    image: rabbitmq:4.1.8-alpine@sha256:1a087dd3a29b91448407409df70f4f6cb213ac0c269a62861bfe2a665f4ced03
+    ports:
+      - "127.0.0.1:${AMQP_TEST_PORT:-15673}:5672"
+    volumes:
+      - ./brokers/rabbitmq.conf:/etc/rabbitmq/rabbitmq.conf:ro
+      - ./brokers/rabbitmq-definitions.json:/etc/rabbitmq/definitions.json:ro
+    healthcheck:
+      test: ["CMD", "rabbitmq-diagnostics", "-q", "check_port_connectivity"]
+      interval: 2s
+      timeout: 5s
+      retries: 30
   mqtt:
     image: eclipse-mosquitto:2.0.22@sha256:212f89e1eaeb2c322d6441b64396e3346026674db8fa9c27beac293405c32b3c
     ports:

examples/transports/demo_amqp.py

@@ -0,0 +1,55 @@
+"""Exercise the AMQP 0.9.1 adapter against the local RabbitMQ broker."""
+
+import os
+from contextlib import AsyncExitStack
+
+import aio_pika
+import anyio
+from aio_pika.abc import AbstractChannel
+from mcp.shared.transport import Transport
+
+from demo_common import verify
+from mcp_transport_examples.amqp import amqp_transport
+
+
+async def open_transport(stack: AsyncExitStack, principal: str, session: str, server_side: bool) -> Transport:
+    user = "server" if server_side else principal
+    connection = await aio_pika.connect(
+        host="127.0.0.1",
+        port=int(os.environ.get("AMQP_TEST_PORT", "15673")),
+        login=user,
+        password=f"test-{user}-password",
+    )
+    await stack.enter_async_context(connection)
+    channel = await connection.channel(publisher_confirms=True, on_return_raises=True)
+    stack.push_async_callback(channel.close)
+    queue = f"mcp.{principal}.{session}"
+    if server_side:
+        await provision(stack, channel, queue)
+    incoming, outgoing = ("requests", "responses") if server_side else ("responses", "requests")
+    return amqp_transport(channel, incoming_queue=f"{queue}.{incoming}", outgoing_queue=f"{queue}.{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="amqp", highlevel=highlevel, mode=mode)
+
+
+async def provision(stack: AsyncExitStack, channel: AbstractChannel, prefix: str) -> None:
+    """Keep exchanges until every transport entered after this setup has exited."""
+    for direction in ("requests", "responses"):
+        name = f"{prefix}.{direction}"
+        destination = await channel.declare_queue(
+            name,
+            auto_delete=True,
+            arguments={"x-expires": 60_000, "x-max-length": 256, "x-overflow": "reject-publish"},
+        )
+        exchange = await channel.declare_exchange(f"{name}.exchange", auto_delete=False)
+        stack.push_async_callback(exchange.delete, if_unused=False)
+        await destination.bind(exchange, routing_key=name)
+
+
+if __name__ == "__main__":
+    anyio.run(main)

examples/transports/demo_amqp_lifecycle.py

@@ -0,0 +1,159 @@
+"""Check AMQP close ordering, routing readiness, and the remote-loss timeout boundary."""
+
+import os
+from contextlib import AsyncExitStack
+from uuid import uuid4
+
+import aio_pika
+import anyio
+from mcp import Client, MCPError
+from mcp.server.mcpserver import MCPServer
+from mcp.shared.transport import SessionMessage
+from mcp.types import CONNECTION_CLOSED, REQUEST_TIMEOUT, JSONRPCRequest
+
+from demo_amqp import provision
+from mcp_transport_examples.amqp import amqp_transport
+
+
+async def main() -> None:
+    """Exercise real broker behavior, including peer loss that requires a configured request deadline."""
+    port = int(os.environ.get("AMQP_TEST_PORT", "15673"))
+    for scenario in ("server-first", "client-first", "queued", "return", "raise-return", "peer-loss"):
+        with anyio.fail_after(5):
+            async with AsyncExitStack() as stack:
+                owner = await stack.enter_async_context(
+                    await aio_pika.connect(
+                        host="127.0.0.1",
+                        port=port,
+                        login="server",
+                        password="test-server-password",
+                    )
+                )
+                setup = await owner.channel()
+                stack.push_async_callback(setup.close)
+                prefix = f"mcp.alice.{uuid4().hex}"
+                await provision(stack, setup, prefix)
+                server_connection = await stack.enter_async_context(
+                    await aio_pika.connect(
+                        host="127.0.0.1",
+                        port=port,
+                        login="server",
+                        password="test-server-password",
+                    )
+                )
+                client_connection = await stack.enter_async_context(
+                    await aio_pika.connect(
+                        host="127.0.0.1",
+                        port=port,
+                        login="alice",
+                        password="test-alice-password",
+                    )
+                )
+                server_channel = await server_connection.channel()
+                stack.push_async_callback(server_channel.close)
+                client_channel = await client_connection.channel(
+                    publisher_confirms=True, on_return_raises=scenario != "return"
+                )
+                stack.push_async_callback(client_channel.close)
+                server_transport = amqp_transport(
+                    server_channel,
+                    incoming_queue=f"{prefix}.requests",
+                    outgoing_queue=f"{prefix}.responses",
+                    expiry=120,
+                )
+                client_transport = amqp_transport(
+                    client_channel,
+                    incoming_queue=f"{prefix}.responses",
+                    outgoing_queue=f"{prefix}.requests",
+                )
+                if scenario in ("server-first", "client-first"):
+                    first, second = (
+                        (server_transport, client_transport)
+                        if scenario == "server-first"
+                        else (client_transport, server_transport)
+                    )
+                    async with second as (receive, write):
+                        async with first:
+                            pass
+                        try:
+                            await receive.receive()
+                        except anyio.EndOfStream:
+                            pass
+                        else:
+                            raise AssertionError("The peer close must end the read stream")
+                        try:
+                            await write.send(SessionMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")))
+                        except anyio.ClosedResourceError:
+                            pass
+                        else:
+                            raise AssertionError("The peer close must also close the writer")
+                elif scenario == "queued":
+                    message = SessionMessage(
+                        JSONRPCRequest(
+                            jsonrpc="2.0",
+                            id=1,
+                            method="initialize",
+                            params={
+                                "protocolVersion": "2025-11-25",
+                                "capabilities": {},
+                                "clientInfo": {"name": "lifecycle", "version": "1"},
+                            },
+                        )
+                    )
+                    async with client_transport as (_, write):
+                        await write.send(message)
+                        async with server_transport as (read, _):
+                            received = await read.receive()
+                            assert isinstance(received, SessionMessage)
+                            assert received.message == message.message
+                elif scenario in ("return", "raise-return"):
+                    queue = await setup.get_queue(f"{prefix}.requests", ensure=False)
+                    exchange = await setup.get_exchange(f"{prefix}.requests.exchange", ensure=False)
+                    await queue.unbind(exchange, routing_key=f"{prefix}.requests")
+                    async with Client(client_transport, mode="2026-07-28") as client:
+                        try:
+                            await client.list_tools()
+                        except MCPError as exc:
+                            assert exc.code == CONNECTION_CLOSED
+                        else:
+                            raise AssertionError("An unroutable publish must fail immediately")
+                else:
+                    entered = anyio.Event()
+                    finished = anyio.Event()
+                    server = MCPServer("AMQP peer loss")
+
+                    @server.tool()
+                    async def hold() -> str:
+                        entered.set()
+                        await anyio.sleep_forever()
+                        raise NotImplementedError
+
+                    runtime = await stack.enter_async_context(server.serve())
+                    await runtime.connect(server_transport)
+                    # Remote-idle loss is not signalled by AMQP; the deadline is the behavior under test.
+                    async with Client(client_transport, mode="2026-07-28", read_timeout_seconds=1) as client:
+
+                        async def call() -> None:
+                            try:
+                                await client.call_tool("hold")
+                            except MCPError as exc:
+                                assert exc.code == REQUEST_TIMEOUT
+                            else:
+                                raise AssertionError("The remote-loss deadline must settle the request")
+                            finished.set()
+
+                        async with anyio.create_task_group() as tg:
+                            tg.start_soon(call)
+                            await entered.wait()
+                            assert not finished.is_set()
+                            await server_connection.close()
+                            await finished.wait()
+                assert not client_channel.is_closed
+                await client_channel.set_qos(prefetch_count=16)
+                if scenario != "peer-loss":
+                    assert not server_channel.is_closed
+                    await server_channel.set_qos(prefetch_count=16)
+
+
+if __name__ == "__main__":
+    anyio.run(main)