Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 59 additions & 0 deletions tests/mcp_tests/test_proxy_mcp_e2e.py
Original file line number Diff line number Diff line change
Expand Up @@ -234,3 +234,62 @@ async def _call_and_get_text(
)
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"

Loading