From 526ae912be763df06a273b1ef79a52eda9efb502 Mon Sep 17 00:00:00 2001
From: Will James via Codex <will.james+codex@dbtlabs.com>
Date: Mon, 28 Sep 2026 12:35:48 -0500
Subject: [PATCH 1/6] fix: retain stateless task cleanup on MCP SDK 1.30

Reported-by: Will James
Github-Issue: #1764
---
 src/mcp/server/streamable_http_manager.py    | 28 ++++++--
 tests/server/test_streamable_http_manager.py | 74 ++++++++++++++++++++
 2 files changed, 95 insertions(+), 7 deletions(-)

diff --git a/src/mcp/server/streamable_http_manager.py b/src/mcp/server/streamable_http_manager.py
index eb36296e8f..b26636bc8a 100644
--- a/src/mcp/server/streamable_http_manager.py
+++ b/src/mcp/server/streamable_http_manager.py
@@ -121,6 +121,7 @@ def __init__(
         # Session tracking (only used if not stateless)
         self._session_creation_lock = anyio.Lock()
         self._server_instances: dict[str, StreamableHTTPServerTransport] = {}
+        self._stateless_transports: set[StreamableHTTPServerTransport] = set()
         # Identity of the credential that created each session; requests for a
         # session must present the same credential.
         self._session_owners: dict[str, AuthorizationContext] = {}
@@ -166,11 +167,19 @@ async def lifespan(app: Starlette) -> AsyncIterator[None]:
                 yield  # Let the application run
             finally:
                 logger.info("StreamableHTTP session manager shutting down")
+                # Close in-flight streams before cancelling the manager. This
+                # lets SSE responses send their final HTTP body chunk.
+                for transport in (*self._server_instances.values(), *self._stateless_transports):
+                    try:
+                        await transport.terminate()
+                    except Exception:  # pragma: no cover
+                        logger.debug("Error terminating transport during shutdown", exc_info=True)
                 # Cancel task group to stop all spawned tasks
                 tg.cancel_scope.cancel()
                 self._task_group = None
                 # Clear any remaining server instances
                 self._server_instances.clear()
+                self._stateless_transports.clear()
                 self._session_owners.clear()
 
     async def handle_request(
@@ -244,15 +253,20 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA
                 except Exception:  # pragma: no cover
                     logger.exception("Stateless session crashed")
 
-        # The per-request server task only ends once the transport is
-        # terminated, so terminate it even if the request was cancelled.
-        assert self._task_group is not None
+        self._stateless_transports.add(http_transport)
         try:
-            await self._task_group.start(run_stateless_server)
-            await http_transport.handle_request(scope, receive, send)
+            # A request-scoped group ensures its server task cannot outlive a
+            # disconnected client and accumulate in the manager's task group.
+            async with anyio.create_task_group() as request_tg:
+                try:
+                    await request_tg.start(run_stateless_server)
+                    await http_transport.handle_request(scope, receive, send)
+                finally:
+                    with anyio.CancelScope(shield=True):
+                        await http_transport.terminate()
+                    request_tg.cancel_scope.cancel()
         finally:
-            with anyio.CancelScope(shield=True):
-                await http_transport.terminate()
+            self._stateless_transports.discard(http_transport)
 
     async def _handle_stateful_request(
         self,
diff --git a/tests/server/test_streamable_http_manager.py b/tests/server/test_streamable_http_manager.py
index 9ecfd2764b..b1cadd677a 100644
--- a/tests/server/test_streamable_http_manager.py
+++ b/tests/server/test_streamable_http_manager.py
@@ -351,6 +351,80 @@ async def mock_receive():
             assert len(transport._request_streams) == 0, "Transport should have no active request streams"
 
 
+@pytest.mark.anyio
+async def test_stateless_request_cancels_server_task_when_request_ends() -> None:
+    """A completed stateless request must not leave its server task running."""
+    app = Server("test-stateless-task-cleanup")
+    manager = StreamableHTTPSessionManager(app=app, stateless=True)
+    server_started = anyio.Event()
+    server_finished = anyio.Event()
+    sent_messages: list[Message] = []
+
+    async def blocked_server(*args: object, **kwargs: object) -> None:
+        server_started.set()
+        try:
+            await anyio.sleep_forever()
+        finally:
+            server_finished.set()
+
+    async def receive() -> Message:
+        return {"type": "http.request", "body": b"", "more_body": False}
+
+    async def send(message: Message) -> None:
+        sent_messages.append(message)
+
+    scope: Scope = {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp",
+        "headers": [(b"content-type", b"application/json"), (b"accept", b"application/json, text/event-stream")],
+    }
+    with patch.object(app, "run", blocked_server):
+        async with manager.run():
+            with anyio.fail_after(5):
+                await manager.handle_request(scope, receive, send)
+                await server_started.wait()
+            assert server_finished.is_set()
+
+    assert any(message["type"] == "http.response.start" for message in sent_messages)
+
+
+@pytest.mark.anyio
+async def test_stateless_stream_closes_cleanly_on_shutdown() -> None:
+    """Shutdown should send the final body chunk for an active SSE stream."""
+    manager = StreamableHTTPSessionManager(app=Server("test-stateless-shutdown"), stateless=True)
+    stream_started = anyio.Event()
+    stream_closed = anyio.Event()
+    request_received = False
+
+    async def receive() -> Message:
+        nonlocal request_received
+        if not request_received:
+            request_received = True
+            return {"type": "http.request", "body": b"", "more_body": False}
+        await anyio.sleep_forever()
+        raise AssertionError("unreachable")  # pragma: no cover
+
+    async def send(message: Message) -> None:
+        if message["type"] == "http.response.start":
+            stream_started.set()
+        if message["type"] == "http.response.body" and not message.get("more_body", False):
+            stream_closed.set()
+
+    scope: Scope = {
+        "type": "http",
+        "method": "GET",
+        "path": "/mcp",
+        "headers": [(b"accept", b"text/event-stream")],
+    }
+    with anyio.fail_after(5):
+        async with anyio.create_task_group() as requests:
+            async with manager.run():
+                requests.start_soon(manager.handle_request, scope, receive, send)
+                await stream_started.wait()
+            await stream_closed.wait()
+
+
 @pytest.mark.anyio
 async def test_unknown_session_id_returns_404():
     """Test that requests with unknown session IDs return HTTP 404 per MCP spec."""

From 49ac11436e8231a0063b6b0f428ccb501853fa62 Mon Sep 17 00:00:00 2001
From: Will James via Codex <will.james+codex@dbtlabs.com>
Date: Mon, 28 Sep 2026 13:45:28 -0500
Subject: [PATCH 2/6] chore: derive dbt patch version from Git tag

Reported-by: Will James
Github-Issue: #1764
---
 pyproject.toml | 1 +
 1 file changed, 1 insertion(+)

diff --git a/pyproject.toml b/pyproject.toml
index 6bee7f4923..32a4de71cd 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -88,6 +88,7 @@ source = "uv-dynamic-versioning"
 vcs = "git"
 style = "pep440"
 bump = true
+tagged-metadata = true
 
 [project.urls]
 Homepage = "https://modelcontextprotocol.io"

From 8d341c4efc410b2717ca5942fd9d85fa273b61a0 Mon Sep 17 00:00:00 2001
From: Will James via Codex <will.james+codex@dbtlabs.com>
Date: Mon, 28 Sep 2026 16:46:08 -0500
Subject: [PATCH 3/6] test: reserve Streamable HTTP server sockets across
 workers

Reported-by: Will James
---
 tests/shared/test_streamable_http.py | 70 +++++++++++++++-------------
 1 file changed, 37 insertions(+), 33 deletions(-)

diff --git a/tests/shared/test_streamable_http.py b/tests/shared/test_streamable_http.py
index 35376bc1c9..82512cb118 100644
--- a/tests/shared/test_streamable_http.py
+++ b/tests/shared/test_streamable_http.py
@@ -434,7 +434,7 @@ def create_app(
 
 
 def run_server(
-    port: int,
+    server_socket: socket.socket,
     is_json_response_enabled: bool = False,
     event_store: EventStore | None = None,
     retry_interval: int | None = None,
@@ -442,7 +442,7 @@ def run_server(
     """Run the test server.
 
     Args:
-        port: Port to listen on.
+        server_socket: Bound socket passed from the test process.
         is_json_response_enabled: If True, use JSON responses instead of SSE streams.
         event_store: Optional event store for testing resumability.
         retry_interval: Retry interval in milliseconds for SSE polling.
@@ -453,7 +453,7 @@ def run_server(
     config = uvicorn.Config(
         app=app,
         host="127.0.0.1",
-        port=port,
+        port=server_socket.getsockname()[1],
         log_level="info",
         limit_concurrency=10,
         timeout_keep_alive=5,
@@ -465,7 +465,7 @@ def run_server(
 
     # This is important to catch exceptions and prevent test hangs
     try:
-        server.run()
+        server.run(sockets=[server_socket])
     except Exception:
         import traceback
 
@@ -474,29 +474,30 @@ def run_server(
 
 # Test fixtures - using same approach as SSE tests
 @pytest.fixture
-def basic_server_port() -> int:
-    """Find an available port for the basic server."""
+def basic_server_socket() -> Generator[socket.socket, None, None]:
+    """Reserve the basic server's port until the server starts."""
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
 
 
 @pytest.fixture
-def json_server_port() -> int:
-    """Find an available port for the JSON response server."""
+def json_server_socket() -> Generator[socket.socket, None, None]:
+    """Reserve the JSON response server's port until the server starts."""
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
 
 
 @pytest.fixture
-def basic_server(basic_server_port: int) -> Generator[None, None, None]:
+def basic_server(basic_server_socket: socket.socket) -> Generator[None, None, None]:
     """Start a basic server."""
-    proc = multiprocessing.Process(target=run_server, kwargs={"port": basic_server_port}, daemon=True)
+    port = basic_server_socket.getsockname()[1]
+    proc = multiprocessing.Process(target=run_server, kwargs={"server_socket": basic_server_socket}, daemon=True)
     proc.start()
 
     # Wait for server to be running
-    wait_for_server(basic_server_port)
+    wait_for_server(port)
 
     yield
 
@@ -512,29 +513,30 @@ def event_store() -> SimpleEventStore:
 
 
 @pytest.fixture
-def event_server_port() -> int:
-    """Find an available port for the event store server."""
+def event_server_socket() -> Generator[socket.socket, None, None]:
+    """Reserve the event store server's port until the server starts."""
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
 
 
 @pytest.fixture
 def event_server(
-    event_server_port: int, event_store: SimpleEventStore
+    event_server_socket: socket.socket, event_store: SimpleEventStore
 ) -> Generator[tuple[SimpleEventStore, str], None, None]:
     """Start a server with event store and retry_interval enabled."""
+    port = event_server_socket.getsockname()[1]
     proc = multiprocessing.Process(
         target=run_server,
-        kwargs={"port": event_server_port, "event_store": event_store, "retry_interval": 500},
+        kwargs={"event_store": event_store, "retry_interval": 500, "server_socket": event_server_socket},
         daemon=True,
     )
     proc.start()
 
     # Wait for server to be running
-    wait_for_server(event_server_port)
+    wait_for_server(port)
 
-    yield event_store, f"http://127.0.0.1:{event_server_port}"
+    yield event_store, f"http://127.0.0.1:{port}"
 
     # Clean up
     proc.kill()
@@ -542,17 +544,18 @@ def event_server(
 
 
 @pytest.fixture
-def json_response_server(json_server_port: int) -> Generator[None, None, None]:
+def json_response_server(json_server_socket: socket.socket) -> Generator[None, None, None]:
     """Start a server with JSON response enabled."""
+    port = json_server_socket.getsockname()[1]
     proc = multiprocessing.Process(
         target=run_server,
-        kwargs={"port": json_server_port, "is_json_response_enabled": True},
+        kwargs={"is_json_response_enabled": True, "server_socket": json_server_socket},
         daemon=True,
     )
     proc.start()
 
     # Wait for server to be running
-    wait_for_server(json_server_port)
+    wait_for_server(port)
 
     yield
 
@@ -562,15 +565,15 @@ def json_response_server(json_server_port: int) -> Generator[None, None, None]:
 
 
 @pytest.fixture
-def basic_server_url(basic_server_port: int) -> str:
+def basic_server_url(basic_server_socket: socket.socket) -> str:
     """Get the URL for the basic test server."""
-    return f"http://127.0.0.1:{basic_server_port}"
+    return f"http://127.0.0.1:{basic_server_socket.getsockname()[1]}"
 
 
 @pytest.fixture
-def json_server_url(json_server_port: int) -> str:
+def json_server_url(json_server_socket: socket.socket) -> str:
     """Get the URL for the JSON response test server."""
-    return f"http://127.0.0.1:{json_server_port}"
+    return f"http://127.0.0.1:{json_server_socket.getsockname()[1]}"
 
 
 # Basic request validation tests
@@ -1560,7 +1563,7 @@ async def handle_call_tool(name: str, args: dict[str, Any]) -> list[TextContent]
 
 
 # Server runner for context-aware testing
-def run_context_aware_server(port: int):  # pragma: no cover
+def run_context_aware_server(server_socket: socket.socket):  # pragma: no cover
     """Run the context-aware test server."""
     server = ContextAwareServerTest()
 
@@ -1582,21 +1585,22 @@ def run_context_aware_server(port: int):  # pragma: no cover
         config=uvicorn.Config(
             app=app,
             host="127.0.0.1",
-            port=port,
+            port=server_socket.getsockname()[1],
             log_level="error",
         )
     )
-    server_instance.run()
+    server_instance.run(sockets=[server_socket])
 
 
 @pytest.fixture
-def context_aware_server(basic_server_port: int) -> Generator[None, None, None]:
+def context_aware_server(basic_server_socket: socket.socket) -> Generator[None, None, None]:
     """Start the context-aware server in a separate process."""
-    proc = multiprocessing.Process(target=run_context_aware_server, args=(basic_server_port,), daemon=True)
+    port = basic_server_socket.getsockname()[1]
+    proc = multiprocessing.Process(target=run_context_aware_server, args=(basic_server_socket,), daemon=True)
     proc.start()
 
     # Wait for server to be running
-    wait_for_server(basic_server_port)
+    wait_for_server(port)
 
     yield
 

From ddb86ae96a95d356ca08d581b01e2c06fcc07f86 Mon Sep 17 00:00:00 2001
From: Will James via Codex <will.james+codex@dbtlabs.com>
Date: Mon, 28 Sep 2026 16:56:17 -0500
Subject: [PATCH 4/6] test: reserve ports across SDK server fixtures

Reported-by: Will James
---
 tests/client/test_http_unicode.py             | 25 +++++---
 tests/client/test_notification_response.py    | 23 +++++---
 tests/server/fastmcp/test_integration.py      | 29 +++++++---
 tests/server/test_sse_security.py             | 57 +++++++++++--------
 tests/server/test_streamable_http_security.py | 54 +++++++++++-------
 tests/server/test_websocket_security.py       | 44 ++++++++------
 tests/shared/test_sse.py                      | 36 +++++++-----
 tests/shared/test_ws.py                       | 18 ++++--
 8 files changed, 179 insertions(+), 107 deletions(-)

diff --git a/tests/client/test_http_unicode.py b/tests/client/test_http_unicode.py
index ec38f35838..88ca5f5254 100644
--- a/tests/client/test_http_unicode.py
+++ b/tests/client/test_http_unicode.py
@@ -35,7 +35,7 @@
 }
 
 
-def run_unicode_server(port: int) -> None:  # pragma: no cover
+def run_unicode_server(server_socket: socket.socket) -> None:  # pragma: no cover
     """Run the Unicode test server in a separate process."""
     # Import inside the function since this runs in a separate process
     from collections.abc import AsyncGenerator
@@ -137,25 +137,34 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
     config = uvicorn.Config(
         app=app,
         host="127.0.0.1",
-        port=port,
+        port=server_socket.getsockname()[1],
         log_level="error",
     )
     uvicorn_server = uvicorn.Server(config)
-    uvicorn_server.run()
+    uvicorn_server.run(sockets=[server_socket])
 
 
 @pytest.fixture
-def unicode_server_port() -> int:
-    """Find an available port for the Unicode test server."""
+def unicode_server_socket() -> Generator[socket.socket, None, None]:
+    """Reserve the Unicode test server's port until it starts."""
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
 
 
 @pytest.fixture
-def running_unicode_server(unicode_server_port: int) -> Generator[str, None, None]:
+def unicode_server_port(unicode_server_socket: socket.socket) -> int:
+    return unicode_server_socket.getsockname()[1]
+
+
+@pytest.fixture
+def running_unicode_server(
+    unicode_server_port: int, unicode_server_socket: socket.socket
+) -> Generator[str, None, None]:
     """Start a Unicode test server in a separate process."""
-    proc = multiprocessing.Process(target=run_unicode_server, kwargs={"port": unicode_server_port}, daemon=True)
+    proc = multiprocessing.Process(
+        target=run_unicode_server, kwargs={"server_socket": unicode_server_socket}, daemon=True
+    )
     proc.start()
 
     # Wait for server to be ready
diff --git a/tests/client/test_notification_response.py b/tests/client/test_notification_response.py
index 7500abee73..58b96eb61f 100644
--- a/tests/client/test_notification_response.py
+++ b/tests/client/test_notification_response.py
@@ -67,31 +67,38 @@ async def handle_mcp_request(request: Request) -> Response:
     return app
 
 
-def run_non_sdk_server(port: int) -> None:  # pragma: no cover
+def run_non_sdk_server(server_socket: socket.socket) -> None:  # pragma: no cover
     """Run the non-SDK server in a separate process."""
     app = create_non_sdk_server_app()
     config = uvicorn.Config(
         app=app,
         host="127.0.0.1",
-        port=port,
+        port=server_socket.getsockname()[1],
         log_level="error",  # Reduce noise in tests
     )
     server = uvicorn.Server(config=config)
-    server.run()
+    server.run(sockets=[server_socket])
 
 
 @pytest.fixture
-def non_sdk_server_port() -> int:
-    """Get an available port for the test server."""
+def non_sdk_server_socket() -> Generator[socket.socket, None, None]:
+    """Reserve the test server's port until it starts."""
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
 
 
 @pytest.fixture
-def non_sdk_server(non_sdk_server_port: int) -> Generator[None, None, None]:
+def non_sdk_server_port(non_sdk_server_socket: socket.socket) -> int:
+    return non_sdk_server_socket.getsockname()[1]
+
+
+@pytest.fixture
+def non_sdk_server(non_sdk_server_port: int, non_sdk_server_socket: socket.socket) -> Generator[None, None, None]:
     """Start a non-SDK server for testing."""
-    proc = multiprocessing.Process(target=run_non_sdk_server, kwargs={"port": non_sdk_server_port}, daemon=True)
+    proc = multiprocessing.Process(
+        target=run_non_sdk_server, kwargs={"server_socket": non_sdk_server_socket}, daemon=True
+    )
     proc.start()
 
     # Wait for server to be ready
diff --git a/tests/server/fastmcp/test_integration.py b/tests/server/fastmcp/test_integration.py
index 70948bd7e2..e57e49cb56 100644
--- a/tests/server/fastmcp/test_integration.py
+++ b/tests/server/fastmcp/test_integration.py
@@ -88,11 +88,16 @@ async def handle_generic_notification(
 
 # Common fixtures
 @pytest.fixture
-def server_port() -> int:
-    """Get a free port for testing."""
+def server_socket() -> Generator[socket.socket, None, None]:
+    """Reserve a port for the test server until it starts."""
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
+
+
+@pytest.fixture
+def server_port(server_socket: socket.socket) -> int:
+    return server_socket.getsockname()[1]
 
 
 @pytest.fixture
@@ -101,7 +106,9 @@ def server_url(server_port: int) -> str:
     return f"http://127.0.0.1:{server_port}"
 
 
-def run_server_with_transport(module_name: str, port: int, transport: str) -> None:  # pragma: no cover
+def run_server_with_transport(
+    module_name: str, server_socket: socket.socket, transport: str
+) -> None:  # pragma: no cover
     """Run server with specified transport."""
     # Get the MCP instance based on module name
     if module_name == "basic_tool":
@@ -135,13 +142,17 @@ def run_server_with_transport(module_name: str, port: int, transport: str) -> No
     else:
         raise ValueError(f"Invalid transport for test server: {transport}")
 
-    server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=port, log_level="error"))
-    print(f"Starting {transport} server on port {port}")
-    server.run()
+    server = uvicorn.Server(
+        config=uvicorn.Config(app=app, host="127.0.0.1", port=server_socket.getsockname()[1], log_level="error")
+    )
+    print(f"Starting {transport} server on port {server_socket.getsockname()[1]}")
+    server.run(sockets=[server_socket])
 
 
 @pytest.fixture
-def server_transport(request: pytest.FixtureRequest, server_port: int) -> Generator[str, None, None]:
+def server_transport(
+    request: pytest.FixtureRequest, server_port: int, server_socket: socket.socket
+) -> Generator[str, None, None]:
     """Start server in a separate process with specified MCP instance and transport.
 
     Args:
@@ -155,7 +166,7 @@ def server_transport(request: pytest.FixtureRequest, server_port: int) -> Genera
 
     proc = multiprocessing.Process(
         target=run_server_with_transport,
-        args=(module_name, server_port, transport),
+        args=(module_name, server_socket, transport),
         daemon=True,
     )
     proc.start()
diff --git a/tests/server/test_sse_security.py b/tests/server/test_sse_security.py
index f262d6b473..2e46a4384c 100644
--- a/tests/server/test_sse_security.py
+++ b/tests/server/test_sse_security.py
@@ -4,6 +4,7 @@
 import multiprocessing
 import re
 import socket
+from collections.abc import Generator
 from typing import Any
 
 import anyio
@@ -29,10 +30,15 @@
 
 
 @pytest.fixture
-def server_port() -> int:
+def server_socket() -> Generator[socket.socket, None, None]:
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
+
+
+@pytest.fixture
+def server_port(server_socket: socket.socket) -> int:
+    return server_socket.getsockname()[1]
 
 
 @pytest.fixture
@@ -48,7 +54,9 @@ async def on_list_tools(self) -> list[Tool]:
         return []
 
 
-def run_server_with_settings(port: int, security_settings: TransportSecuritySettings | None = None):  # pragma: no cover
+def run_server_with_settings(
+    server_socket: socket.socket, security_settings: TransportSecuritySettings | None = None
+):  # pragma: no cover
     """Run the SSE server with specified security settings."""
     app = SecurityTestServer()
     sse_transport = SseServerTransport("/messages/", security_settings)
@@ -69,22 +77,25 @@ async def handle_sse(request: Request):
     ]
 
     starlette_app = Starlette(routes=routes)
-    uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="error")
+    server = uvicorn.Server(
+        uvicorn.Config(starlette_app, host="127.0.0.1", port=server_socket.getsockname()[1], log_level="error")
+    )
+    server.run(sockets=[server_socket])
 
 
-def start_server_process(port: int, security_settings: TransportSecuritySettings | None = None):
+def start_server_process(server_socket: socket.socket, security_settings: TransportSecuritySettings | None = None):
     """Start server in a separate process."""
-    process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings))
+    process = multiprocessing.Process(target=run_server_with_settings, args=(server_socket, security_settings))
     process.start()
     # Wait for server to be ready to accept connections
-    wait_for_server(port)
+    wait_for_server(server_socket.getsockname()[1])
     return process
 
 
 @pytest.mark.anyio
-async def test_sse_security_default_settings(server_port: int):
+async def test_sse_security_default_settings(server_port: int, server_socket: socket.socket):
     """Test SSE with default security settings (protection disabled)."""
-    process = start_server_process(server_port)
+    process = start_server_process(server_socket)
 
     try:
         headers = {"Host": "evil.com", "Origin": "http://evil.com"}
@@ -98,11 +109,11 @@ async def test_sse_security_default_settings(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_sse_security_invalid_host_header(server_port: int):
+async def test_sse_security_invalid_host_header(server_port: int, server_socket: socket.socket):
     """Test SSE with invalid Host header."""
     # Enable security by providing settings with an empty allowed_hosts list
     security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["example.com"])
-    process = start_server_process(server_port, security_settings)
+    process = start_server_process(server_socket, security_settings)
 
     try:
         # Test with invalid host header
@@ -119,13 +130,13 @@ async def test_sse_security_invalid_host_header(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_sse_security_invalid_origin_header(server_port: int):
+async def test_sse_security_invalid_origin_header(server_port: int, server_socket: socket.socket):
     """Test SSE with invalid Origin header."""
     # Configure security to allow the host but restrict origins
     security_settings = TransportSecuritySettings(
         enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://localhost:*"]
     )
-    process = start_server_process(server_port, security_settings)
+    process = start_server_process(server_socket, security_settings)
 
     try:
         # Test with invalid origin header
@@ -142,13 +153,13 @@ async def test_sse_security_invalid_origin_header(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_sse_security_post_invalid_content_type(server_port: int):
+async def test_sse_security_post_invalid_content_type(server_port: int, server_socket: socket.socket):
     """Test POST endpoint with invalid Content-Type header."""
     # Configure security to allow the host
     security_settings = TransportSecuritySettings(
         enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://127.0.0.1:*"]
     )
-    process = start_server_process(server_port, security_settings)
+    process = start_server_process(server_socket, security_settings)
 
     try:
         async with httpx.AsyncClient(timeout=5.0) as client:
@@ -175,10 +186,10 @@ async def test_sse_security_post_invalid_content_type(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_sse_security_disabled(server_port: int):
+async def test_sse_security_disabled(server_port: int, server_socket: socket.socket):
     """Test SSE with security disabled."""
     settings = TransportSecuritySettings(enable_dns_rebinding_protection=False)
-    process = start_server_process(server_port, settings)
+    process = start_server_process(server_socket, settings)
 
     try:
         # Test with invalid host header - should still work
@@ -196,14 +207,14 @@ async def test_sse_security_disabled(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_sse_security_custom_allowed_hosts(server_port: int):
+async def test_sse_security_custom_allowed_hosts(server_port: int, server_socket: socket.socket):
     """Test SSE with custom allowed hosts."""
     settings = TransportSecuritySettings(
         enable_dns_rebinding_protection=True,
         allowed_hosts=["localhost", "127.0.0.1", "custom.host"],
         allowed_origins=["http://localhost", "http://127.0.0.1", "http://custom.host"],
     )
-    process = start_server_process(server_port, settings)
+    process = start_server_process(server_socket, settings)
 
     try:
         # Test with custom allowed host
@@ -229,14 +240,14 @@ async def test_sse_security_custom_allowed_hosts(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_sse_security_wildcard_ports(server_port: int):
+async def test_sse_security_wildcard_ports(server_port: int, server_socket: socket.socket):
     """Test SSE with wildcard port patterns."""
     settings = TransportSecuritySettings(
         enable_dns_rebinding_protection=True,
         allowed_hosts=["localhost:*", "127.0.0.1:*"],
         allowed_origins=["http://localhost:*", "http://127.0.0.1:*"],
     )
-    process = start_server_process(server_port, settings)
+    process = start_server_process(server_socket, settings)
 
     try:
         # Test with various port numbers
@@ -263,13 +274,13 @@ async def test_sse_security_wildcard_ports(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_sse_security_post_valid_content_type(server_port: int):
+async def test_sse_security_post_valid_content_type(server_port: int, server_socket: socket.socket):
     """Test POST endpoint with valid Content-Type headers."""
     # Configure security to allow the host
     security_settings = TransportSecuritySettings(
         enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://127.0.0.1:*"]
     )
-    process = start_server_process(server_port, security_settings)
+    process = start_server_process(server_socket, security_settings)
 
     try:
         async with httpx.AsyncClient() as client:
diff --git a/tests/server/test_streamable_http_security.py b/tests/server/test_streamable_http_security.py
index a637b1dce0..a99a4f524e 100644
--- a/tests/server/test_streamable_http_security.py
+++ b/tests/server/test_streamable_http_security.py
@@ -3,7 +3,7 @@
 import logging
 import multiprocessing
 import socket
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, Generator
 from contextlib import asynccontextmanager
 
 import httpx
@@ -24,10 +24,15 @@
 
 
 @pytest.fixture
-def server_port() -> int:
+def server_socket() -> Generator[socket.socket, None, None]:
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
+
+
+@pytest.fixture
+def server_port(server_socket: socket.socket) -> int:
+    return server_socket.getsockname()[1]
 
 
 @pytest.fixture
@@ -43,7 +48,9 @@ async def on_list_tools(self) -> list[Tool]:
         return []
 
 
-def run_server_with_settings(port: int, security_settings: TransportSecuritySettings | None = None):  # pragma: no cover
+def run_server_with_settings(
+    server_socket: socket.socket, security_settings: TransportSecuritySettings | None = None
+):  # pragma: no cover
     """Run the StreamableHTTP server with specified security settings."""
     app = SecurityTestServer()
 
@@ -70,22 +77,25 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
     ]
 
     starlette_app = Starlette(routes=routes, lifespan=lifespan)
-    uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="error")
+    server = uvicorn.Server(
+        uvicorn.Config(starlette_app, host="127.0.0.1", port=server_socket.getsockname()[1], log_level="error")
+    )
+    server.run(sockets=[server_socket])
 
 
-def start_server_process(port: int, security_settings: TransportSecuritySettings | None = None):
+def start_server_process(server_socket: socket.socket, security_settings: TransportSecuritySettings | None = None):
     """Start server in a separate process."""
-    process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings))
+    process = multiprocessing.Process(target=run_server_with_settings, args=(server_socket, security_settings))
     process.start()
     # Wait for server to be ready to accept connections
-    wait_for_server(port)
+    wait_for_server(server_socket.getsockname()[1])
     return process
 
 
 @pytest.mark.anyio
-async def test_streamable_http_security_default_settings(server_port: int):
+async def test_streamable_http_security_default_settings(server_port: int, server_socket: socket.socket):
     """Test StreamableHTTP with default security settings (protection enabled)."""
-    process = start_server_process(server_port)
+    process = start_server_process(server_socket)
 
     try:
         # Test with valid localhost headers
@@ -108,10 +118,10 @@ async def test_streamable_http_security_default_settings(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_streamable_http_security_invalid_host_header(server_port: int):
+async def test_streamable_http_security_invalid_host_header(server_port: int, server_socket: socket.socket):
     """Test StreamableHTTP with invalid Host header."""
     security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True)
-    process = start_server_process(server_port, security_settings)
+    process = start_server_process(server_socket, security_settings)
 
     try:
         # Test with invalid host header
@@ -136,10 +146,10 @@ async def test_streamable_http_security_invalid_host_header(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_streamable_http_security_invalid_origin_header(server_port: int):
+async def test_streamable_http_security_invalid_origin_header(server_port: int, server_socket: socket.socket):
     """Test StreamableHTTP with invalid Origin header."""
     security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"])
-    process = start_server_process(server_port, security_settings)
+    process = start_server_process(server_socket, security_settings)
 
     try:
         # Test with invalid origin header
@@ -164,9 +174,9 @@ async def test_streamable_http_security_invalid_origin_header(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_streamable_http_security_invalid_content_type(server_port: int):
+async def test_streamable_http_security_invalid_content_type(server_port: int, server_socket: socket.socket):
     """Test StreamableHTTP POST with invalid Content-Type header."""
-    process = start_server_process(server_port)
+    process = start_server_process(server_socket)
 
     try:
         async with httpx.AsyncClient(timeout=5.0) as client:
@@ -197,10 +207,10 @@ async def test_streamable_http_security_invalid_content_type(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_streamable_http_security_disabled(server_port: int):
+async def test_streamable_http_security_disabled(server_port: int, server_socket: socket.socket):
     """Test StreamableHTTP with security disabled."""
     settings = TransportSecuritySettings(enable_dns_rebinding_protection=False)
-    process = start_server_process(server_port, settings)
+    process = start_server_process(server_socket, settings)
 
     try:
         # Test with invalid host header - should still work
@@ -225,14 +235,14 @@ async def test_streamable_http_security_disabled(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_streamable_http_security_custom_allowed_hosts(server_port: int):
+async def test_streamable_http_security_custom_allowed_hosts(server_port: int, server_socket: socket.socket):
     """Test StreamableHTTP with custom allowed hosts."""
     settings = TransportSecuritySettings(
         enable_dns_rebinding_protection=True,
         allowed_hosts=["localhost", "127.0.0.1", "custom.host"],
         allowed_origins=["http://localhost", "http://127.0.0.1", "http://custom.host"],
     )
-    process = start_server_process(server_port, settings)
+    process = start_server_process(server_socket, settings)
 
     try:
         # Test with custom allowed host
@@ -256,10 +266,10 @@ async def test_streamable_http_security_custom_allowed_hosts(server_port: int):
 
 
 @pytest.mark.anyio
-async def test_streamable_http_security_get_request(server_port: int):
+async def test_streamable_http_security_get_request(server_port: int, server_socket: socket.socket):
     """Test StreamableHTTP GET request with security."""
     security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1"])
-    process = start_server_process(server_port, security_settings)
+    process = start_server_process(server_socket, security_settings)
 
     try:
         # Test GET request with invalid host header
diff --git a/tests/server/test_websocket_security.py b/tests/server/test_websocket_security.py
index 35f778080d..bbdf7c9a5c 100644
--- a/tests/server/test_websocket_security.py
+++ b/tests/server/test_websocket_security.py
@@ -6,6 +6,7 @@
 import multiprocessing
 import socket
 import warnings
+from collections.abc import Generator
 
 import pytest
 import uvicorn
@@ -32,13 +33,20 @@
 
 
 @pytest.fixture
-def server_port() -> int:
+def server_socket() -> Generator[socket.socket, None, None]:
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
 
 
-def run_server_with_settings(port: int, security_settings: TransportSecuritySettings | None = None):  # pragma: no cover
+@pytest.fixture
+def server_port(server_socket: socket.socket) -> int:
+    return server_socket.getsockname()[1]
+
+
+def run_server_with_settings(
+    server_socket: socket.socket, security_settings: TransportSecuritySettings | None = None
+):  # pragma: no cover
     """Run a WebSocket MCP server with the given security settings."""
     warnings.filterwarnings("ignore", category=DeprecationWarning)
     server = Server(SERVER_NAME)
@@ -53,21 +61,23 @@ async def handle_ws(websocket: WebSocket) -> None:
             logger.debug(f"WebSocket connection failed validation: {exc}")
 
     app = Starlette(routes=[WebSocketRoute("/ws", endpoint=handle_ws)])
-    uvicorn.run(app, host="127.0.0.1", port=port, log_level="error")
+    uvicorn.Server(uvicorn.Config(app, host="127.0.0.1", port=server_socket.getsockname()[1], log_level="error")).run(
+        sockets=[server_socket]
+    )
 
 
-def start_server_process(port: int, security_settings: TransportSecuritySettings | None = None):
+def start_server_process(server_socket: socket.socket, security_settings: TransportSecuritySettings | None = None):
     """Start the server in a subprocess and wait until it accepts connections."""
-    process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings))
+    process = multiprocessing.Process(target=run_server_with_settings, args=(server_socket, security_settings))
     process.start()
-    wait_for_server(port)
+    wait_for_server(server_socket.getsockname()[1])
     return process
 
 
 @pytest.mark.anyio
-async def test_ws_security_default_settings(server_port: int) -> None:
+async def test_ws_security_default_settings(server_port: int, server_socket: socket.socket) -> None:
     """With no security settings the WebSocket transport accepts any Origin (matches SSE/StreamableHTTP default)."""
-    process = start_server_process(server_port)
+    process = start_server_process(server_socket)
     try:
         async with connect(
             f"ws://127.0.0.1:{server_port}/ws",
@@ -81,12 +91,12 @@ async def test_ws_security_default_settings(server_port: int) -> None:
 
 
 @pytest.mark.anyio
-async def test_ws_security_invalid_origin_header(server_port: int) -> None:
+async def test_ws_security_invalid_origin_header(server_port: int, server_socket: socket.socket) -> None:
     """An Origin not in allowed_origins is rejected before the handshake completes."""
     settings = TransportSecuritySettings(
         enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://localhost:*"]
     )
-    process = start_server_process(server_port, settings)
+    process = start_server_process(server_socket, settings)
     try:
         with pytest.raises(InvalidStatus) as exc_info:
             async with connect(
@@ -102,10 +112,10 @@ async def test_ws_security_invalid_origin_header(server_port: int) -> None:
 
 
 @pytest.mark.anyio
-async def test_ws_security_invalid_host_header(server_port: int) -> None:
+async def test_ws_security_invalid_host_header(server_port: int, server_socket: socket.socket) -> None:
     """A Host not in allowed_hosts is rejected before the handshake completes."""
     settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["example.com"])
-    process = start_server_process(server_port, settings)
+    process = start_server_process(server_socket, settings)
     try:
         with pytest.raises(InvalidStatus) as exc_info:
             async with connect(f"ws://127.0.0.1:{server_port}/ws", subprotocols=[Subprotocol("mcp")]):
@@ -117,12 +127,12 @@ async def test_ws_security_invalid_host_header(server_port: int) -> None:
 
 
 @pytest.mark.anyio
-async def test_ws_security_allowed_origin(server_port: int) -> None:
+async def test_ws_security_allowed_origin(server_port: int, server_socket: socket.socket) -> None:
     """An Origin matching allowed_origins is accepted."""
     settings = TransportSecuritySettings(
         enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"], allowed_origins=["http://localhost:*"]
     )
-    process = start_server_process(server_port, settings)
+    process = start_server_process(server_socket, settings)
     try:
         async with connect(
             f"ws://127.0.0.1:{server_port}/ws",
@@ -136,10 +146,10 @@ async def test_ws_security_allowed_origin(server_port: int) -> None:
 
 
 @pytest.mark.anyio
-async def test_ws_security_disabled(server_port: int) -> None:
+async def test_ws_security_disabled(server_port: int, server_socket: socket.socket) -> None:
     """Explicitly disabling protection accepts any Origin."""
     settings = TransportSecuritySettings(enable_dns_rebinding_protection=False)
-    process = start_server_process(server_port, settings)
+    process = start_server_process(server_socket, settings)
     try:
         async with connect(
             f"ws://127.0.0.1:{server_port}/ws",
diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py
index 77d6cac65f..ef1015cc60 100644
--- a/tests/shared/test_sse.py
+++ b/tests/shared/test_sse.py
@@ -44,10 +44,15 @@
 
 
 @pytest.fixture
-def server_port() -> int:
+def server_socket() -> Generator[socket.socket, None, None]:
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
+
+
+@pytest.fixture
+def server_port(server_socket: socket.socket) -> int:
+    return server_socket.getsockname()[1]
 
 
 @pytest.fixture
@@ -111,11 +116,12 @@ async def handle_sse(request: Request) -> Response:
     return app
 
 
-def run_server(server_port: int) -> None:  # pragma: no cover
+def run_server(server_socket: socket.socket) -> None:  # pragma: no cover
     app = make_server_app()
+    server_port = server_socket.getsockname()[1]
     server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=server_port, log_level="error"))
     print(f"starting server on {server_port}")
-    server.run()
+    server.run(sockets=[server_socket])
 
     # Give server time to start
     while not server.started:
@@ -124,8 +130,8 @@ def run_server(server_port: int) -> None:  # pragma: no cover
 
 
 @pytest.fixture()
-def server(server_port: int) -> Generator[None, None, None]:
-    proc = multiprocessing.Process(target=run_server, kwargs={"server_port": server_port}, daemon=True)
+def server(server_port: int, server_socket: socket.socket) -> Generator[None, None, None]:
+    proc = multiprocessing.Process(target=run_server, kwargs={"server_socket": server_socket}, daemon=True)
     print("starting process")
     proc.start()
 
@@ -289,12 +295,13 @@ async def test_sse_client_timeout(  # pragma: no cover
     pytest.fail("the client should have timed out and returned an error already")
 
 
-def run_mounted_server(server_port: int) -> None:  # pragma: no cover
+def run_mounted_server(server_socket: socket.socket) -> None:  # pragma: no cover
     app = make_server_app()
     main_app = Starlette(routes=[Mount("/mounted_app", app=app)])
+    server_port = server_socket.getsockname()[1]
     server = uvicorn.Server(config=uvicorn.Config(app=main_app, host="127.0.0.1", port=server_port, log_level="error"))
     print(f"starting server on {server_port}")
-    server.run()
+    server.run(sockets=[server_socket])
 
     # Give server time to start
     while not server.started:
@@ -303,8 +310,8 @@ def run_mounted_server(server_port: int) -> None:  # pragma: no cover
 
 
 @pytest.fixture()
-def mounted_server(server_port: int) -> Generator[None, None, None]:
-    proc = multiprocessing.Process(target=run_mounted_server, kwargs={"server_port": server_port}, daemon=True)
+def mounted_server(server_port: int, server_socket: socket.socket) -> Generator[None, None, None]:
+    proc = multiprocessing.Process(target=run_mounted_server, kwargs={"server_socket": server_socket}, daemon=True)
     print("starting process")
     proc.start()
 
@@ -379,7 +386,7 @@ async def handle_list_tools() -> list[Tool]:
             ]
 
 
-def run_context_server(server_port: int) -> None:  # pragma: no cover
+def run_context_server(server_socket: socket.socket) -> None:  # pragma: no cover
     """Run a server that captures request context"""
     # Configure security with allowed hosts/origins for testing
     security_settings = TransportSecuritySettings(
@@ -400,15 +407,16 @@ async def handle_sse(request: Request) -> Response:
         ]
     )
 
+    server_port = server_socket.getsockname()[1]
     server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=server_port, log_level="error"))
     print(f"starting context server on {server_port}")
-    server.run()
+    server.run(sockets=[server_socket])
 
 
 @pytest.fixture()
-def context_server(server_port: int) -> Generator[None, None, None]:
+def context_server(server_port: int, server_socket: socket.socket) -> Generator[None, None, None]:
     """Fixture that provides a server with request context capture"""
-    proc = multiprocessing.Process(target=run_context_server, kwargs={"server_port": server_port}, daemon=True)
+    proc = multiprocessing.Process(target=run_context_server, kwargs={"server_socket": server_socket}, daemon=True)
     print("starting context server process")
     proc.start()
 
diff --git a/tests/shared/test_ws.py b/tests/shared/test_ws.py
index 64790e2b08..eeefe5ca49 100644
--- a/tests/shared/test_ws.py
+++ b/tests/shared/test_ws.py
@@ -38,10 +38,15 @@
 
 
 @pytest.fixture
-def server_port() -> int:
+def server_socket() -> Generator[socket.socket, None, None]:
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
-        return s.getsockname()[1]
+        yield s
+
+
+@pytest.fixture
+def server_port(server_socket: socket.socket) -> int:
+    return server_socket.getsockname()[1]
 
 
 @pytest.fixture
@@ -98,11 +103,12 @@ async def handle_ws(websocket: WebSocket):
     return app
 
 
-def run_server(server_port: int) -> None:  # pragma: no cover
+def run_server(server_socket: socket.socket) -> None:  # pragma: no cover
     app = make_server_app()
+    server_port = server_socket.getsockname()[1]
     server = uvicorn.Server(config=uvicorn.Config(app=app, host="127.0.0.1", port=server_port, log_level="error"))
     print(f"starting server on {server_port}")
-    server.run()
+    server.run(sockets=[server_socket])
 
     # Give server time to start
     while not server.started:
@@ -111,8 +117,8 @@ def run_server(server_port: int) -> None:  # pragma: no cover
 
 
 @pytest.fixture()  # pragma: no cover
-def server(server_port: int) -> Generator[None, None, None]:
-    proc = multiprocessing.Process(target=run_server, kwargs={"server_port": server_port}, daemon=True)
+def server(server_port: int, server_socket: socket.socket) -> Generator[None, None, None]:
+    proc = multiprocessing.Process(target=run_server, kwargs={"server_socket": server_socket}, daemon=True)
     print("starting process")
     proc.start()
 

From 710d832beceead90fc60b97b128f74f6d19d5d69 Mon Sep 17 00:00:00 2001
From: Will James via Codex <will.james+codex@dbtlabs.com>
Date: Mon, 28 Sep 2026 17:03:27 -0500
Subject: [PATCH 5/6] test: assert SSE context while client is active

Reported-by: Will James
---
 tests/shared/test_sse.py | 13 +++----------
 1 file changed, 3 insertions(+), 10 deletions(-)

diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py
index ef1015cc60..346bedd3c1 100644
--- a/tests/shared/test_sse.py
+++ b/tests/shared/test_sse.py
@@ -469,8 +469,6 @@ async def test_request_context_propagation(context_server: None, server_url: str
 @pytest.mark.anyio
 async def test_request_context_isolation(context_server: None, server_url: str) -> None:
     """Test that request contexts are isolated between different SSE clients."""
-    contexts: list[dict[str, Any]] = []
-
     # Create multiple clients with different headers
     for i in range(3):
         headers = {"X-Request-Id": f"request-{i}", "X-Custom-Value": f"value-{i}"}
@@ -489,14 +487,9 @@ async def test_request_context_isolation(context_server: None, server_url: str)
                 context_data = json.loads(
                     tool_result.content[0].text if tool_result.content[0].type == "text" else "{}"
                 )
-                contexts.append(context_data)
-
-    # Verify each request had its own context
-    assert len(contexts) == 3
-    for i, ctx in enumerate(contexts):
-        assert ctx["request_id"] == f"request-{i}"
-        assert ctx["headers"].get("x-request-id") == f"request-{i}"
-        assert ctx["headers"].get("x-custom-value") == f"value-{i}"
+                assert context_data["request_id"] == f"request-{i}"
+                assert context_data["headers"].get("x-request-id") == f"request-{i}"
+                assert context_data["headers"].get("x-custom-value") == f"value-{i}"
 
 
 def test_sse_message_id_coercion():

From 8f2b084b377e7ace1046512cbfa409b23e24fde2 Mon Sep 17 00:00:00 2001
From: Will James via Codex <will.james+codex@dbtlabs.com>
Date: Mon, 28 Sep 2026 17:10:06 -0500
Subject: [PATCH 6/6] test: exclude unobservable SSE loop exit from coverage

Reported-by: Will James
---
 tests/shared/test_sse.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/shared/test_sse.py b/tests/shared/test_sse.py
index 346bedd3c1..99b6692271 100644
--- a/tests/shared/test_sse.py
+++ b/tests/shared/test_sse.py
@@ -470,7 +470,7 @@ async def test_request_context_propagation(context_server: None, server_url: str
 async def test_request_context_isolation(context_server: None, server_url: str) -> None:
     """Test that request contexts are isolated between different SSE clients."""
     # Create multiple clients with different headers
-    for i in range(3):
+    for i in range(3):  # pragma: no branch - Python 3.11 coverage can miss exit after async SSE cleanup
         headers = {"X-Request-Id": f"request-{i}", "X-Custom-Value": f"value-{i}"}
 
         async with sse_client(server_url + "/sse", headers=headers) as (