feat(mcp): support stateless and stateful clients via session-id routing

[Sameerlite]

Summary

Support both stateless (curl, Inspector) and stateful (Claude Code, Cursor, VSCode) MCP clients simultaneously by routing based on mcp-session-id header and request method.

Fixes LIT-2196
Fixes LIT-2656

Changes

• Add session_manager_stateful (stateless=False) alongside stateless manager
• Route: has session ID → stateful; initialize (no ID) → stateful; else → stateless
• Peek POST body to detect initialize for routing; replay via wrapped receive
• Handle stale session IDs for both managers

Why

• stateless=True: curl/Inspector work ✅
• stateless=False: progress notifications + mcp-session-id ✅
• Routing: stateless for clients without session ID, stateful for clients with session ID (or initialize)

---

Note

Medium Risk
Changes MCP request routing and stateful session lifecycle/auth-context handling; regressions could break MCP clients, leak session state across callers, or cause stuck/terminated sessions under concurrency.

Overview
Adds dual MCP StreamableHTTP session managers (stateless and stateful) and routes each request based on mcp-session-id presence and whether a POST body is a JSON-RPC initialize call (peek+replay with a capped buffer).

Introduces stateful session tracking: per-session auth-context storage/refresh, per-session request serialization (POST/DELETE), idle-timeout cleanup with background task, and owner-binding (hashed API key/OAuth bearer/IP) to reject hijacked mcp-session-ids with 403; updates stale-session DELETE handling to clear tracking. Tests are expanded/updated to cover routing, body-peek limits, owner binding, concurrency locking, and cleanup behavior.

^{Reviewed by Cursor Bugbot for commit 57a3c7a. Bugbot is set up for automated code reviews on this repo. Configure here.}

[greptile-apps]

Greptile Summary

This PR adds dual StreamableHTTP session managers (stateless + stateful) so that both stateless clients (curl, Inspector) and stateful clients (Claude Code, Cursor, VSCode) can connect simultaneously. Routing is determined by the presence of an mcp-session-id header or by peeking the POST body to detect a JSON-RPC initialize call.

• New stateful session infrastructure: per-session owner-binding (hashed credential fingerprint), per-session locks to serialize POST/DELETE, active-request reference counting to protect in-flight sessions from cleanup, and an idle-timeout background task.
• Routing: has session-id → stateful, no session-id + initialize → stateful, no session-id + other → stateless; stale session IDs are handled before routing, and a per-owner session cap (100) with LRU eviction guards against initialize spam.
• Test suite substantially expanded: new routing, ownership-binding, concurrency, and session-cap tests added; existing tests updated to patch session_manager_stateless/session_manager_stateful instead of the removed session_manager alias; cleanup-task leak in test_concurrent_initialize_session_managers is fixed.

Confidence Score: 4/5

Routing, ownership binding, and session lifecycle logic are all sound; the change is safe to merge with the minor body-peek inefficiency noted.

The dual-manager routing and stateful session lifecycle are well-structured and the tests cover the main paths. The one issue found—body peeking for every POST even when the session_id is already known—is a performance overhead with no correctness impact, not a blocking defect.

litellm/proxy/_experimental/mcp_server/server.py — the body-peek optimization and the previously noted anonymous fingerprint edge case in deployments with no credentials.

Important Files Changed

Filename	Overview
litellm/proxy/_experimental/mcp_server/server.py	Core routing change: adds session_manager_stateful (stateless=False) alongside the existing stateless manager, routes via mcp-session-id + body peek, and introduces owner-binding, per-session locking, active-request tracking, and idle-timeout cleanup. Logic is well-structured; a minor inefficiency remains where the body is peeked for POST requests even when the session_id is already known.
litellm/proxy/_experimental/mcp_server/auth/litellm_auth_handler.py	One-line change: user_api_key_auth parameter in MCPAuthenticatedUser.init made Optional to support unauthenticated/passthrough contexts used by the new stateful session tracking.
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py	Test suite substantially expanded: patches updated from session_manager alias to session_manager_stateless/stateful, cleanup task leak fixed in test_concurrent_initialize_session_managers, and new routing/ownership/concurrency tests added with correct patch targets and proper teardown.
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_stale_session.py	Stale-session tests updated to patch session_manager_stateful instead of the stateless alias; new test_failed_delete_preserves_stateful_session_tracking added and test_valid_mcp_session_id_is_preserved correctly targets the stateful manager.
tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py	Minor fix: test_mcp_path_based_server_segregation now monkeypatches both session_manager_stateless and session_manager_stateful to keep path-segregation coverage working after the dual-manager refactor.
tests/mcp_tests/test_mcp_server.py	test_streamable_http_mcp_handler_mock updated: patches renamed to session_manager_stateless/stateful, receive mock now returns a non-initialize POST body ({}) so routing correctly hits stateless, and assertions verify that only the stateless manager is called.

_{Reviews (18): Last reviewed commit: "Merge branch 'litellm_internal_staging' ..." | Re-trigger Greptile}

[codecov]

Codecov Report

