Preserve borrowed AMQP channels during peer shutdown

Kludex · Sep 17, 2026 · Sep 17, 2026 · Sep 17, 2026 · Sep 17, 2026 · Sep 17, 2026
commit cdf9c241821b0f22e9305bc00b4a6124e050327c
diff --git a/.github/workflows/shared.yml b/.github/workflows/shared.yml
@@ -125,7 +125,7 @@ jobs:
         run: docker compose -p mcp-sdk-transport-check -f examples/transports/compose.yaml up --wait
       - name: Check live broker routing and peer loss
         run: |
-          for script in demo_amqp.py demo_amqp_permissions.py 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

diff --git a/examples/transports/README.md b/examples/transports/README.md
@@ -113,12 +113,13 @@ docker compose -p mcp-sdk-transport-check -f examples/transports/compose.yaml up
 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. CI runs both programs against the pinned RabbitMQ fixture.
+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. 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.
+`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; the trusted provisioner then deletes them and must reclaim any left by a process crash. Queue expiry does not delete these exchanges. 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
 
@@ -128,21 +129,29 @@ The programs reuse the same two-peer application checks as MQTT for both server
 | 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 | Publisher confirmations; acknowledge before SDK handoff |
-| Close | Empty JSON-typed message body |
+| 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 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 15672).
+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.
diff --git a/examples/transports/compose.yaml b/examples/transports/compose.yaml
@@ -2,7 +2,7 @@ services:
   amqp:
     image: rabbitmq:4.1.8-alpine@sha256:1a087dd3a29b91448407409df70f4f6cb213ac0c269a62861bfe2a665f4ced03
     ports:
-      - "127.0.0.1:${AMQP_TEST_PORT:-15672}:5672"
+      - "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

diff --git a/examples/transports/demo_amqp.py b/examples/transports/demo_amqp.py
@@ -5,6 +5,7 @@
 
 import aio_pika
 import anyio
+from aio_pika.abc import AbstractChannel
 from mcp.shared.transport import Transport
 
 from demo_common import verify
@@ -15,24 +16,16 @@ async def open_transport(stack: AsyncExitStack, principal: str, session: str, se
     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", "15672")),
+        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)
+    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:
-        for direction in ("requests", "responses"):
-            name = f"{queue}.{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=True)
-            await destination.bind(exchange, routing_key=name)
+        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}")
 
@@ -44,5 +37,19 @@ async def main() -> None:
                 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)
diff --git a/examples/transports/demo_amqp_lifecycle.py b/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)
diff --git a/examples/transports/demo_amqp_permissions.py b/examples/transports/demo_amqp_permissions.py
@@ -10,7 +10,7 @@
 
 async def main() -> None:
     """Deny client topology changes and reply injection, including queue/exchange name collisions."""
-    port = int(os.environ.get("AMQP_TEST_PORT", "15672"))
+    port = int(os.environ.get("AMQP_TEST_PORT", "15673"))
     with anyio.fail_after(5):
         async with await aio_pika.connect(
             host="127.0.0.1", port=port, login="server", password="test-server-password"