diff --git a/tests/mcp_tests/test_proxy_mcp_e2e.py b/tests/mcp_tests/test_proxy_mcp_e2e.py index 2b8cde54710..5aff63a51ef 100644 --- a/tests/mcp_tests/test_proxy_mcp_e2e.py +++ b/tests/mcp_tests/test_proxy_mcp_e2e.py @@ -234,3 +234,62 @@ class TestProxyMcpSimpleConnections: ) assert stdio_result == "5" assert streamable_result == "9" + + +class TestProxyMcpStatelessBehavior: + """ + Verify that the LiteLLM MCP proxy operates in stateless mode. + + When StreamableHTTPSessionManager is configured with stateless=True, + independent clients must be able to connect, list tools, and call tools + without sharing or inheriting session state from other clients. + + With stateless=False this fails because the server tracks sessions and + expects clients to supply an mcp-session-id header obtained from a + prior handshake — breaking clients that don't manage session IDs. + + Regression test for https://github.com/BerriAI/litellm/issues/20242 + """ + + @pytest.mark.asyncio + async def test_independent_clients_no_shared_session( + self, proxy_server_url: str + ) -> None: + """Two independent clients connect and operate without sharing session state.""" + async with asyncio.timeout(30): + # --- Client A: connect, initialize, call tool --- + async with streamablehttp_client( + url=f"{proxy_server_url}/mcp", + headers={ + "Authorization": PROXY_AUTHORIZATION_HEADER, + "x-mcp-servers": "math_stdio", + }, + ) as (read_a, write_a, _get_sid_a): + async with ClientSession(read_a, write_a) as session_a: + await session_a.initialize() + result_a = await session_a.call_tool( + "add", arguments={"a": 10, "b": 20} + ) + assert result_a.content + text_a = getattr(result_a.content[0], "text", None) + assert text_a == "30" + + # --- Client B: completely independent connection --- + async with streamablehttp_client( + url=f"{proxy_server_url}/mcp", + headers={ + "Authorization": PROXY_AUTHORIZATION_HEADER, + "x-mcp-servers": "math_stdio", + }, + ) as (read_b, write_b, _get_sid_b): + async with ClientSession(read_b, write_b) as session_b: + await session_b.initialize() + tools = await session_b.list_tools() + assert any(t.name.endswith("add") for t in tools.tools) + result_b = await session_b.call_tool( + "add", arguments={"a": 100, "b": 200} + ) + assert result_b.content + text_b = getattr(result_b.content[0], "text", None) + assert text_b == "300" +