❌ Patch coverage is 93.60000% with 16 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
litellm/proxy/_experimental/mcp_server/server.py	93.60%	16 Missing ⚠️

📢 Thoughts on this report? Let us know!

[Sameerlite]

@greptile-apps re-review

[Sameerlite]

@greptile-apps re-review

[krrish-berri-2]

@Sameerlite — could you add a screenshot or short video showing that this change works as expected? It really helps reviewers verify the fix quickly. Thanks!

[Sameerlite]

This video shows how litellm now sends mcp-session-id which client can use later
https://www.loom.com/share/1a21e2458e4b4bbeaaaf7519a7f947dc

[krrish-berri-2]

how was this qa'ed? @Sameerlite

[Sameerlite]

@krrish-berri-2 The video i added is the one how i qaed it. I created a client which sends init which is required to start a stateful session, litellm returned the id as you can see in the video, and it also worked with other tool calls

[mateo-berri]

bugbot run

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

Autofix Details

Bugbot Autofix prepared fixes for both issues found in the latest run.

• ✅ Fixed: Stale sessions stay stateful
  • Stale session handling now runs before manager selection and re-resolves the stripped session header so non-initialize stale requests route stateless.
• ✅ Fixed: Chunked initialize routes stateless
  • Initialize detection now reads and replays all ASGI request body chunks before routing, so chunked initialize POSTs route stateful.

_{You can send follow-ups to the cloud agent here.}

[CLAassistant]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you all sign our Contributor License Agreement before we can accept your contribution.
3 out of 5 committers have signed the CLA.

✅ Sameerlite
✅ mateo-berri
✅ yuneng-berri
❌ cursoragent
❌ claude
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

[mateo-berri]

bugbot run

[mateo-berri]

bugbot run

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

Autofix Details

Bugbot Autofix prepared a fix for the issue found in the latest run.

• ✅ Fixed: Stateful sessions never expire
  • Added idle cleanup that terminates stale stateful MCP sessions and removes their stored auth contexts when clients disconnect without DELETE.

Preview (6ea4046135)

diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py
--- a/litellm/proxy/_experimental/mcp_server/server.py
+++ b/litellm/proxy/_experimental/mcp_server/server.py
@@ -6,6 +6,7 @@
 
 import asyncio
 import contextlib
+import json
 import time
 import types
 import traceback
@@ -27,7 +28,7 @@
 from pydantic import AnyUrl, ConfigDict
 from starlette.requests import Request as StarletteRequest
 from starlette.responses import JSONResponse
-from starlette.types import Receive, Scope, Send
+from starlette.types import Message, Receive, Scope, Send
 
 from litellm._logging import verbose_logger
 from litellm.constants import MAXIMUM_TRACEBACK_LINES_TO_LOG
@@ -69,6 +70,7 @@
 _byok_cred_cache: Dict[Tuple[str, str], Tuple[Optional[str], float]] = {}
 _BYOK_CRED_CACHE_TTL = 60  # seconds
 _BYOK_CRED_CACHE_MAX_SIZE = 4096  # cap to prevent unbounded growth
+_STATEFUL_SESSION_IDLE_TIMEOUT_SECONDS = 30 * 60
 
 
 def _invalidate_byok_cred_cache(user_id: str, server_id: str) -> None:
@@ -236,13 +238,25 @@
     sse: SseServerTransport = SseServerTransport("/mcp/sse/messages")
 
     # Create session managers
