fix(mcp): stream SSE responses, fix mount order, remove OAuth root pollution

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -132,20 +132,11 @@ def _resolve_oauth2_server_for_root_endpoints(
     """
     Resolve the MCP server for root-level OAuth endpoints (no server name in path).
 
-    When the MCP SDK hits root-level endpoints like /register, /authorize, /token
-    without a server name prefix, we try to find the right server automatically.
-    Returns the server if exactly one OAuth2 server is configured, else None.
+    Always returns None. Root-level OAuth discovery endpoints should not
+    auto-resolve to an arbitrary server because doing so pollutes non-OAuth
+    servers' discovery responses when any single OAuth2 server is configured.
+    Clients should use server-specific paths instead (e.g. /{server_name}/authorize).
     """
-    from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
-        global_mcp_server_manager,
-    )
-
-    registry = global_mcp_server_manager.get_filtered_registry(client_ip=client_ip)
-    oauth2_servers = [
-        s for s in registry.values() if s.auth_type == MCPAuth.oauth2
-    ]
-    if len(oauth2_servers) == 1:
-        return oauth2_servers[0]
     return None
 
 

litellm/proxy/_experimental/mcp_server/server.py

@@ -2149,10 +2149,9 @@ def get_mcp_server_enabled() -> Dict[str, bool]:
         return {"enabled": MCP_AVAILABLE}
 
     # Mount the MCP handlers
-    app.mount("/", handle_streamable_http_mcp)
-    app.mount("/mcp", handle_streamable_http_mcp)
-    app.mount("/{mcp_server_name}/mcp", handle_streamable_http_mcp)
+    # /sse must be mounted before the "/" catch-all so it's matched first
     app.mount("/sse", handle_sse_mcp)
+    app.mount("/", handle_streamable_http_mcp)
     app.add_middleware(AuthContextMiddleware)
 
     ########################################################

litellm/proxy/proxy_server.py

@@ -12917,7 +12917,6 @@ async def dynamic_mcp_route(mcp_server_name: str, request: Request):
             global_mcp_server_manager,
         )
         from litellm.proxy.auth.ip_address_utils import IPAddressUtils
-        from litellm.types.mcp import MCPAuth
 
         client_ip = IPAddressUtils.get_mcp_client_ip(request)
         mcp_server = global_mcp_server_manager.get_mcp_server_by_name(
@@ -12938,35 +12937,75 @@ async def dynamic_mcp_route(mcp_server_name: str, request: Request):
             handle_streamable_http_mcp,
         )
 
-        # Create a custom send function to capture the response
-        response_started = False
-        response_body = b""
-        response_status = 200
-        response_headers = []
+        # Stream the ASGI response instead of buffering it. This is critical for
+        # SSE (text/event-stream) responses used by the MCP Streamable HTTP
+        # transport — buffering would break incremental event delivery.
+        response_meta: dict = {}
+        headers_ready = asyncio.Event()
+        body_queue: asyncio.Queue[bytes | None] = asyncio.Queue()
+        handler_error: list = []
 
-        async def custom_send(message):
-            nonlocal response_started, response_body, response_status, response_headers
+        async def streaming_send(message):
             if message["type"] == "http.response.start":
-                response_started = True
-                response_status = message["status"]
-                response_headers = message.get("headers", [])
+                response_meta["status"] = message["status"]
+                response_meta["headers"] = {
+                    k.decode(): v.decode()
+                    for k, v in message.get("headers", [])
+                }
+                headers_ready.set()
             elif message["type"] == "http.response.body":
-                response_body += message.get("body", b"")
+                chunk = message.get("body", b"")
+                if chunk:
+                    await body_queue.put(chunk)
+                if not message.get("more_body", False):
+                    await body_queue.put(None)  # sentinel
 
