user [email protected] usertags 1145925 python3.15 tags 1145925 patch thanks
Hi! While rebuilding the python related packages against the Python 3.15rc2 version I ran into this FTBFS [1]. To solve this, I had to backport 3 upstream commits: - Upstream commit ae3338f: Complete the FastMCP Settings model at import time (#3352) [2] - Upstream commit b3025f9: Run transport security tests in process instead of over sockets (#2764) [3] - Upstream commit ddb7b78: Verbatim commit message: Deflake the issue-1363 tests: wait for lifespan startup instead of sleeping (#2879) [4] This should make the tests run successfully. I've applied these fixes in the sandbox [5] to verify that python-mcp builds successfully, these should be applied in Debian to get the package back to a healthy state. Happy hacking, [1]: https://debusine.debian.net/debian/r-python-python3.15/artifact/4516818/ [2]: https://github.com/modelcontextprotocol/python-sdk/commit/ae3338f6b396378217921aeadeca10a68e273c2d [3]: https://github.com/modelcontextprotocol/python-sdk/commit/b3025f93b4ba63c6d0d04b1357947ecdb78eb0bd [4]: https://github.com/modelcontextprotocol/python-sdk/commit/ddb7b78ef4dbf54906a7cb1f7efbf880503baba9 [5]: https://debusine.debian.net/debian/r-python-python3.15/ -- "Can you imagine what I would do if I could do all I can?" -- Sun Tzu Saludos /\/\ /\ >< `/
From: Max <[email protected]> Date: Mon, 15 Jun 2026 16:50:57 +0100 Subject: Deflake the issue-1363 tests: wait for lifespan startup instead of sleeping (#2879) --- tests/issues/test_1363_race_condition_streamable_http.py | 21 +++++++++++++------ 1 file changed, 15 insertions(+), 6 deletions(-) Index: python-mcp/tests/issues/test_1363_race_condition_streamable_http.py =================================================================== --- python-mcp.orig/tests/issues/test_1363_race_condition_streamable_http.py +++ python-mcp/tests/issues/test_1363_race_condition_streamable_http.py @@ -21,6 +21,7 @@ from collections.abc import AsyncGenerat from contextlib import asynccontextmanager import anyio +import anyio.to_thread import httpx import pytest from starlette.applications import Starlette @@ -68,6 +69,7 @@ class ServerThread(threading.Thread): super().__init__(daemon=True) self.app = app self._stop_event = threading.Event() + self._ready_event = threading.Event() def run(self) -> None: """Run the lifespan in a new event loop.""" @@ -78,12 +80,19 @@ class ServerThread(threading.Thread): lifespan_context = getattr(self.app.router, "lifespan_context", None) assert lifespan_context is not None # Tests always create apps with lifespan async with lifespan_context(self.app): + # Only signal readiness once lifespan startup has completed, i.e. the + # session manager's task group exists and requests can be handled. + self._ready_event.set() # Wait until stop is requested while not self._stop_event.is_set(): await anyio.sleep(0.1) anyio.run(run_lifespan) + def wait_ready(self, timeout: float = 5.0) -> None: + """Block until the lifespan has started; call from a worker thread, not the event loop.""" + assert self._ready_event.wait(timeout), "server thread did not start its lifespan in time" + def stop(self) -> None: """Signal the thread to stop.""" self._stop_event.set() @@ -134,8 +143,8 @@ async def test_race_condition_invalid_ac server_thread.start() try: - # Give the server thread a moment to start - await anyio.sleep(0.1) + # Wait for the server thread to enter the lifespan before sending requests + await anyio.to_thread.run_sync(server_thread.wait_ready) # Suppress WARNING logs (expected validation errors) and capture ERROR logs with caplog.at_level(logging.ERROR): @@ -206,8 +215,8 @@ async def test_race_condition_invalid_co server_thread.start() try: - # Give the server thread a moment to start - await anyio.sleep(0.1) + # Wait for the server thread to enter the lifespan before sending requests + await anyio.to_thread.run_sync(server_thread.wait_ready) # Suppress WARNING logs (expected validation errors) and capture ERROR logs with caplog.at_level(logging.ERROR): @@ -247,8 +256,8 @@ async def test_race_condition_message_ro server_thread.start() try: - # Give the server thread a moment to start - await anyio.sleep(0.1) + # Wait for the server thread to enter the lifespan before sending requests + await anyio.to_thread.run_sync(server_thread.wait_ready) # Suppress WARNING logs (expected validation errors) and capture ERROR logs with caplog.at_level(logging.ERROR):
From: Max <[email protected]> Date: Tue, 2 Jun 2026 20:30:17 +0100 Subject: Run transport security tests in process instead of over sockets (#2764) Index: python-mcp/tests/server/test_streamable_http_security.py =================================================================== --- python-mcp.orig/tests/server/test_streamable_http_security.py +++ python-mcp/tests/server/test_streamable_http_security.py @@ -1,293 +1,130 @@ """Tests for StreamableHTTP server DNS rebinding protection.""" -import logging -import multiprocessing -import socket -from collections.abc import AsyncGenerator +from collections.abc import AsyncIterator from contextlib import asynccontextmanager import httpx import pytest -import uvicorn from starlette.applications import Starlette from starlette.routing import Mount -from starlette.types import Receive, Scope, Send from mcp.server import Server from mcp.server.streamable_http_manager import StreamableHTTPSessionManager from mcp.server.transport_security import TransportSecuritySettings -from mcp.types import Tool -from tests.test_helpers import wait_for_server +from tests.interaction.transports import StreamingASGITransport -logger = logging.getLogger(__name__) SERVER_NAME = "test_streamable_http_security_server" +# The in-process app is mounted at this origin purely so URLs are well-formed and the default +# Host header is a localhost form; nothing listens here. +BASE_URL = "http://127.0.0.1:8000"; [email protected] -def server_port() -> int: - with socket.socket() as s: - s.bind(("127.0.0.1", 0)) - return s.getsockname()[1] +@asynccontextmanager +async def streamable_http_security_client( + security_settings: TransportSecuritySettings | None = None, +) -> AsyncIterator[httpx.AsyncClient]: + """Yield an httpx client served in process by a StreamableHTTP app with the given settings.""" + session_manager = StreamableHTTPSessionManager(app=Server(SERVER_NAME), security_settings=security_settings) + app = Starlette(routes=[Mount("/", app=session_manager.handle_request)]) [email protected] -def server_url(server_port: int) -> str: # pragma: no cover - return f"http://127.0.0.1:{server_port}"; + async with session_manager.run(): + async with httpx.AsyncClient(transport=StreamingASGITransport(app), base_url=BASE_URL) as client: + yield client -class SecurityTestServer(Server): # pragma: no cover - def __init__(self): - super().__init__(SERVER_NAME) +def _base_headers() -> dict[str, str]: + """Headers every well-formed request carries, so each test varies only the header under test.""" + return {"Accept": "application/json, text/event-stream", "Content-Type": "application/json"} - async def on_list_tools(self) -> list[Tool]: - return [] - -def run_server_with_settings(port: int, security_settings: TransportSecuritySettings | None = None): # pragma: no cover - """Run the StreamableHTTP server with specified security settings.""" - app = SecurityTestServer() - - # Create session manager with security settings - session_manager = StreamableHTTPSessionManager( - app=app, - json_response=False, - stateless=False, - security_settings=security_settings, - ) - - # Create the ASGI handler - async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None: - await session_manager.handle_request(scope, receive, send) - - # Create Starlette app with lifespan - @asynccontextmanager - async def lifespan(app: Starlette) -> AsyncGenerator[None, None]: - async with session_manager.run(): - yield - - routes = [ - Mount("/", app=handle_streamable_http), - ] - - starlette_app = Starlette(routes=routes, lifespan=lifespan) - uvicorn.run(starlette_app, host="127.0.0.1", port=port, log_level="error") - - -def start_server_process(port: int, security_settings: TransportSecuritySettings | None = None): - """Start server in a separate process.""" - process = multiprocessing.Process(target=run_server_with_settings, args=(port, security_settings)) - process.start() - # Wait for server to be ready to accept connections - wait_for_server(port) - return process +def _initialize_body() -> dict[str, object]: + """A minimal initialize POST body; these tests assert header validation, not the handshake.""" + return {"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}} @pytest.mark.anyio -async def test_streamable_http_security_default_settings(server_port: int): - """Test StreamableHTTP with default security settings (protection enabled).""" - process = start_server_process(server_port) - - try: - # Test with valid localhost headers - async with httpx.AsyncClient(timeout=5.0) as client: - # POST request to initialize session - response = await client.post( - f"http://127.0.0.1:{server_port}/";, - json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, - headers={ - "Accept": "application/json, text/event-stream", - "Content-Type": "application/json", - }, - ) - assert response.status_code == 200 - assert "mcp-session-id" in response.headers - - finally: - process.terminate() - process.join() +async def test_streamable_http_security_default_settings() -> None: + """With default security settings, a request with localhost headers is served.""" + async with streamable_http_security_client() as client: + response = await client.post("/", json=_initialize_body(), headers=_base_headers()) + assert response.status_code == 200 + assert "mcp-session-id" in response.headers @pytest.mark.anyio -async def test_streamable_http_security_invalid_host_header(server_port: int): - """Test StreamableHTTP with invalid Host header.""" +async def test_streamable_http_security_invalid_host_header() -> None: + """A Host header outside allowed_hosts is rejected with 421.""" security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True) - process = start_server_process(server_port, security_settings) - try: - # Test with invalid host header - headers = { - "Host": "evil.com", - "Accept": "application/json, text/event-stream", - "Content-Type": "application/json", - } - - async with httpx.AsyncClient(timeout=5.0) as client: - response = await client.post( - f"http://127.0.0.1:{server_port}/";, - json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, - headers=headers, - ) - assert response.status_code == 421 - assert response.text == "Invalid Host header" - - finally: - process.terminate() - process.join() + async with streamable_http_security_client(security_settings) as client: + response = await client.post("/", json=_initialize_body(), headers=_base_headers() | {"Host": "evil.com"}) + assert response.status_code == 421 + assert response.text == "Invalid Host header" @pytest.mark.anyio -async def test_streamable_http_security_invalid_origin_header(server_port: int): - """Test StreamableHTTP with invalid Origin header.""" +async def test_streamable_http_security_invalid_origin_header() -> None: + """An Origin header outside allowed_origins is rejected with 403.""" security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1:*"]) - process = start_server_process(server_port, security_settings) - try: - # Test with invalid origin header - headers = { - "Origin": "http://evil.com";, - "Accept": "application/json, text/event-stream", - "Content-Type": "application/json", - } - - async with httpx.AsyncClient(timeout=5.0) as client: - response = await client.post( - f"http://127.0.0.1:{server_port}/";, - json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, - headers=headers, - ) - assert response.status_code == 403 - assert response.text == "Invalid Origin header" - - finally: - process.terminate() - process.join() + async with streamable_http_security_client(security_settings) as client: + response = await client.post( + "/", json=_initialize_body(), headers=_base_headers() | {"Origin": "http://evil.com"} + ) + assert response.status_code == 403 + assert response.text == "Invalid Origin header" @pytest.mark.anyio -async def test_streamable_http_security_invalid_content_type(server_port: int): - """Test StreamableHTTP POST with invalid Content-Type header.""" - process = start_server_process(server_port) - - try: - async with httpx.AsyncClient(timeout=5.0) as client: - # Test POST with invalid content type - response = await client.post( - f"http://127.0.0.1:{server_port}/";, - headers={ - "Content-Type": "text/plain", - "Accept": "application/json, text/event-stream", - }, - content="test", - ) - assert response.status_code == 400 - assert response.text == "Invalid Content-Type header" - - # Test POST with missing content type - response = await client.post( - f"http://127.0.0.1:{server_port}/";, - headers={"Accept": "application/json, text/event-stream"}, - content="test", - ) - assert response.status_code == 400 - assert response.text == "Invalid Content-Type header" - - finally: - process.terminate() - process.join() +async def test_streamable_http_security_invalid_content_type() -> None: + """A POST whose Content-Type is not application/json (or is missing) is rejected with 400.""" + async with streamable_http_security_client() as client: + response = await client.post("/", headers=_base_headers() | {"Content-Type": "text/plain"}, content="test") + assert response.status_code == 400 + assert response.text == "Invalid Content-Type header" + + response = await client.post("/", headers={"Accept": "application/json, text/event-stream"}, content="test") + assert response.status_code == 400 + assert response.text == "Invalid Content-Type header" @pytest.mark.anyio -async def test_streamable_http_security_disabled(server_port: int): - """Test StreamableHTTP with security disabled.""" +async def test_streamable_http_security_disabled() -> None: + """With protection explicitly disabled, a disallowed Host is still served.""" settings = TransportSecuritySettings(enable_dns_rebinding_protection=False) - process = start_server_process(server_port, settings) - try: - # Test with invalid host header - should still work - headers = { - "Host": "evil.com", - "Accept": "application/json, text/event-stream", - "Content-Type": "application/json", - } - - async with httpx.AsyncClient(timeout=5.0) as client: - response = await client.post( - f"http://127.0.0.1:{server_port}/";, - json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, - headers=headers, - ) - # Should connect successfully even with invalid host - assert response.status_code == 200 - - finally: - process.terminate() - process.join() + async with streamable_http_security_client(settings) as client: + response = await client.post("/", json=_initialize_body(), headers=_base_headers() | {"Host": "evil.com"}) + assert response.status_code == 200 @pytest.mark.anyio -async def test_streamable_http_security_custom_allowed_hosts(server_port: int): - """Test StreamableHTTP with custom allowed hosts.""" +async def test_streamable_http_security_custom_allowed_hosts() -> None: + """A custom entry in allowed_hosts is served.""" 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) - try: - # Test with custom allowed host - headers = { - "Host": "custom.host", - "Accept": "application/json, text/event-stream", - "Content-Type": "application/json", - } - - async with httpx.AsyncClient(timeout=5.0) as client: - response = await client.post( - f"http://127.0.0.1:{server_port}/";, - json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}}, - headers=headers, - ) - # Should connect successfully with custom host - assert response.status_code == 200 - finally: - process.terminate() - process.join() + async with streamable_http_security_client(settings) as client: + response = await client.post("/", json=_initialize_body(), headers=_base_headers() | {"Host": "custom.host"}) + assert response.status_code == 200 @pytest.mark.anyio -async def test_streamable_http_security_get_request(server_port: int): - """Test StreamableHTTP GET request with security.""" +async def test_streamable_http_security_get_request() -> None: + """GET requests pass the same Host validation before any session handling.""" security_settings = TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=["127.0.0.1"]) - process = start_server_process(server_port, security_settings) - try: - # Test GET request with invalid host header - headers = { - "Host": "evil.com", - "Accept": "text/event-stream", - } - - async with httpx.AsyncClient(timeout=5.0) as client: - response = await client.get(f"http://127.0.0.1:{server_port}/";, headers=headers) - assert response.status_code == 421 - assert response.text == "Invalid Host header" - - # Test GET request with valid host header - headers = { - "Host": "127.0.0.1", - "Accept": "text/event-stream", - } - - async with httpx.AsyncClient(timeout=5.0) as client: - # GET requests need a session ID in StreamableHTTP - # So it will fail with "Missing session ID" not security error - response = await client.get(f"http://127.0.0.1:{server_port}/";, headers=headers) - # This should pass security but fail on session validation - assert response.status_code == 400 - body = response.json() - assert "Missing session ID" in body["error"]["message"] - - finally: - process.terminate() - process.join() + async with streamable_http_security_client(security_settings) as client: + response = await client.get("/", headers={"Accept": "text/event-stream", "Host": "evil.com"}) + assert response.status_code == 421 + assert response.text == "Invalid Host header" + + response = await client.get("/", headers={"Accept": "text/event-stream", "Host": "127.0.0.1"}) + # An allowed host passes security and fails on session validation instead. + assert response.status_code == 400 + body = response.json() + assert "Missing session ID" in body["error"]["message"] Index: python-mcp/src/mcp/server/session.py =================================================================== --- python-mcp.orig/src/mcp/server/session.py +++ python-mcp/src/mcp/server/session.py @@ -103,6 +103,7 @@ class ServerSession( ServerRequestResponder ](0) self._exit_stack.push_async_callback(lambda: self._incoming_message_stream_reader.aclose()) + self._exit_stack.push_async_callback(lambda: self._incoming_message_stream_writer.aclose()) @property def client_params(self) -> types.InitializeRequestParams | None: Index: python-mcp/src/mcp/server/streamable_http.py =================================================================== --- python-mcp.orig/src/mcp/server/streamable_http.py +++ python-mcp/src/mcp/server/streamable_http.py @@ -391,8 +391,9 @@ class StreamableHTTPServerTransport: if request_id in self._request_streams: try: # Close the request stream - await self._request_streams[request_id][0].aclose() - await self._request_streams[request_id][1].aclose() + with anyio.CancelScope(shield=True): + await self._request_streams[request_id][0].aclose() + await self._request_streams[request_id][1].aclose() except Exception: # During cleanup, we catch all exceptions since streams might be in various states logger.debug("Error closing memory streams - may already be closed") @@ -642,9 +643,11 @@ class StreamableHTTPServerTransport: await writer.send(session_message) except Exception: logger.exception("SSE response error") - await sse_stream_writer.aclose() - await sse_stream_reader.aclose() - await self._clean_up_memory_streams(request_id) + finally: + with anyio.CancelScope(shield=True): + await sse_stream_writer.aclose() + await sse_stream_reader.aclose() + await self._clean_up_memory_streams(request_id) except Exception as err: logger.exception("Error handling POST request") @@ -747,9 +750,11 @@ class StreamableHTTPServerTransport: await response(request.scope, request.receive, send) except Exception: logger.exception("Error in standalone SSE response") - await sse_stream_writer.aclose() - await sse_stream_reader.aclose() - await self._clean_up_memory_streams(GET_STREAM_KEY) + finally: + with anyio.CancelScope(shield=True): + await sse_stream_writer.aclose() + await sse_stream_reader.aclose() + await self._clean_up_memory_streams(GET_STREAM_KEY) async def _handle_delete_request(self, request: Request, send: Send) -> None: # pragma: no cover """Handle DELETE requests for explicit session termination.""" @@ -791,23 +796,24 @@ class StreamableHTTPServerTransport: request_stream_keys = list(self._request_streams.keys()) # Close all request streams asynchronously - for key in request_stream_keys: # pragma: no cover - await self._clean_up_memory_streams(key) + with anyio.CancelScope(shield=True): + for key in request_stream_keys: # pragma: no cover + await self._clean_up_memory_streams(key) - # Clear the request streams dictionary immediately - self._request_streams.clear() - try: - if self._read_stream_writer is not None: # pragma: no branch - await self._read_stream_writer.aclose() - if self._read_stream is not None: # pragma: no branch - await self._read_stream.aclose() - if self._write_stream_reader is not None: # pragma: no branch - await self._write_stream_reader.aclose() - if self._write_stream is not None: # pragma: no branch - await self._write_stream.aclose() - except Exception as e: # pragma: no cover - # During cleanup, we catch all exceptions since streams might be in various states - logger.debug(f"Error closing streams: {e}") + # Clear the request streams dictionary immediately + self._request_streams.clear() + try: + if self._read_stream_writer is not None: # pragma: no branch + await self._read_stream_writer.aclose() + if self._read_stream is not None: # pragma: no branch + await self._read_stream.aclose() + if self._write_stream_reader is not None: # pragma: no branch + await self._write_stream_reader.aclose() + if self._write_stream is not None: # pragma: no branch + await self._write_stream.aclose() + except Exception as e: # pragma: no cover + # During cleanup, we catch all exceptions since streams might be in various states + logger.debug(f"Error closing streams: {e}") async def _handle_unsupported_request(self, request: Request, send: Send) -> None: # pragma: no cover """Handle unsupported HTTP methods.""" @@ -1071,16 +1077,17 @@ class StreamableHTTPServerTransport: # Yield the streams for the caller to use yield read_stream, write_stream finally: - for stream_id in list(self._request_streams.keys()): # pragma: no cover - await self._clean_up_memory_streams(stream_id) - self._request_streams.clear() - - # Clean up the read and write streams - try: - await read_stream_writer.aclose() - await read_stream.aclose() - await write_stream_reader.aclose() - await write_stream.aclose() - except Exception as e: # pragma: no cover - # During cleanup, we catch all exceptions since streams might be in various states - logger.debug(f"Error closing streams: {e}") + with anyio.CancelScope(shield=True): + for stream_id in list(self._request_streams.keys()): # pragma: no cover + await self._clean_up_memory_streams(stream_id) + self._request_streams.clear() + + # Clean up the read and write streams + try: + await read_stream_writer.aclose() + await read_stream.aclose() + await write_stream_reader.aclose() + await write_stream.aclose() + except Exception as e: # pragma: no cover + # During cleanup, we catch all exceptions since streams might be in various states + logger.debug(f"Error closing streams: {e}") Index: python-mcp/tests/interaction/transports/__init__.py =================================================================== --- /dev/null +++ python-mcp/tests/interaction/transports/__init__.py @@ -0,0 +1,9 @@ +"""Transport-specific interaction tests, and the in-process streaming bridge they are built on. + +`StreamingASGITransport` is re-exported here as the sanctioned import point for test code +outside this suite (the bridge module itself is suite-private). +""" + +from tests.interaction.transports._bridge import StreamingASGITransport + +__all__ = ["StreamingASGITransport"] Index: python-mcp/tests/interaction/transports/_bridge.py =================================================================== --- /dev/null +++ python-mcp/tests/interaction/transports/_bridge.py @@ -0,0 +1,169 @@ +"""An in-process, full-duplex HTTP transport for driving ASGI applications from httpx. + +`httpx.ASGITransport` runs the application to completion and only then hands the buffered +response to the caller, so a server that streams its response — the streamable HTTP transport's +SSE responses — can never converse with the client mid-request: a server-initiated request +nested inside a still-open call deadlocks. `StreamingASGITransport` removes that limitation by +running the application as a background task and forwarding every `http.response.body` chunk to +the client the moment it is sent. Everything happens on the one event loop: no sockets, no +threads, no sleeps, no extra dependencies. + +The behavioural contract, pinned by `test_bridge.py`: + +- The request body is buffered before the application is invoked (MCP requests are small JSON + documents); the response streams chunk by chunk. +- Closing the response — or the whole client — delivers `http.disconnect` to the application, + exactly as a real server sees when its peer goes away. +- An exception the application raises before sending `http.response.start` fails the originating + request with that same exception. After the response has started, a failure is visible to the + client only through the response itself (status code, truncated body) — the same signal a real + server over a real socket would give. + +The transport owns an anyio task group for the application tasks; it is opened and closed by +`httpx.AsyncClient`'s own context manager, so use the client as a context manager (the suite +always does). Closing the transport cancels every running application task by default; set +`cancel_on_close=False` to wait for the application's own disconnect handling instead. +""" + +import math +from collections.abc import AsyncIterator +from types import TracebackType + +import anyio +import anyio.abc +import httpx +from anyio.streams.memory import MemoryObjectReceiveStream +from starlette.types import ASGIApp, Message, Scope + + +class _StreamingResponseBody(httpx.AsyncByteStream): + """A response body that yields chunks as the application produces them. + + Closing it tells the application the client has gone away (`http.disconnect`), mirroring a + peer that drops the connection mid-response. + """ + + def __init__(self, chunks: MemoryObjectReceiveStream[bytes], client_disconnected: anyio.Event) -> None: + self._chunks = chunks + self._client_disconnected = client_disconnected + + async def __aiter__(self) -> AsyncIterator[bytes]: + async for chunk in self._chunks: + yield chunk + + async def aclose(self) -> None: + self._client_disconnected.set() + await self._chunks.aclose() + + +class StreamingASGITransport(httpx.AsyncBaseTransport): + """Drive an ASGI application in-process, streaming each response as it is produced. + + With `cancel_on_close` (the default), closing the transport cancels every application task + still running so harness teardown can never hang. Setting it to False makes the transport wait + for the application's own disconnect handling to complete instead, which is the path the legacy + SSE server transport relies on for resource cleanup. + """ + + _task_group: anyio.abc.TaskGroup + + def __init__(self, app: ASGIApp, *, cancel_on_close: bool = True) -> None: + self._app = app + self._cancel_on_close = cancel_on_close + + async def __aenter__(self) -> "StreamingASGITransport": + self._task_group = anyio.create_task_group() + await self._task_group.__aenter__() + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None = None, + exc_value: BaseException | None = None, + traceback: TracebackType | None = None, + ) -> None: + # httpx closes every streamed response before closing the transport, so by now each + # application task has been delivered `http.disconnect`. Either cancel immediately, or wait + # for the application's own disconnect handling to unwind. + if self._cancel_on_close: + self._task_group.cancel_scope.cancel() + await self._task_group.__aexit__(exc_type, exc_value, traceback) + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + assert isinstance(request.stream, httpx.AsyncByteStream) + request_body = b"".join([chunk async for chunk in request.stream]) + + scope: Scope = { + "type": "http", + "asgi": {"version": "3.0"}, + "http_version": "1.1", + "method": request.method, + "scheme": request.url.scheme, + "path": request.url.path, + "raw_path": request.url.raw_path.split(b"?", maxsplit=1)[0], + "query_string": request.url.query, + "root_path": "", + "headers": [(name.lower(), value) for name, value in request.headers.raw], + "server": (request.url.host, request.url.port), + "client": ("127.0.0.1", 1234), + } + + request_delivered = False + client_disconnected = anyio.Event() + response_started = anyio.Event() + response_status = 0 + response_headers: list[tuple[bytes, bytes]] = [] + application_error: Exception | None = None + chunk_writer, chunk_reader = anyio.create_memory_object_stream[bytes](math.inf) + + async def receive_request() -> Message: + nonlocal request_delivered + if not request_delivered: + request_delivered = True + return {"type": "http.request", "body": request_body, "more_body": False} + await client_disconnected.wait() + return {"type": "http.disconnect"} + + async def send_response(message: Message) -> None: + nonlocal response_status, response_headers + if message["type"] == "http.response.start": + response_status = message["status"] + response_headers = list(message.get("headers", [])) + response_started.set() + return + assert message["type"] == "http.response.body" + body: bytes = message.get("body", b"") + if body: + await chunk_writer.send(body) + if not message.get("more_body", False): + await chunk_writer.aclose() + + async def run_application() -> None: + nonlocal application_error + try: + await self._app(scope, receive_request, send_response) + except Exception as exc: # The bridge is the application's outermost boundary: a crash + # must fail the originating request (or show up in the already-started response), + # never tear down the task group shared with every other in-flight request. + application_error = exc + finally: + response_started.set() + await chunk_writer.aclose() + + self._task_group.start_soon(run_application) + try: + await response_started.wait() + if application_error is not None: + raise application_error + except BaseException: + # No response will be built, so close the reader the response body would have owned + # and tell the application its peer has gone away. + client_disconnected.set() + await chunk_reader.aclose() + raise + return httpx.Response( + status_code=response_status, + headers=response_headers, + stream=_StreamingResponseBody(chunk_reader, client_disconnected), + request=request, + )
From: Max <[email protected]> Date: Fri, 21 Aug 2026 12:02:46 +0100 Subject: [v1.x] Complete the FastMCP Settings model at import time (#3352) --- src/mcp/server/fastmcp/server.py | 5 +++++ tests/server/fastmcp/test_server.py | 6 ++++++ 2 files changed, 11 insertions(+) Index: python-mcp/src/mcp/server/fastmcp/server.py =================================================================== --- python-mcp.orig/src/mcp/server/fastmcp/server.py +++ python-mcp/src/mcp/server/fastmcp/server.py @@ -1083,6 +1083,11 @@ class FastMCP(Generic[LifespanResultT]): raise ValueError(str(e)) +# `Settings.lifespan` refers to FastMCP, which is only defined above; complete the model now so +# settings sources never see an unresolved annotation when a FastMCP instance is created. +Settings.model_rebuild() + + class StreamableHTTPASGIApp: """ ASGI application for Streamable HTTP server transport. Index: python-mcp/tests/server/fastmcp/test_server.py =================================================================== --- python-mcp.orig/tests/server/fastmcp/test_server.py +++ python-mcp/tests/server/fastmcp/test_server.py @@ -10,6 +10,7 @@ from starlette.routing import Mount, Rou from mcp.server.fastmcp import Context, FastMCP from mcp.server.fastmcp.prompts.base import Message, UserMessage from mcp.server.fastmcp.resources import FileResource, FunctionResource +from mcp.server.fastmcp.server import Settings from mcp.server.fastmcp.utilities.types import Audio, Image from mcp.server.session import ServerSession from mcp.server.transport_security import TransportSecuritySettings @@ -31,6 +32,11 @@ if TYPE_CHECKING: from mcp.server.fastmcp import Context +def test_settings_model_is_complete_at_import(): + """The Settings model resolves its FastMCP annotation at import, so building one needs no deferred rebuild.""" + assert Settings.__pydantic_complete__ + + class TestServer: @pytest.mark.anyio async def test_create_server(self):
