From 979a0613966bd4cab1d346ebd6c602ebf8fb9fa9 Mon Sep 17 00:00:00 2001 From: mateo-berri <277851410+mateo-berri@users.noreply.github.com> Date: Tue, 5 May 2026 22:40:24 +0000 Subject: [PATCH] fix(mcp): don't hold stateful session lock for streaming GETs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The per-session lock wraps handle_request, so a long-lived GET (SSE stream held open for the life of the session) would block every subsequent POST on the same mcp-session-id. Only POST/DELETE mutate the shared MCPAuthenticatedUser, so it's sufficient to serialize those — GETs run lock-free and stream concurrently. --- .../proxy/_experimental/mcp_server/server.py | 10 +- .../mcp_server/test_mcp_server.py | 91 +++++++++++++++++++ 2 files changed, 100 insertions(+), 1 deletion(-) diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index a5ad9dd8762..d8e32ef46e9 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -3044,8 +3044,16 @@ if MCP_AVAILABLE: # Serialize requests on the same stateful session so concurrent # callers don't clobber each other's auth context mid-flight. + # + # Skip the lock for streaming GETs (SSE channels held open for the + # life of the session): holding a per-session lock for a long-lived + # stream would block every subsequent POST on the same session. + # POST/DELETE are the methods that actually mutate the shared + # auth context, so serializing those is sufficient for the + # clobbering race between concurrent JSON-RPC calls. session_lock: Optional[asyncio.Lock] = None - if use_stateful and session_id: + request_method = (scope.get("method") or "").upper() + if use_stateful and session_id and request_method in ("POST", "DELETE"): session_lock = _stateful_session_locks.setdefault( session_id, asyncio.Lock() ) diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py index bfe6120cd51..4e42d15b266 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py @@ -1623,6 +1623,97 @@ async def test_stateful_mcp_session_serializes_concurrent_requests(): ), "concurrent requests on same stateful session must be serialized" +@pytest.mark.asyncio +async def test_stateful_mcp_get_stream_does_not_block_post(): + """ + A long-lived GET (server-to-client SSE stream) on a stateful session + must NOT hold the per-session lock — otherwise subsequent POSTs on the + same mcp-session-id hang for the lifetime of the stream. + """ + try: + from litellm.proxy._experimental.mcp_server import server as mcp_server + from litellm.proxy._experimental.mcp_server.server import ( + handle_streamable_http_mcp, + session_manager_stateful, + ) + except ImportError: + pytest.skip("MCP server not available") + + session_id = "stream-session-1" + owner_auth = UserAPIKeyAuth(api_key="owner-key", user_id="owner") + mcp_server._stateful_session_auth_contexts[session_id] = ( + mcp_server.MCPAuthenticatedUser(user_api_key_auth=owner_auth) + ) + mcp_server._stateful_session_owners[session_id] = mcp_server._owner_fingerprint_for( + owner_auth + ) + + stream_release = asyncio.Event() + post_finished = asyncio.Event() + + async def handle(s, r, se): + if s.get("method") == "GET": + await stream_release.wait() + else: + post_finished.set() + + async def call(method: str, body: bytes = b""): + scope = { + "type": "http", + "method": method, + "path": "/mcp", + "headers": [(b"mcp-session-id", session_id.encode())], + } + receive = AsyncMock( + return_value={ + "type": "http.request", + "body": body, + "more_body": False, + } + ) + await handle_streamable_http_mcp(scope, receive, AsyncMock()) + + try: + with ( + patch( + "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context", + new_callable=AsyncMock, + return_value=(owner_auth, None, None, None, None, None), + ), + patch( + "litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED", + True, + ), + patch.object( + session_manager_stateful, "handle_request", side_effect=handle + ), + patch.object( + session_manager_stateful, + "_server_instances", + {session_id: MagicMock()}, + ), + ): + stream_task = asyncio.create_task(call("GET")) + await asyncio.sleep(0.05) + assert not stream_task.done(), "GET stream should still be open" + + post_task = asyncio.create_task( + call( + "POST", + body=b'{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}', + ) + ) + await asyncio.wait_for(post_finished.wait(), timeout=1.0) + await post_task + + stream_release.set() + await stream_task + finally: + mcp_server._stateful_session_auth_contexts.pop(session_id, None) + mcp_server._stateful_session_owners.pop(session_id, None) + mcp_server._stateful_session_locks.pop(session_id, None) + + @pytest.mark.asyncio @pytest.mark.no_parallel async def test_mcp_routing_with_conflicting_alias_and_group_name():