-    session_manager = StreamableHTTPSessionManager(
+    session_manager_stateless = StreamableHTTPSessionManager(
         app=server,
         event_store=None,
         json_response=False,  # enables SSE streaming
         stateless=True,
     )
 
+    session_manager_stateful = StreamableHTTPSessionManager(
+        app=server,
+        event_store=None,  # TODO: Add EventStore for reconnection/event replay if needed
+        json_response=False,  # enables SSE streaming
+        stateless=False,
+    )
+    _stateful_session_auth_contexts: Dict[str, MCPAuthenticatedUser] = {}
+    _stateful_session_auth_context_last_seen: Dict[str, float] = {}
+
+    # Keep this alias so existing references to session_manager still work
+    session_manager = session_manager_stateless
+
     # Create SSE session manager
     sse_session_manager = StreamableHTTPSessionManager(
         app=server,
@@ -253,11 +267,42 @@
 
     # Context managers for proper lifecycle management
     _session_manager_cm = None
+    _session_manager_stateful_cm = None
     _sse_session_manager_cm = None
+    _stateful_auth_context_cleanup_task: Optional[asyncio.Task] = None
 
+    async def _purge_expired_stateful_session_auth_contexts(
+        now: Optional[float] = None,
+    ) -> None:
+        """Terminate expired stateful sessions and drop their auth contexts."""
+        now = now or time.monotonic()
+        server_instances = getattr(session_manager_stateful, "_server_instances", {})
+        expired_session_ids = [
+            session_id
+            for session_id, last_seen in _stateful_session_auth_context_last_seen.items()
+            if now - last_seen >= _STATEFUL_SESSION_IDLE_TIMEOUT_SECONDS
+            or session_id not in server_instances
+        ]
+
+        for session_id in expired_session_ids:
+            _stateful_session_auth_contexts.pop(session_id, None)
+            _stateful_session_auth_context_last_seen.pop(session_id, None)
+            transport = server_instances.pop(session_id, None)
+            if transport is not None:
+                await transport.terminate()
+
+        for session_id in list(_stateful_session_auth_context_last_seen):
+            if session_id not in _stateful_session_auth_contexts:
+                _stateful_session_auth_context_last_seen.pop(session_id, None)
+
+    async def _cleanup_expired_stateful_session_auth_contexts() -> None:
+        while True:
+            await asyncio.sleep(_STATEFUL_SESSION_IDLE_TIMEOUT_SECONDS)
+            await _purge_expired_stateful_session_auth_contexts()
+
     async def initialize_session_managers():
         """Initialize the session managers. Can be called from main app lifespan."""
-        global _SESSION_MANAGERS_INITIALIZED, _session_manager_cm, _sse_session_manager_cm
+        global _SESSION_MANAGERS_INITIALIZED, _session_manager_cm, _session_manager_stateful_cm, _sse_session_manager_cm, _stateful_auth_context_cleanup_task
 
         # Use async lock to prevent concurrent initialization
         async with _INITIALIZATION_LOCK:
@@ -267,12 +312,17 @@
             verbose_logger.info("Initializing MCP session managers...")
 
             # Start the session managers with context managers
-            _session_manager_cm = session_manager.run()
+            _session_manager_cm = session_manager_stateless.run()
+            _session_manager_stateful_cm = session_manager_stateful.run()
             _sse_session_manager_cm = sse_session_manager.run()
 
             # Enter the context managers
             await _session_manager_cm.__aenter__()
+            await _session_manager_stateful_cm.__aenter__()
             await _sse_session_manager_cm.__aenter__()
+            _stateful_auth_context_cleanup_task = asyncio.create_task(
+                _cleanup_expired_stateful_session_auth_contexts()
+            )
 
             _SESSION_MANAGERS_INITIALIZED = True
             verbose_logger.info(
@@ -281,21 +331,29 @@
 
     async def shutdown_session_managers():
         """Shutdown the session managers."""
-        global _SESSION_MANAGERS_INITIALIZED, _session_manager_cm, _sse_session_manager_cm
+        global _SESSION_MANAGERS_INITIALIZED, _session_manager_cm, _session_manager_stateful_cm, _sse_session_manager_cm, _stateful_auth_context_cleanup_task
 
         if _SESSION_MANAGERS_INITIALIZED:
             verbose_logger.info("Shutting down MCP session managers...")
 
             try:
+                if _stateful_auth_context_cleanup_task:
+                    _stateful_auth_context_cleanup_task.cancel()
+                    with contextlib.suppress(asyncio.CancelledError):
+                        await _stateful_auth_context_cleanup_task
                 if _session_manager_cm:
                     await _session_manager_cm.__aexit__(None, None, None)
+                if _session_manager_stateful_cm:
+                    await _session_manager_stateful_cm.__aexit__(None, None, None)
                 if _sse_session_manager_cm:
                     await _sse_session_manager_cm.__aexit__(None, None, None)
             except Exception as e:
                 verbose_logger.exception(f"Error during session manager shutdown: {e}")
 
             _session_manager_cm = None
+            _session_manager_stateful_cm = None
             _sse_session_manager_cm = None
+            _stateful_auth_context_cleanup_task = None
             _SESSION_MANAGERS_INITIALIZED = False
 
     @contextlib.asynccontextmanager
@@ -360,7 +418,7 @@
 
     @server.call_tool()
     async def mcp_server_tool_call(
-        name: str, arguments: Dict[str, Any] | None
+        name: str, arguments: Optional[Dict[str, Any]]
     ) -> CallToolResult:
         """
         Call a specific tool with the provided arguments
@@ -403,7 +461,7 @@
                 if host_token and hasattr(host_ctx, "session") and host_ctx.session:
                     host_session = host_ctx.session
 
-                    async def forward_progress(progress: float, total: float | None):
+                    async def forward_progress(progress: float, total: Optional[float]):
                         """Forward progress notifications from external MCP to Host"""
                         try:
                             await host_session.send_progress_notification(
@@ -545,7 +603,7 @@
 
     @server.get_prompt()
     async def get_prompt(
-        name: str, arguments: dict[str, str] | None
+        name: str, arguments: Optional[Dict[str, str]]
     ) -> GetPromptResult:
         """
         Get a specific prompt with the provided arguments
@@ -2579,6 +2637,61 @@
             raw_headers,
         )
 
+    def _get_session_id_from_scope(scope: Scope) -> Optional[str]:
+        """
+        Extract mcp-session-id from ASGI scope headers.
+        Returns None if not present.
+        """
+        for header_name, header_value in scope.get("headers", []):
+            name = (
+                header_name if isinstance(header_name, bytes) else header_name.encode()
+            )
+            if name.lower() == b"mcp-session-id":
+                return (
+                    header_value.decode()
+                    if isinstance(header_value, bytes)
+                    else str(header_value)
+                )
+        return None
+
+    def _is_initialize_request(body: bytes) -> bool:
+        """
+        Check if the request body is a JSON-RPC initialize method.
+        Returns True if method is "initialize", False otherwise or on parse error.
+        """
+        if not body:
+            return False
+        try:
+            data = json.loads(body)
+            return isinstance(data, dict) and data.get("method") == "initialize"
+        except (json.JSONDecodeError, TypeError):
+            return False
+
+    async def _read_request_body_for_routing(
+        receive: Receive,
+    ) -> Tuple[List[Message], bytes]:
+        """
+        Consume request body messages for routing, returning them for replay.
+        """
+        consumed_messages: List[Message] = []
+        body_chunks: List[bytes] = []
+
+        while True:
+            message = await receive()
+            consumed_messages.append(message)
+
+            if message.get("type") != "http.request":
+                break
+
+            body = message.get("body", b"") or b""
+            if body:
+                body_chunks.append(body)
+
+            if not message.get("more_body", False):
+                break
+
+        return consumed_messages, b"".join(body_chunks)
+
     async def _handle_stale_mcp_session(
         scope: Scope,
         receive: Receive,
@@ -2724,7 +2837,7 @@
             )
         return user_api_key_auth.model_copy(update={"object_permission": updated_op})
 
-    async def handle_streamable_http_mcp(
+    async def handle_streamable_http_mcp(  # noqa: PLR0915
         scope: Scope, receive: Receive, send: Send
     ) -> None:
         """Handle MCP requests through StreamableHTTP."""
@@ -2810,8 +2923,59 @@
             if _debug_headers:
                 send = MCPDebug.wrap_send_with_debug_headers(send, _debug_headers)
 
-            # Set the auth context variable for easy access in MCP functions
-            set_auth_context(
+            # Ensure session managers are initialized
+            if not _SESSION_MANAGERS_INITIALIZED:
+                await initialize_session_managers()
+                # Give it a moment to start up
+                await asyncio.sleep(0.1)
+
+            # Route based on mcp-session-id and request method:
+            # - Has session ID → stateful (Claude Code, Cursor, VSCode)
+            # - No session ID + initialize → stateful (so client gets mcp-session-id)
+            # - No session ID + other → stateless (curl, Inspector, Notion)
+            session_id = _get_session_id_from_scope(scope)
+            is_initialize = False
+            consumed_messages: List[Message] = []
+
+            # Handle stale session IDs before choosing a target manager. Stale
+            # non-DELETE requests have their session header stripped and should
+            # be routed as no-session requests.
+            if session_id:
+                handled = await _handle_stale_mcp_session(
+                    scope, receive, send, session_manager_stateful
+                )
+                if handled:
+                    # Request was fully handled (e.g., DELETE on non-existent session)
+                    return
+                session_id = _get_session_id_from_scope(scope)
+
+            if scope.get("method") == "POST" and not session_id:
+                consumed_messages, body = await _read_request_body_for_routing(receive)
+                is_initialize = _is_initialize_request(body)
+
+            use_stateful = bool(session_id or is_initialize)
+            target_manager = (
+                session_manager_stateful if use_stateful else session_manager_stateless
+            )
+
+            verbose_logger.debug(
+                f"MCP routing to {'stateful' if use_stateful else 'stateless'} manager"
+                + (f" (session={session_id[:8]}...)" if session_id else "")
+                + (" (initialize)" if is_initialize else "")
+            )
+
+            # Replay body messages if we consumed them for peeking
+            original_receive = receive
+            if consumed_messages:
+
+                async def wrapped_receive():
+                    if consumed_messages:
+                        return consumed_messages.pop(0)
+                    return await original_receive()
+
+                receive = wrapped_receive
+
+            auth_user = _set_or_update_auth_context(
                 user_api_key_auth=user_api_key_auth,
                 mcp_auth_header=mcp_auth_header,
                 mcp_servers=mcp_servers,
@@ -2819,29 +2983,22 @@
                 oauth2_headers=oauth2_headers,
                 raw_headers=raw_headers,
                 client_ip=_client_ip,
+                session_id=session_id if use_stateful else None,
             )
+            if use_stateful and is_initialize:
+                send = _wrap_send_with_stateful_session_auth_context(send, auth_user)
 
-            # Ensure session managers are initialized
-            if not _SESSION_MANAGERS_INITIALIZED:
-                await initialize_session_managers()
-                # Give it a moment to start up
-                await asyncio.sleep(0.1)
-
-            # Handle stale session IDs - either strip them for reconnection
-            # or return success for idempotent DELETE operations
-            handled = await _handle_stale_mcp_session(
-                scope, receive, send, session_manager
-            )
-            if handled:
-                # Request was fully handled (e.g., DELETE on non-existent session)
-                return
-
             async with _gateway_initialize_instructions_request_scope(
                 user_api_key_auth,
                 mcp_servers,
                 _client_ip,
             ):
-                await session_manager.handle_request(scope, receive, send)
+                try:
+                    await target_manager.handle_request(scope, receive, send)
+                finally:
+                    if use_stateful and session_id and scope.get("method") == "DELETE":
+                        _stateful_session_auth_contexts.pop(session_id, None)
+                        _stateful_session_auth_context_last_seen.pop(session_id, None)
         except HTTPException:
             # Re-raise HTTP exceptions to preserve status codes and details
             raise
@@ -2955,15 +3112,33 @@
     ############ Auth Context Functions ####################
     ########################################################
 
+    def _update_auth_context(
+        auth_user: MCPAuthenticatedUser,
+        user_api_key_auth: Optional[UserAPIKeyAuth],
+        mcp_auth_header: Optional[str] = None,
+        mcp_servers: Optional[List[str]] = None,
+        mcp_server_auth_headers: Optional[Dict[str, Dict[str, str]]] = None,
+        oauth2_headers: Optional[Dict[str, str]] = None,
+        raw_headers: Optional[Dict[str, str]] = None,
+        client_ip: Optional[str] = None,
+    ) -> None:
+        auth_user.user_api_key_auth = user_api_key_auth
+        auth_user.mcp_auth_header = mcp_auth_header
+        auth_user.mcp_servers = mcp_servers
+        auth_user.mcp_server_auth_headers = mcp_server_auth_headers or {}
+        auth_user.oauth2_headers = oauth2_headers
+        auth_user.raw_headers = raw_headers
+        auth_user.client_ip = client_ip
+
     def set_auth_context(
-        user_api_key_auth: UserAPIKeyAuth,
+        user_api_key_auth: Optional[UserAPIKeyAuth],
         mcp_auth_header: Optional[str] = None,
         mcp_servers: Optional[List[str]] = None,
         mcp_server_auth_headers: Optional[Dict[str, Dict[str, str]]] = None,
         oauth2_headers: Optional[Dict[str, str]] = None,
         raw_headers: Optional[Dict[str, str]] = None,
         client_ip: Optional[str] = None,
-    ) -> None:
+    ) -> MCPAuthenticatedUser:
         """
         Set the UserAPIKeyAuth in the auth context variable.
 
@@ -2984,7 +3159,63 @@
             client_ip=client_ip,
         )
         auth_context_var.set(auth_user)
+        return auth_user
 
+    def _set_or_update_auth_context(
+        user_api_key_auth: Optional[UserAPIKeyAuth],
+        mcp_auth_header: Optional[str] = None,
+        mcp_servers: Optional[List[str]] = None,
+        mcp_server_auth_headers: Optional[Dict[str, Dict[str, str]]] = None,
+        oauth2_headers: Optional[Dict[str, str]] = None,
+        raw_headers: Optional[Dict[str, str]] = None,
+        client_ip: Optional[str] = None,
+        session_id: Optional[str] = None,
+    ) -> MCPAuthenticatedUser:
+        auth_user = (
+            _stateful_session_auth_contexts.get(session_id) if session_id else None
+        )
+        if auth_user is not None and session_id is not None:
+            _stateful_session_auth_context_last_seen[session_id] = time.monotonic()
+            _update_auth_context(
+                auth_user=auth_user,
+                user_api_key_auth=user_api_key_auth,
+                mcp_auth_header=mcp_auth_header,
+                mcp_servers=mcp_servers,
+                mcp_server_auth_headers=mcp_server_auth_headers,
+                oauth2_headers=oauth2_headers,
+                raw_headers=raw_headers,
+                client_ip=client_ip,
+            )
+            auth_context_var.set(auth_user)
+            return auth_user
+        return set_auth_context(
+            user_api_key_auth=user_api_key_auth,
+            mcp_auth_header=mcp_auth_header,
+            mcp_servers=mcp_servers,
+            mcp_server_auth_headers=mcp_server_auth_headers,
+            oauth2_headers=oauth2_headers,
+            raw_headers=raw_headers,
+            client_ip=client_ip,
+        )
+
+    def _wrap_send_with_stateful_session_auth_context(
+        send: Send,
+        auth_user: MCPAuthenticatedUser,
+    ) -> Send:
+        async def wrapped_send(message: Message) -> None:
+            if message.get("type") == "http.response.start":
+                for key, value in message.get("headers", []):
+                    if key.lower() == b"mcp-session-id":
+                        session_id = value.decode()
+                        _stateful_session_auth_contexts[session_id] = auth_user
+                        _stateful_session_auth_context_last_seen[session_id] = (
+                            time.monotonic()
+                        )
+                        break
+            await send(message)
+
+        return wrapped_send
+
     def get_auth_context() -> Tuple[
         Optional[UserAPIKeyAuth],
         Optional[str],

diff --git a/tests/mcp_tests/test_mcp_server.py b/tests/mcp_tests/test_mcp_server.py
--- a/tests/mcp_tests/test_mcp_server.py
+++ b/tests/mcp_tests/test_mcp_server.py
@@ -395,12 +395,12 @@
 @pytest.mark.asyncio
 async def test_streamable_http_mcp_handler_mock():
     """Test the streamable HTTP MCP handler functionality"""
-    from litellm.proxy._types import UserAPIKeyAuth
+    # Mock streamable HTTP session managers and their methods
+    mock_session_manager_stateless = AsyncMock()
+    mock_session_manager_stateless.handle_request = AsyncMock()
+    mock_session_manager_stateful = AsyncMock()
+    mock_session_manager_stateful.handle_request = AsyncMock()
 
-    # Mock the session manager and its methods
-    mock_session_manager = AsyncMock()
-    mock_session_manager.handle_request = AsyncMock()
-
     # Mock scope, receive, send with proper ASGI scope format
     mock_scope = {
         "type": "http",
@@ -411,7 +411,7 @@
         "server": ("localhost", 8000),
         "scheme": "http",
     }
-    mock_receive = AsyncMock()
+    mock_receive = AsyncMock(return_value={"body": b"{}", "more_body": False})
     mock_send = AsyncMock()
 
     # Mock extract_mcp_auth_context to bypass auth checks in the handler
@@ -423,10 +423,14 @@
             True,
         ),
         patch(
-            "litellm.proxy._experimental.mcp_server.server.session_manager",
-            mock_session_manager,
+            "litellm.proxy._experimental.mcp_server.server.session_manager_stateless",
+            mock_session_manager_stateless,
         ),
         patch(
+            "litellm.proxy._experimental.mcp_server.server.session_manager_stateful",
+            mock_session_manager_stateful,
+        ),
+        patch(
             "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
             AsyncMock(return_value=mock_auth_context),
         ),
@@ -441,8 +445,9 @@
         # Call the handler
         await handle_streamable_http_mcp(mock_scope, mock_receive, mock_send)
 
-        # Verify session manager handle_request was called
-        mock_session_manager.handle_request.assert_called_once()
+        # Verify stateless session manager handle_request was called
+        mock_session_manager_stateless.handle_request.assert_called_once()
+        mock_session_manager_stateful.handle_request.assert_not_called()
 
 
 @pytest.mark.asyncio

diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py b/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py
--- a/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py
@@ -1520,10 +1520,14 @@
         )
 
     monkeypatch.setattr(
-        "litellm.proxy._experimental.mcp_server.server.session_manager",
+        "litellm.proxy._experimental.mcp_server.server.session_manager_stateless",
         MagicMock(handle_request=dummy_handle_request),
     )
     monkeypatch.setattr(
+        "litellm.proxy._experimental.mcp_server.server.session_manager_stateful",
+        MagicMock(handle_request=dummy_handle_request),
+    )
+    monkeypatch.setattr(
         "litellm.proxy._experimental.mcp_server.server.initialize_session_managers",
         AsyncMock(),
     )

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
--- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
+++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
@@ -1,4 +1,5 @@
 import asyncio
+import contextvars
 from datetime import datetime, timedelta
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -1000,19 +1001,24 @@
     # Reset state before test
     original_initialized = mcp_server._SESSION_MANAGERS_INITIALIZED
     original_session_cm = mcp_server._session_manager_cm
+    original_session_stateful_cm = mcp_server._session_manager_stateful_cm
     original_sse_session_cm = mcp_server._sse_session_manager_cm
 
     try:
         mcp_server._SESSION_MANAGERS_INITIALIZED = False
         mcp_server._session_manager_cm = None
+        mcp_server._session_manager_stateful_cm = None
         mcp_server._sse_session_manager_cm = None
 
         # Mock the session managers to avoid actual MCP initialization
         with (
             patch(
-                "litellm.proxy._experimental.mcp_server.server.session_manager"
-            ) as mock_session_manager,
+                "litellm.proxy._experimental.mcp_server.server.session_manager_stateless"
+            ) as mock_session_manager_stateless,
             patch(
+                "litellm.proxy._experimental.mcp_server.server.session_manager_stateful"
+            ) as mock_session_manager_stateful,
+            patch(
                 "litellm.proxy._experimental.mcp_server.server.sse_session_manager"
             ) as mock_sse_session_manager,
             patch("litellm.proxy._experimental.mcp_server.server.verbose_logger"),
@@ -1022,7 +1028,8 @@
             mock_cm.__aenter__ = AsyncMock()
             mock_cm.__aexit__ = AsyncMock()
 
-            mock_session_manager.run.return_value = mock_cm
+            mock_session_manager_stateless.run.return_value = mock_cm
+            mock_session_manager_stateful.run.return_value = mock_cm
             mock_sse_session_manager.run.return_value = mock_cm
 
             # Create multiple concurrent tasks that call initialize_session_managers
@@ -1039,18 +1046,21 @@
                 result == "success" for result in results
             ), f"Some tasks failed: {results}"
 
-            # session_manager.run() should only be called once due to the lock
+            # Each session manager.run() should only be called once due to the lock
             assert (
-                mock_session_manager.run.call_count == 1
-            ), f"Expected 1 call to session_manager.run(), got {mock_session_manager.run.call_count}"
+                mock_session_manager_stateless.run.call_count == 1
+            ), f"Expected 1 call to session_manager_stateless.run(), got {mock_session_manager_stateless.run.call_count}"
             assert (
+                mock_session_manager_stateful.run.call_count == 1
+            ), f"Expected 1 call to session_manager_stateful.run(), got {mock_session_manager_stateful.run.call_count}"
+            assert (
                 mock_sse_session_manager.run.call_count == 1
             ), f"Expected 1 call to sse_session_manager.run(), got {mock_sse_session_manager.run.call_count}"
 
-            # The context managers should only be entered once each
+            # The context managers should only be entered once each (3 managers)
             assert (
-                mock_cm.__aenter__.call_count == 2
-            ), f"Expected 2 calls to __aenter__ (one for each session manager), got {mock_cm.__aenter__.call_count}"
+                mock_cm.__aenter__.call_count == 3
+            ), f"Expected 3 calls to __aenter__ (one per session manager), got {mock_cm.__aenter__.call_count}"
 
             # State should be properly set
             assert mcp_server._SESSION_MANAGERS_INITIALIZED is True
@@ -1059,34 +1069,355 @@
         # Restore original state
         mcp_server._SESSION_MANAGERS_INITIALIZED = original_initialized
         mcp_server._session_manager_cm = original_session_cm
+        mcp_server._session_manager_stateful_cm = original_session_stateful_cm
         mcp_server._sse_session_manager_cm = original_sse_session_cm
 
 
 @pytest.mark.asyncio
 async def test_streamable_http_session_manager_is_stateless():
     """
-    Test that the StreamableHTTPSessionManager is initialized with stateless=True.
+    Test that the StreamableHTTPSessionManager is initialized with both stateless and stateful managers.
 
     Regression test for GitHub issue #20242 / PR #19809.
     When stateless=False, the mcp library rejects non-initialize requests
     that lack an mcp-session-id header, breaking clients like MCP Inspector,
     curl, and any HTTP client without automatic session management.
+    
+    Now we support both:
+    - stateless manager for clients without session IDs (curl, Inspector)
+    - stateful manager for clients with session IDs (Claude Code, Cursor, VSCode)
     """
     try:
-        from litellm.proxy._experimental.mcp_server.server import session_manager
+        from litellm.proxy._experimental.mcp_server.server import (
+            session_manager_stateful,
+            session_manager_stateless,
+        )
     except ImportError:
         pytest.skip("MCP server not available")
 
-    # The session manager must be stateless to avoid requiring mcp-session-id
+    # The stateless session manager must be stateless to avoid requiring mcp-session-id
     # on every request. This was regressed by PR #19809 (stateless=True -> False).
-    assert session_manager.stateless is True, (
-        "StreamableHTTPSessionManager must be initialized with stateless=True. "
+    assert session_manager_stateless.stateless is True, (
+        "session_manager_stateless must be initialized with stateless=True. "
         "stateless=False breaks MCP clients that don't manage session IDs. "
         "See: https://github.com/BerriAI/litellm/issues/20242"
     )
 
+    # The stateful session manager must be stateful to support progress notifications
+    assert session_manager_stateful.stateless is False, (
+        "session_manager_stateful must be initialized with stateless=False. "
+        "stateless=True breaks progress notifications for clients that manage session IDs."
+    )
 
+
 @pytest.mark.asyncio
+async def test_mcp_routing_initialize_to_stateful_no_session_to_stateless():
+    """
+    Test that routing correctly sends:
+    - initialize (no mcp-session-id) → stateful manager (so client gets mcp-session-id)
+    - tools/list (no mcp-session-id) → stateless manager (curl, Inspector)
+    """
+    try:
+        from litellm.proxy._experimental.mcp_server.server import (
+            handle_streamable_http_mcp,
+            session_manager_stateful,
+            session_manager_stateless,
+        )
+    except ImportError:
+        pytest.skip("MCP server not available")
+
+    async def make_request(method_body: bytes, path: str = "/mcp/progress_test"):
+        scope = {
+            "type": "http",
+            "method": "POST",
+            "path": path,
+            "headers": [
+                (b"content-type", b"application/json"),
+                (b"authorization", b"Bearer test-key"),
+            ],
+        }
+        receive = AsyncMock(
+            return_value={
+                "type": "http.request",
+                "body": method_body,
+                "more_body": False,
+            }
+        )
+        send = AsyncMock()
+
+        stateless_called = []
+        stateful_called = []
+
+        async def stateless_handle(s, r, se):
+            stateless_called.append(1)
+
+        async def stateful_handle(s, r, se):
+            stateful_called.append(1)
+
+        with (
+            patch(
+                "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
+                new_callable=AsyncMock,
+                return_value=(MagicMock(), None, ["progress_test"], None, None, None),
+            ),
+            patch(
+                "litellm.proxy._experimental.mcp_server.server.set_auth_context",
+            ),
+            patch(
+                "litellm.proxy._experimental.mcp_server.server._SESSION_MANAGERS_INITIALIZED",
+                True,
+            ),
+            patch.object(
+                session_manager_stateless,
+                "handle_request",
+                side_effect=stateless_handle,
+            ),
+            patch.object(
+                session_manager_stateful,
+                "handle_request",
+                side_effect=stateful_handle,
+            ),
+            patch.object(
+                session_manager_stateless,
+                "_server_instances",
+                {},
+            ),
+            patch.object(
+                session_manager_stateful,
+                "_server_instances",
+                {},
+            ),
+        ):
+            await handle_streamable_http_mcp(scope, receive, send)
+
+        return bool(stateless_called), bool(stateful_called)
+
+    # initialize → stateful
+    init_body = b'{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05"}}'
+    stateless_called, stateful_called = await make_request(init_body)
+    assert (
+        stateful_called and not stateless_called
+    ), "initialize (no session) should route to stateful, not stateless"
+
+    # tools/list → stateless
+    tools_body = b'{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}'
+    stateless_called, stateful_called = await make_request(tools_body)
+    assert (
+        stateless_called and not stateful_called
+    ), "tools/list (no session) should route to stateless, not stateful"
+
+
+@pytest.mark.asyncio
+async def test_mcp_routing_chunked_initialize_to_stateful():
+    """
+    Test that chunked initialize requests route to the stateful manager.
+    """
+    try:
+        from litellm.proxy._experimental.mcp_server.server import (
+            handle_streamable_http_mcp,
+            session_manager_stateful,
+            session_manager_stateless,
+        )
+    except ImportError:
+        pytest.skip("MCP server not available")
+
+    scope = {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp/progress_test",
+        "headers": [
+            (b"content-type", b"application/json"),
+            (b"authorization", b"Bearer test-key"),
+        ],
+    }
+    messages = [
+        {
+            "type": "http.request",
+            "body": b'{"jsonrpc":"2.0","id":1,',
+            "more_body": True,
+        },
+        {
+            "type": "http.request",
+            "body": b'"method":"initialize","params":{}}',
+            "more_body": False,
+        },
+    ]
+    receive = AsyncMock(side_effect=messages)
+    send = AsyncMock()
+    stateless_called = []
+    stateful_called = []
+
+    async def stateless_handle(s, r, se):
+        stateless_called.append(1)
+
+    async def stateful_handle(s, r, se):
+        stateful_called.append(1)
+
+    with patch(
+        "litellm.proxy._experimental.mcp_server.server.extract_mcp_auth_context",
+        new_callable=AsyncMock,
+        return_value=(MagicMock(), None, ["progress_test"], None, None, None),
... diff truncated: showing 800 of 1080 lines

_{You can send follow-ups to the cloud agent here.}

[cursor]

Cursor Bugbot has reviewed your changes and found 2 potential issues.

^{Bugbot Autofix is ON, but it could not run because the spend limit has been reached. To enable Bugbot Autofix, have a team admin raise the spend limit in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 78d04f6. Configure here.}

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[Sameerlite]

Is this good to merge @mateo-berri ?

[mateo-berri]

1. Add scope["client"] socket peer fallback before "anonymous" in _owner_fingerprint_for (line 2811). This would make the "anonymous" sentinel from "best-effort fallback" -> "essentially unreachable"

2. Add a per-owner stateful session cap (+ global cap) in routing path before initialize reaches stateful manager. This'd addresses veria issue where authed caller can spam initialize -> 30-min retention each. Even if not malicious but accidental -> can lead to perf problems for themselves + others

3. Please rate-limit initialize calls per owner (to e.g. 10/min). Caps protect against total count, and rate-limit protects against burst exhaustion before LRU eviction catches up

[mateo-berri]

Also, @Sameerlite I watched your video proof of it working: https://www.loom.com/share/1a21e2458e4b4bbeaaaf7519a7f947dc, and I don't see anything but the static dashboard. I think you shared the wrong screen/window

[ryan-crabbe-berri]

can we support stateful as a default instead? so that we dont have conditional logic to go between stateful and stateless?

[Sameerlite]

can we support stateful as a default instead? so that we dont have conditional logic to go between stateful and stateless?

Issue with this is that if a mcp is a stateless mcp server gets this , it might cause an error

[Sameerlite]

@mateo-berri updated video: https://www.loom.com/share/4f1a63a4812a4855b2daa25b68af1f78

[mateo-berri]

This PR might be eventually made obsolete with: https://blog.modelcontextprotocol.io/posts/2026-07-28-release-candidate/, which gets rid of protocol-level sessions entirely. I'm happy to land this as a short-lived bridge for today's stateful clients (Claude Code/Cursor/VSCode), though, and then rip it out as RC gets finalized

[Sameerlite]

@mateo-berri that will helpful thanks!

[mateo-berri]

@mateo-berri updated video: https://www.loom.com/share/4f1a63a4812a4855b2daa25b68af1f78

@Sameerlite Thanks for the video but it says unavailable for me:

Can you check? Also, did you address my comments?

[Sameerlite]

Here: https://www.loom.com/share/55f6901a2f664fc7b02a396ad4e4e96f

[Sameerlite]

As mentioned on slack, i will add the comments in a new PR as I need to do more work on this feat anyways

[mateo-berri]

bugbot run

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[mateo-berri]

@greptileai

[mateo-berri]

LGTM; thanks!