From 1f3037dd443af283abe3771c55aca4b6096d290a Mon Sep 17 00:00:00 2001 From: Cursor Agent Date: Tue, 8 Sep 2026 02:06:06 +0000 Subject: [PATCH] fix(mcp): gate SSE messages until initialize completes SSE /mcp/messages POSTs could reach ServerSession before the handshake finished, causing -32602 "Received request before initialization was complete". Mirror the WebSocket init_done latch in the bridge so later frames wait for the initialize result. Co-authored-by: Zsanz3 --- deploy/docker/mcp_bridge.py | 52 ++++++++++- tests/test_issue_2233_mcp_sse.py | 147 +++++++++++++++++++++++++++++++ 2 files changed, 198 insertions(+), 1 deletion(-) create mode 100644 tests/test_issue_2233_mcp_sse.py diff --git a/deploy/docker/mcp_bridge.py b/deploy/docker/mcp_bridge.py index 653bc368b..04b0ff077 100644 --- a/deploy/docker/mcp_bridge.py +++ b/deploy/docker/mcp_bridge.py @@ -247,7 +247,9 @@ async def ws_to_srv(): class _MCPSseApp: async def __call__(self, scope, receive, send): async with sse.connect_sse(scope, receive, send) as (read_stream, write_stream): - await mcp.run(read_stream, write_stream, init_opts) + # Same init_done latch as /mcp/ws (#2233): concurrent POSTs that + # arrive during initialize must not reach ServerSession yet. + await _run_with_init_gate(mcp.run, read_stream, write_stream, init_opts) app.routes.append(Route(f"{base}/sse", endpoint=_MCPSseApp())) app.routes.append(Mount(f"{base}/messages", app=sse.handle_post_message)) @@ -263,6 +265,54 @@ async def _schema_endpoint(): # ── helpers ──────────────────────────────────────────────────── +async def _run_with_init_gate(run, read_stream, write_stream, init_opts) -> None: + """Hold inbound messages until the server finishes initialize. + + Mirrors the WebSocket ``init_done`` gate: the first client frame + (``initialize``) is forwarded immediately; later frames wait until the + server writes its first outbound message (the initialize result). This + lives in the bridge so we do not depend on fixing the upstream MCP SDK + race (python-sdk #423 / #2583). + """ + c2s_send, c2s_recv = anyio.create_memory_object_stream(100) + s2c_send, s2c_recv = anyio.create_memory_object_stream(100) + init_done = anyio.Event() + + async def srv_to_client(): + first = True + try: + async for msg in s2c_recv: + await write_stream.send(msg) + if first: + init_done.set() + first = False + finally: + # make sure cleanup survives TaskGroup cancellation + with anyio.CancelScope(shield=True): + with suppress(Exception): + await write_stream.aclose() + + async def client_to_srv(): + try: + # 1st frame is always "initialize" + first = await read_stream.receive() + await c2s_send.send(first) + await init_done.wait() # block until server ready + async for msg in read_stream: + await c2s_send.send(msg) + except (anyio.EndOfStream, anyio.ClosedResourceError): + pass + finally: + with anyio.CancelScope(shield=True): + with suppress(Exception): + await c2s_send.aclose() + + async with anyio.create_task_group() as tg: + tg.start_soon(run, c2s_recv, s2c_send, init_opts) + tg.start_soon(client_to_srv) + tg.start_soon(srv_to_client) + + def _route_name(path: str) -> str: return re.sub(r"[/{}}]", "_", path).strip("_") diff --git a/tests/test_issue_2233_mcp_sse.py b/tests/test_issue_2233_mcp_sse.py new file mode 100644 index 000000000..ddb7715f6 --- /dev/null +++ b/tests/test_issue_2233_mcp_sse.py @@ -0,0 +1,147 @@ +""" +Tests for issue #2233 — MCP SSE rejects concurrent requests during initialize. + +Clients that POST initialize plus tools/call in the same burst hit +ServerSession before the handshake finishes and get -32602 +("Received request before initialization was complete"). + +The WebSocket transport already gates on init_done; SSE must do the same +in mcp_bridge (not by patching the upstream MCP SDK race). +""" + +import importlib.util +import sys +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + + +BRIDGE = Path("deploy/docker/mcp_bridge.py") + + +def _bridge_source() -> str: + return BRIDGE.read_text() + + +def _sse_app_source() -> str: + source = _bridge_source() + start = source.index("class _MCPSseApp") + end = source.index("app.routes.append", start) + return source[start:end] + + +class TestSSEInitGateInBridge: + """Source-level checks that SSE uses the same init_done latch as WS.""" + + def test_sse_calls_init_gate_helper(self): + body = _sse_app_source() + assert "_run_with_init_gate" in body + assert "mcp.run" in body + + def test_init_gate_helper_mirrors_ws_latch(self): + source = _bridge_source() + assert "async def _run_with_init_gate" in source + assert "init_done = anyio.Event()" in source + assert "await init_done.wait()" in source + assert "init_done.set()" in source + # first inbound frame is forwarded before the wait (initialize) + helper = source[source.index("async def _run_with_init_gate") :] + wait_at = helper.index("await init_done.wait()") + send_at = helper.index("await c2s_send.send(first)") + assert send_at < wait_at + + def test_callable_class_pattern_preserved(self): + """#1850: Route must still receive a class instance, not a function.""" + source = _bridge_source() + assert "class _MCPSseApp" in source + assert "async def __call__(self, scope, receive, send)" in source + assert "_MCPSseApp()" in source + + +def _load_bridge_with_mocks(): + """Import mcp_bridge without Docker/MCP runtime deps.""" + mods = [ + "mcp", + "mcp.types", + "mcp.server", + "mcp.server.sse", + "mcp.server.lowlevel", + "mcp.server.lowlevel.server", + "mcp.server.models", + "fastapi", + "fastapi.responses", + "starlette", + "starlette.routing", + "httpx", + "pydantic", + "auth", + ] + saved = {name: sys.modules.get(name) for name in mods} + for name in mods: + sys.modules.setdefault(name, MagicMock()) + + # pydantic.BaseModel / TypeAdapter and FastAPI are only needed at + # attach_mcp() time; module import just binds names. + spec = importlib.util.spec_from_file_location("mcp_bridge_2233", BRIDGE) + module = importlib.util.module_from_spec(spec) + try: + spec.loader.exec_module(module) + finally: + for name, prev in saved.items(): + if prev is None: + sys.modules.pop(name, None) + else: + sys.modules[name] = prev + return module + + +class TestRunWithInitGate: + """Behavioral test of the bridge latch (no live MCP server).""" + + def test_later_messages_wait_until_initialize_result(self): + anyio = pytest.importorskip("anyio") + bridge = _load_bridge_with_mocks() + + async def body(): + inbound_send, inbound_recv = anyio.create_memory_object_stream(100) + outbound_send, outbound_recv = anyio.create_memory_object_stream(100) + seen: list = [] + released = anyio.Event() + + async def fake_run(read, write, _opts): + first = await read.receive() + seen.append(first) + # initialize is in flight — extra frames must not appear yet + with anyio.move_on_after(0.05) as scope: + extra = await read.receive() + assert scope.cancel_called, extra if not scope.cancel_called else None + await write.send("initialize-result") + released.set() + async for msg in read: + seen.append(msg) + await write.aclose() + + async def client(): + await inbound_send.send("initialize") + await inbound_send.send("tools/call-1") + await inbound_send.send("tools/call-2") + await inbound_send.aclose() + + async with anyio.create_task_group() as tg: + tg.start_soon( + bridge._run_with_init_gate, + fake_run, + inbound_recv, + outbound_send, + None, + ) + tg.start_soon(client) + result = await outbound_recv.receive() + assert result == "initialize-result" + await released.wait() + + assert seen[0] == "initialize" + assert seen[1:] == ["tools/call-1", "tools/call-2"] + + anyio.run(body)