-        # Call the existing MCP handler
-        await handle_streamable_http_mcp(
-            scope, receive=request.receive, send=custom_send
-        )
+        async def run_handler():
+            try:
+                await handle_streamable_http_mcp(
+                    scope, receive=request.receive, send=streaming_send
+                )
+            except Exception as exc:
+                handler_error.append(exc)
+            finally:
+                # Ensure consumers aren't stuck waiting if the handler exits
+                # without sending a complete response.
+                headers_ready.set()
+                await body_queue.put(None)
+
+        handler_task = asyncio.create_task(run_handler())
+
+        # Wait for the ASGI handler to send http.response.start
+        await headers_ready.wait()
 
-        # Return the response
-        from starlette.responses import Response
+        # If the handler errored before sending headers, raise
+        if handler_error and "status" not in response_meta:
+            await handler_task
+            raise handler_error[0]
 
-        headers_dict = {k.decode(): v.decode() for k, v in response_headers}
-        return Response(
-            content=response_body,
-            status_code=response_status,
-            headers=headers_dict,
-            media_type=headers_dict.get("content-type", "application/json"),
+        async def body_generator():
+            try:
+                while True:
+                    chunk = await body_queue.get()
+                    if chunk is None:
+                        break
+                    yield chunk
+            finally:
+                if not handler_task.done():
+                    handler_task.cancel()
+                    try:
+                        await handler_task
+                    except asyncio.CancelledError:
+                        pass
+
+        headers = response_meta.get("headers", {})
+        media_type = headers.pop("content-type", "application/json")
+
+        return StreamingResponse(
+            content=body_generator(),
+            status_code=response_meta.get("status", 200),
+            headers=headers,
+            media_type=media_type,
         )
 
     except HTTPException as e:

tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py

@@ -1258,8 +1258,12 @@ def _create_oauth2_server(
 
 
 @pytest.mark.asyncio
-async def test_authorize_root_resolves_single_oauth2_server():
-    """When /authorize is hit without server name and exactly 1 OAuth2 server exists, resolve it."""
+async def test_authorize_root_returns_404_without_server_name():
+    """When /authorize is hit without server name, return 404 even if 1 OAuth2 server exists.
+
+    Root auto-resolution was removed to prevent OAuth discovery pollution across
+    unrelated MCP servers.
+    """
     try:
         from fastapi import Request
 
@@ -1281,25 +1285,16 @@ async def test_authorize_root_resolves_single_oauth2_server():
     mock_request.headers = {}
 
     try:
-        with patch(
-            "litellm.proxy._experimental.mcp_server.discoverable_endpoints.encrypt_value_helper"
-        ) as mock_encrypt:
-            mock_encrypt.return_value = "mocked_encrypted_state"
-
-            # Call /authorize WITHOUT mcp_server_name, with dummy_client as client_id
-            response = await authorize(
+        with pytest.raises(HTTPException) as exc_info:
+            await authorize(
                 request=mock_request,
                 client_id="dummy_client",
                 mcp_server_name=None,
                 redirect_uri="http://localhost:62646/callback",
                 state="test_state",
             )
-
-        # Should resolve to the single OAuth2 server and redirect
-        assert response.status_code == 307
-        location = response.headers["location"]
-        assert "https://provider.com/oauth/authorize" in location
-        assert "client_id=test_client_id" in location
+        assert exc_info.value.status_code == 404
+        assert "MCP server not found" in str(exc_info.value.detail)
     finally:
         global_mcp_server_manager.registry.clear()
 
@@ -1349,8 +1344,12 @@ async def test_authorize_root_fails_with_multiple_oauth2_servers():
 
 
 @pytest.mark.asyncio
-async def test_token_root_resolves_single_oauth2_server():
-    """When /token is hit without server name and exactly 1 OAuth2 server exists, resolve it."""
+async def test_token_root_returns_404_without_server_name():
+    """When /token is hit without server name, return 404 even if 1 OAuth2 server exists.
+
+    Root auto-resolution was removed to prevent OAuth discovery pollution across
+    unrelated MCP servers.
+    """
     try:
         from fastapi import Request
 
@@ -1371,25 +1370,9 @@ async def test_token_root_resolves_single_oauth2_server():
     mock_request.base_url = "https://llm.example.com/"
     mock_request.headers = {}
 
-    mock_response = MagicMock()
-    mock_response.json.return_value = {
-        "access_token": "ya29.test_token",
-        "token_type": "Bearer",
-        "expires_in": 3599,
-    }
-    mock_response.raise_for_status = MagicMock()
-
-    mock_async_client = MagicMock()
-    mock_async_client.post = AsyncMock(return_value=mock_response)
-
     try:
-        with patch(
-            "litellm.proxy._experimental.mcp_server.discoverable_endpoints.get_async_httpx_client"
-        ) as mock_get_client:
-            mock_get_client.return_value = mock_async_client
-
-            # Call /token WITHOUT mcp_server_name
-            response = await token_endpoint(
+        with pytest.raises(HTTPException) as exc_info:
+            await token_endpoint(
                 request=mock_request,
                 grant_type="authorization_code",
                 code="test_auth_code",
@@ -1399,23 +1382,20 @@ async def test_token_root_resolves_single_oauth2_server():
                 client_secret=None,
                 code_verifier="test_verifier",
             )
-
-        # Should resolve and exchange token with the upstream server
-        import json
-
-        token_data = json.loads(response.body)
-        assert token_data["access_token"] == "ya29.test_token"
-
-        # Verify it called the correct upstream token URL
-        call_args = mock_async_client.post.call_args
-        assert call_args.args[0] == "https://provider.com/oauth/token"
+        assert exc_info.value.status_code == 404
+        assert "MCP server not found" in str(exc_info.value.detail)
     finally:
         global_mcp_server_manager.registry.clear()
 
 
 @pytest.mark.asyncio
-async def test_register_root_resolves_single_oauth2_server():
-    """When /register is hit without server name and exactly 1 OAuth2 server exists, resolve it."""
+async def test_register_root_returns_dummy_without_server_name():
+    """When /register is hit without server name, return dummy registration even if 1 OAuth2 server exists.
+
+    Root auto-resolution was removed to prevent OAuth discovery pollution across
+    unrelated MCP servers. The register endpoint returns a dummy response when
+    no server can be resolved.
+    """
     try:
         from fastapi import Request
 
@@ -1443,16 +1423,19 @@ async def test_register_root_resolves_single_oauth2_server():
         ):
             result = await register_client(request=mock_request, mcp_server_name=None)
 
-        # Should resolve to the single server and return its name as client_id
-        assert result["client_id"] == "test_oauth"
-        assert "redirect_uris" in result
+        # Should return dummy registration since no server is resolved
+        assert result["client_id"] == "dummy_client"
     finally:
         global_mcp_server_manager.registry.clear()
 
 
 @pytest.mark.asyncio
-async def test_discovery_root_includes_server_name_prefix():
-    """When root discovery is hit and exactly 1 OAuth2 server exists, include server name in URLs."""
+async def test_discovery_root_returns_generic_urls_without_server_name():
+    """When root discovery is hit without server name, return generic URLs without server prefix.
+
+    Root auto-resolution was removed to prevent OAuth discovery pollution across
+    unrelated MCP servers.
+    """
     try:
         from fastapi import Request
 
@@ -1480,11 +1463,11 @@ async def test_discovery_root_includes_server_name_prefix():
             mcp_server_name=None,
         )
 
-        # Should resolve to the single server and include its name in endpoint URLs
-        assert "/test_oauth/authorize" in response["authorization_endpoint"]
-        assert "/test_oauth/token" in response["token_endpoint"]
-        assert "/test_oauth/register" in response["registration_endpoint"]
-        assert response["scopes_supported"] == ["read", "write"]
+        # Should NOT resolve to the single server — return generic URLs instead
+        assert response["authorization_endpoint"] == "https://llm.example.com/authorize"
+        assert response["token_endpoint"] == "https://llm.example.com/token"
+        assert response["registration_endpoint"] == "https://llm.example.com/register"
+        assert response["scopes_supported"] == []
     finally:
         global_mcp_server_manager.registry.clear()