diff --git a/litellm/experimental_mcp_client/client.py b/litellm/experimental_mcp_client/client.py index 5e5dd3cf3f9..fa4d76ecbed 100644 --- a/litellm/experimental_mcp_client/client.py +++ b/litellm/experimental_mcp_client/client.py @@ -14,18 +14,16 @@ from types import MappingProxyType from typing import Any, Final, TypeAlias, TypeVar import httpx2 -from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream from mcp import ClientSession, MCPError, ReadResourceResult, Resource, StdioServerParameters from mcp.client.sse import sse_client from mcp.client.stdio import stdio_client from mcp.client.streamable_http import streamable_http_client +from mcp.shared._stream_protocols import ReadStream, WriteStream from mcp.shared.message import SessionMessage -from typing_extensions import Unpack _TransportStreams: TypeAlias = tuple[ - MemoryObjectReceiveStream[SessionMessage | Exception], - MemoryObjectSendStream[SessionMessage], - Unpack[tuple[object, ...]], + ReadStream[SessionMessage | Exception], + WriteStream[SessionMessage], ] _TransportContext: TypeAlias = AbstractAsyncContextManager[_TransportStreams] @@ -320,7 +318,9 @@ class MCPClient: async def prepare_request_auth(self) -> httpx2.Request: """Preview the authenticated request without sending it, closing the auth flow afterwards.""" - request: Final = httpx2.Request("POST", self.server_url or "http://localhost/", headers=self._get_auth_headers()) + request: Final = httpx2.Request( + "POST", self.server_url or "http://localhost/", headers=self._get_auth_headers() + ) if self._resolved_auth is None: return request flow: Final = self._resolved_auth.async_auth_flow(request) @@ -441,7 +441,8 @@ class MCPClient: transport: Final = await transport_ctx.__aenter__() in_flight_error: BaseException | None = None try: - read_stream, write_stream = transport[0], transport[1] + read_stream: Final = transport[0] + write_stream: Final = transport[1] stream_error: Final[asyncio.Future[Exception]] = asyncio.get_running_loop().create_future() async def receive_message( @@ -917,7 +918,7 @@ class MCPClient: async def _list_resource_templates_operation(session: ClientSession) -> ListResourceTemplatesResult: capabilities: Final = session.server_capabilities if capabilities is not None and capabilities.resources is None: - return ListResourceTemplatesResult(resource_templates=[]) + return ListResourceTemplatesResult(resource_templates=[]) # mutable-ok: MCP result payload try: return await session.list_resource_templates() except MCPError as error: @@ -926,7 +927,7 @@ class MCPClient: verbose_logger.debug( "MCP client list_resource_templates is unsupported by %s: %s", self.server_url or "stdio", error ) - return ListResourceTemplatesResult(resource_templates=[]) + return ListResourceTemplatesResult(resource_templates=[]) # mutable-ok: MCP result payload try: result: Final = await self.run_with_session(_list_resource_templates_operation) diff --git a/litellm/proxy/_experimental/mcp_server/elicitation_handler.py b/litellm/proxy/_experimental/mcp_server/elicitation_handler.py index 57d2d86d506..6155f1f215c 100644 --- a/litellm/proxy/_experimental/mcp_server/elicitation_handler.py +++ b/litellm/proxy/_experimental/mcp_server/elicitation_handler.py @@ -160,7 +160,7 @@ async def _relay_elicitation_to_downstream( verbose_logger.info("MCP elicitation: relaying generic elicitation to downstream") result = await downstream_session.elicit( message=getattr(params, "message", ""), - requested_schema=getattr(params, "requested_schema", {}), + requested_schema=getattr(params, "requested_schema", {}), # mutable-ok: elicitation default schema ) verbose_logger.info( "MCP elicitation: downstream responded with action=%s", diff --git a/litellm/proxy/_experimental/mcp_server/guardrail_translation/handler.py b/litellm/proxy/_experimental/mcp_server/guardrail_translation/handler.py index 01c8e73cad3..08a5d2b4135 100644 --- a/litellm/proxy/_experimental/mcp_server/guardrail_translation/handler.py +++ b/litellm/proxy/_experimental/mcp_server/guardrail_translation/handler.py @@ -135,7 +135,7 @@ class MCPGuardrailTranslationHandler(BaseTranslation): mcp_tool: Final = MCPTool( name=mcp_tool_name, description=mcp_tool_description or "", - input_schema={}, # Call payload has no schema; guardrail gets args from request_data + input_schema={}, # mutable-ok: call payload has no schema; guardrail gets args from request_data ) openai_tool: Final = transform_mcp_tool_to_openai_tool(mcp_tool) fn: Final = openai_tool["function"] diff --git a/litellm/proxy/_experimental/mcp_server/mcp_context.py b/litellm/proxy/_experimental/mcp_server/mcp_context.py index 9d792a429fe..11325a9f127 100644 --- a/litellm/proxy/_experimental/mcp_server/mcp_context.py +++ b/litellm/proxy/_experimental/mcp_server/mcp_context.py @@ -23,6 +23,7 @@ active_mcp_request_ctx_var: Final[ContextVar["ServerRequestContext | None"]] = C def get_active_mcp_request_ctx() -> "ServerRequestContext | None": return active_mcp_request_ctx_var.get() + # Set server-side in proxy_server.py route handlers when a request arrives via # /toolset/{name}/mcp or the toolset fallback in dynamic_mcp_route. # Never populated from client-supplied headers. diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/resolver.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/resolver.py index 41224e9ba2b..e71353e479c 100644 --- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/resolver.py +++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/resolver.py @@ -197,7 +197,9 @@ class UpstreamCredentialProvider: return Error(CredError.of_not_implemented("api_key BYOK source not implemented yet")) assert_never(config.key_source) - async def _id_jag(self, subject: Subject, server: ServerSpec, config: IdJagConfig) -> Result[httpx2.Auth, CredError]: + async def _id_jag( + self, subject: Subject, server: ServerSpec, config: IdJagConfig + ) -> Result[httpx2.Auth, CredError]: match await self._id_jag_subject_token(subject): case Error(err): return Error(err) diff --git a/litellm/proxy/_experimental/mcp_server/rest_endpoints.py b/litellm/proxy/_experimental/mcp_server/rest_endpoints.py index bebee75ad19..d8890ccad56 100644 --- a/litellm/proxy/_experimental/mcp_server/rest_endpoints.py +++ b/litellm/proxy/_experimental/mcp_server/rest_endpoints.py @@ -134,7 +134,16 @@ def _known_connection_error_message(exc: BaseException, url: str | None, timeout return "Failed to connect to MCP server: the connection timed out." if isinstance(exc, (httpx.HTTPStatusError, httpx2.HTTPStatusError)): return f"Failed to connect to MCP server: it returned HTTP {exc.response.status_code}." - if isinstance(exc, (httpx.NetworkError, httpx.RemoteProtocolError, httpx2.NetworkError, httpx2.RemoteProtocolError, ConnectionError)): + if isinstance( + exc, + ( + httpx.NetworkError, + httpx.RemoteProtocolError, + httpx2.NetworkError, + httpx2.RemoteProtocolError, + ConnectionError, + ), + ): return ( "Failed to connect to MCP server: the connection was interrupted. " "Check the server and network connection, then retry." diff --git a/litellm/proxy/_experimental/mcp_server/server.py b/litellm/proxy/_experimental/mcp_server/server.py index 505136f9e18..4a0fb8df65d 100644 --- a/litellm/proxy/_experimental/mcp_server/server.py +++ b/litellm/proxy/_experimental/mcp_server/server.py @@ -13,7 +13,7 @@ import time import traceback import types import uuid -from collections.abc import AsyncIterator, Callable, Mapping, Sequence +from collections.abc import AsyncIterator, Callable, Iterable, Mapping, Sequence from datetime import datetime from typing import TYPE_CHECKING, Any, Final, NoReturn, Protocol @@ -121,6 +121,7 @@ _MCP_TRANSPORT_SPAN_SCOPE_KEY: Final = "litellm_otel_transport_span" _MCP_DESTINATIONS_SCOPE_KEY: Final = "litellm_otel_request_destinations" _MCP_PROTOCOL_VERSION_HEADER: Final = b"mcp-protocol-version" + def unsupported_protocol_version(scope: Scope) -> str | None: """Return the unsupported ``MCP-Protocol-Version`` header value, if any. @@ -128,10 +129,11 @@ def unsupported_protocol_version(scope: Scope) -> str | None: ``HANDSHAKE_PROTOCOL_VERSIONS`` to the modern single-exchange path, which bypasses litellm's session/auth model, so the ASGI entry rejects it. """ - headers: Final = scope.get("headers") or [] - values: Final = [v for k, v in headers if k.lower() == _MCP_PROTOCOL_VERSION_HEADER] - for raw_value in values: - value: Final = raw_value.decode("latin-1").strip() + headers: Final[Iterable[tuple[bytes, bytes]]] = scope.get("headers") or () + values: Final = tuple( + raw.decode("latin-1").strip() for key, raw in headers if key.lower() == _MCP_PROTOCOL_VERSION_HEADER + ) + for value in values: if value and value not in HANDSHAKE_PROTOCOL_VERSIONS: return value return None @@ -880,7 +882,7 @@ if MCP_AVAILABLE: verbose_logger.exception("Error in list_tools endpoint: %s", e) # Return empty list instead of failing completely # This prevents the HTTP stream from failing and allows the client to get a response - return ListToolsResult(tools=[]) + return ListToolsResult(tools=[]) # mutable-ok: MCP result payload finally: _otel_reset_mcp_request_destinations(_destinations_token) _otel_reset_mcp_transport_span(_transport_token) @@ -1191,7 +1193,7 @@ if MCP_AVAILABLE: host_progress_callback: Final = _capture_host_progress_callback(ctx) # Create a body date for logging - body_data: Final = {"name": params.name, "arguments": params.arguments} + body_data: Final = {"name": params.name, "arguments": params.arguments} # mutable-ok: logging payload # Set trace/session id from raw_headers so spend logs and logging_obj stay consistent (same as A2A) chain_id: Final = get_chain_id_from_headers(raw_headers) if chain_id: @@ -1340,7 +1342,7 @@ if MCP_AVAILABLE: verbose_logger.exception("Error in list_prompts endpoint: %s", e) # Return empty list instead of failing completely # This prevents the HTTP stream from failing and allows the client to get a response - return ListPromptsResult(prompts=[]) + return ListPromptsResult(prompts=[]) # mutable-ok: MCP result payload finally: active_mcp_session_var.reset(_session_reset_token) active_mcp_request_ctx_var.reset(_ctx_reset_token) @@ -1416,7 +1418,7 @@ if MCP_AVAILABLE: return ListResourcesResult(resources=resources) except Exception as e: verbose_logger.exception("Error in list_resources endpoint: %s", e) - return ListResourcesResult(resources=[]) + return ListResourcesResult(resources=[]) # mutable-ok: MCP result payload finally: active_mcp_session_var.reset(_session_reset_token) active_mcp_request_ctx_var.reset(_ctx_reset_token) @@ -1461,7 +1463,7 @@ if MCP_AVAILABLE: return ListResourceTemplatesResult(resource_templates=resource_templates) except Exception as e: verbose_logger.exception("Error in list_resource_templates endpoint: %s", e) - return ListResourceTemplatesResult(resource_templates=[]) + return ListResourceTemplatesResult(resource_templates=[]) # mutable-ok: MCP result payload finally: active_mcp_session_var.reset(_session_reset_token) active_mcp_request_ctx_var.reset(_ctx_reset_token) @@ -3618,8 +3620,14 @@ if MCP_AVAILABLE: raise except Exception as e: verbose_logger.exception("Error executing local tool %s: %s", name, e) - return CallToolResult(content=[TextContent(text=f"Error: {e}", type="text")], is_error=True) - return CallToolResult(content=[TextContent(text=str(result), type="text")], is_error=False) + return CallToolResult( + content=[TextContent(text=f"Error: {e}", type="text")], # mutable-ok: MCP result content + is_error=True, + ) + return CallToolResult( + content=[TextContent(text=str(result), type="text")], # mutable-ok: MCP result content + is_error=False, + ) def _get_mcp_servers_in_path(path: str) -> list[str] | None: """ @@ -4363,7 +4371,7 @@ if MCP_AVAILABLE: supported: Final = ", ".join(sorted(HANDSHAKE_PROTOCOL_VERSIONS)) await JSONResponse( status_code=400, - content={ + content={ # mutable-ok: JSON-RPC error payload "jsonrpc": "2.0", "id": None, "error": { diff --git a/litellm/proxy/_experimental/mcp_server/tool_search.py b/litellm/proxy/_experimental/mcp_server/tool_search.py index e6dce446751..a482d02c31d 100644 --- a/litellm/proxy/_experimental/mcp_server/tool_search.py +++ b/litellm/proxy/_experimental/mcp_server/tool_search.py @@ -99,11 +99,20 @@ def mcp_tool_search_settings() -> MCPToolSearchSettings | ValidationError: def _tool_result(tool: Tool) -> ToolSearchResult: - return {"name": tool.name, "description": tool.description or "", "inputSchema": tool.input_schema} + return { + "name": tool.name, + "description": tool.description or "", + "inputSchema": tool.input_schema, + } # mutable-ok: wire schema payload def _scored_result(tool: Tool, score: float) -> ToolSearchResult: - return {"name": tool.name, "description": tool.description or "", "inputSchema": tool.input_schema, "score": score} + return { + "name": tool.name, + "description": tool.description or "", + "inputSchema": tool.input_schema, + "score": score, + } # mutable-ok: wire schema payload _MCP_PROXY_IDENTITY_META_KEY: Final[str] = "litellm.ai/proxy_tool_identity" diff --git a/litellm/proxy/guardrails/guardrail_hooks/cisco_ai_defense/cisco_ai_defense_mcp.py b/litellm/proxy/guardrails/guardrail_hooks/cisco_ai_defense/cisco_ai_defense_mcp.py index 777db999672..8d5a7c7fecb 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/cisco_ai_defense/cisco_ai_defense_mcp.py +++ b/litellm/proxy/guardrails/guardrail_hooks/cisco_ai_defense/cisco_ai_defense_mcp.py @@ -34,9 +34,11 @@ def _serialize_mcp_content_item(item: object) -> dict[str, object]: model_dump: Final = getattr(item, "model_dump", None) if callable(model_dump): try: - return dict(model_dump(exclude_none=True)) + dumped: Final[dict[str, object]] = model_dump(exclude_none=True) + return dict(dumped) except TypeError: - return dict(model_dump()) + dumped_fallback: Final[dict[str, object]] = model_dump() + return dict(dumped_fallback) text: Final = getattr(item, "text", None) if isinstance(text, str): return {"type": getattr(item, "type", "text"), "text": text} @@ -507,8 +509,7 @@ class _CiscoAIDefenseMcpMixin: source: object = None, ) -> dict[str, object]: result: Final[dict[str, object]] = {"content": [_serialize_mcp_content_item(item) for item in content]} - for key in ("structuredContent", "isError"): - snake_key: Final = "structured_content" if key == "structuredContent" else "is_error" + for key, snake_key in (("structuredContent", "structured_content"), ("isError", "is_error")): value = source.get(key) if isinstance(source, dict) else getattr(source, snake_key, None) if value is not None and (key != "isError" or isinstance(value, bool)): result[key] = value diff --git a/tests/e2e/mcp/oauth_chat_client.py b/tests/e2e/mcp/oauth_chat_client.py index 2eaf512cfa5..763b348b197 100644 --- a/tests/e2e/mcp/oauth_chat_client.py +++ b/tests/e2e/mcp/oauth_chat_client.py @@ -22,16 +22,16 @@ from typing import TYPE_CHECKING from urllib.parse import parse_qsl import httpx +import httpx2 import pytest +from e2e_config import PROXY_BASE_URL, REQUEST_TIMEOUT +from e2e_http import AuthHeaders, NoBody, unwrap from mcp import ClientSession from mcp.client.auth import OAuthClientProvider from mcp.client.streamable_http import streamable_http_client -from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken - -from e2e_config import PROXY_BASE_URL, REQUEST_TIMEOUT -from proxy_client import ProxyClient -from e2e_http import AuthHeaders, NoBody, unwrap +from mcp.shared.auth import AuthorizationCodeResult, OAuthClientInformationFull, OAuthClientMetadata, OAuthToken from models import ChatBody, ChatResponse, McpServerCreateBody, McpServerInfo +from proxy_client import ProxyClient if TYPE_CHECKING: from playwright.async_api import Route @@ -88,7 +88,7 @@ async def _browser_follow_authorize(start_url: str, storage_state_path: str) -> if url.startswith(OAUTH_CLIENT_REDIRECT_URI) and "url" not in captured: captured["url"] = url - async def _swallow_redirect(route: "Route") -> None: + async def _swallow_redirect(route: Route) -> None: await route.fulfill(status=200, content_type="text/plain", body="ok") async with async_playwright() as playwright: @@ -139,10 +139,10 @@ def _oauth_provider(url: str, storage: InMemoryTokenStorage, storage_state_path: code_holder["code"] = code code_holder["state"] = state - async def callback_handler() -> tuple[str, str | None]: + async def callback_handler() -> AuthorizationCodeResult: code = code_holder.get("code") assert code is not None, "callback_handler ran before the authorize redirect completed" - return code, code_holder.get("state") + return AuthorizationCodeResult(code=code, state=code_holder.get("state")) return OAuthClientProvider( server_url=url, @@ -161,30 +161,30 @@ def _oauth_provider(url: str, storage: InMemoryTokenStorage, storage_state_path: ) -class _HeaderInjectingTransport(httpx.AsyncBaseTransport): +class _HeaderInjectingTransport(httpx2.AsyncBaseTransport): """Adds the caller's LiteLLM key header to every outgoing SDK request (discovery, DCR, token exchange), so the gateway resolves which user to store the upstream token for from the key on the token exchange, exactly like a production MCP host configured with a LiteLLM key header.""" - def __init__(self, inner: httpx.AsyncBaseTransport, headers: dict[str, str]) -> None: + def __init__(self, inner: httpx2.AsyncBaseTransport, headers: dict[str, str]) -> None: self._inner = inner self._headers = headers - async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response: for name, value in self._headers.items(): if name not in request.headers: request.headers[name] = value return await self._inner.handle_async_request(request) -def _oauth_http_client(headers: dict[str, str], auth: OAuthClientProvider) -> httpx.AsyncClient: - return httpx.AsyncClient( +def _oauth_http_client(headers: dict[str, str], auth: OAuthClientProvider) -> httpx2.AsyncClient: + return httpx2.AsyncClient( headers=headers, auth=auth, - timeout=httpx.Timeout(REQUEST_TIMEOUT), + timeout=httpx2.Timeout(REQUEST_TIMEOUT), follow_redirects=True, - transport=_HeaderInjectingTransport(httpx.AsyncHTTPTransport(), headers), + transport=_HeaderInjectingTransport(httpx2.AsyncHTTPTransport(), headers), ) @@ -192,7 +192,7 @@ async def _seed_via_dance( url: str, headers: dict[str, str], storage: InMemoryTokenStorage, storage_state_path: str ) -> tuple[str, ...]: async with _oauth_http_client(headers, _oauth_provider(url, storage, storage_state_path)) as http_client: - async with streamable_http_client(url, http_client=http_client) as (read, write, _): + async with streamable_http_client(url, http_client=http_client) as (read, write): async with ClientSession(read, write) as session: await session.initialize() listed = await session.list_tools() diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py index 303fa48e877..fbecdd60a26 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py @@ -102,6 +102,7 @@ def _mcp_request_ctx(**overrides): kwargs.update(overrides) return ServerRequestContext(**kwargs) + @pytest.fixture(autouse=True) def enable_eager_mcp_oauth_discovery(monkeypatch): monkeypatch.setenv("LITELLM_MCP_OAUTH_DISCOVERY_ON_STARTUP", "1") @@ -4558,7 +4559,9 @@ class TestMCPServerManager: @pytest.mark.parametrize("auth_type", [MCPAuth.none, MCPAuth.bearer_token, MCPAuth.api_key, MCPAuth.oauth2]) @pytest.mark.parametrize("is_byok", [False, True]) @pytest.mark.parametrize("scheme", ["http", "https"]) - async def test_openapi_health_loads_spec_without_mcp_handshake(self, respx_mock, monkeypatch, auth_type, is_byok, scheme): + async def test_openapi_health_loads_spec_without_mcp_handshake( + self, respx_mock, monkeypatch, auth_type, is_byok, scheme + ): monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True") manager = MCPServerManager() server = MCPServer( @@ -4608,14 +4611,28 @@ class TestMCPServerManager: @pytest.mark.parametrize( ("failure", "expected_status", "expected_error"), [ - (httpx.Response(401, text="secret response content"), "unhealthy", "OpenAPI specification request failed (HTTP 401)"), + ( + httpx.Response(401, text="secret response content"), + "unhealthy", + "OpenAPI specification request failed (HTTP 401)", + ), (httpx.Response(404), "unhealthy", "OpenAPI specification request failed (HTTP 404)"), (httpx.Response(500), "unhealthy", "OpenAPI specification request failed (HTTP 500)"), - (httpx.ConnectError("secret network details"), "unhealthy", "OpenAPI specification could not be loaded (ConnectError)"), - (httpx.Response(200, text="secret invalid JSON body"), "unhealthy", "OpenAPI specification could not be loaded (JSONDecodeError)"), + ( + httpx.ConnectError("secret network details"), + "unhealthy", + "OpenAPI specification could not be loaded (ConnectError)", + ), + ( + httpx.Response(200, text="secret invalid JSON body"), + "unhealthy", + "OpenAPI specification could not be loaded (JSONDecodeError)", + ), ], ) - async def test_openapi_health_reports_safe_failures(self, respx_mock, monkeypatch, failure, expected_status, expected_error): + async def test_openapi_health_reports_safe_failures( + self, respx_mock, monkeypatch, failure, expected_status, expected_error + ): monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True") manager = MCPServerManager() server = MCPServer( @@ -5150,8 +5167,15 @@ class TestMCPServerManager: captured: dict = {} def fake_create_tool_function( - path, method, operation, base_url, headers=None, server_label=None, relays_upstream_auth=False, - auth_type=None, upstream_token_header=None, + path, + method, + operation, + base_url, + headers=None, + server_label=None, + relays_upstream_auth=False, + auth_type=None, + upstream_token_header=None, ): captured["headers"] = headers captured["server_label"] = server_label @@ -5236,8 +5260,15 @@ class TestMCPServerManager: captured: dict = {} def fake_create_tool_function( - path, method, operation, base_url, headers=None, server_label=None, relays_upstream_auth=False, - auth_type=None, upstream_token_header=None, + path, + method, + operation, + base_url, + headers=None, + server_label=None, + relays_upstream_auth=False, + auth_type=None, + upstream_token_header=None, ): captured["headers"] = headers @@ -6114,17 +6145,17 @@ class TestMCPServerManager: tool1 = MagicMock() tool1.name = "allowed_tool_1" tool1.description = "This tool is allowed" - tool1.input_schema= {} + tool1.input_schema = {} tool2 = MagicMock() tool2.name = "blocked_tool" tool2.description = "This tool is not allowed" - tool2.input_schema= {} + tool2.input_schema = {} tool3 = MagicMock() tool3.name = "allowed_tool_2" tool3.description = "This tool is also allowed" - tool3.input_schema= {} + tool3.input_schema = {} # Mock the global_mcp_server_manager._get_tools_from_server from litellm.proxy._experimental.mcp_server import rest_endpoints @@ -6164,17 +6195,17 @@ class TestMCPServerManager: tool1 = MagicMock() tool1.name = "tool_1" tool1.description = "Tool 1" - tool1.input_schema= {} + tool1.input_schema = {} tool2 = MagicMock() tool2.name = "tool_2" tool2.description = "Tool 2" - tool2.input_schema= {} + tool2.input_schema = {} tool3 = MagicMock() tool3.name = "tool_3" tool3.description = "Tool 3" - tool3.input_schema= {} + tool3.input_schema = {} # Mock the global_mcp_server_manager._get_tools_from_server from litellm.proxy._experimental.mcp_server import rest_endpoints @@ -6214,12 +6245,12 @@ class TestMCPServerManager: tool1 = MagicMock() tool1.name = "tool_1" tool1.description = "Tool 1" - tool1.input_schema= {} + tool1.input_schema = {} tool2 = MagicMock() tool2.name = "tool_2" tool2.description = "Tool 2" - tool2.input_schema= {} + tool2.input_schema = {} # Mock the global_mcp_server_manager._get_tools_from_server from litellm.proxy._experimental.mcp_server import rest_endpoints @@ -6559,7 +6590,7 @@ class TestMCPServerManager: # Return a mock CallToolResult result = MagicMock(spec=CallToolResult) result.content = [{"type": "text", "text": "Tool executed successfully"}] - result.is_error= False + result.is_error = False return result mock_client.call_tool.side_effect = mock_call_tool @@ -12744,7 +12775,12 @@ async def test_debug_resolution_matches_final_header_conflict_winner( from litellm.proxy._experimental.mcp_server.auth.litellm_auth_handler import MCPAuthenticatedUser from litellm.proxy._experimental.mcp_server.mcp_debug import MCP_AUTH_DIAGNOSTICS_SCOPE_KEY, MCPAuthDiagnostics from litellm.proxy._experimental.mcp_server.outbound_credentials import ( - ApiKeyConfig, AuthorizationCodeConfig, NoneConfig, ServerSpec, SharedKey, UpstreamCredentialProvider, + ApiKeyConfig, + AuthorizationCodeConfig, + NoneConfig, + ServerSpec, + SharedKey, + UpstreamCredentialProvider, ) from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import OAuthToken from litellm.types.mcp_server.mcp_server_manager import MCPServer @@ -12760,9 +12796,11 @@ async def test_debug_resolution_matches_final_header_conflict_winner( store = Store() context = MCPAuthenticatedUser(UserAPIKeyAuth(user_id="alice")) diagnostics = MCPAuthDiagnostics() - token = active_mcp_request_ctx_var.set(_mcp_request_ctx( - request=Request({"type": "http", MCP_AUTH_DIAGNOSTICS_SCOPE_KEY: diagnostics}), - )) + token = active_mcp_request_ctx_var.set( + _mcp_request_ctx( + request=Request({"type": "http", MCP_AUTH_DIAGNOSTICS_SCOPE_KEY: diagnostics}), + ) + ) selected = { "stored": AuthorizationCodeConfig(), "static": ApiKeyConfig(key_source=SharedKey(value=SecretStr("static-token"))), @@ -12771,7 +12809,10 @@ async def test_debug_resolution_matches_final_header_conflict_winner( try: auth, remaining = await MCPServerManager()._resolve_v2_auth( server=MCPServer( - server_id="s", name="s", transport="http", url="https://up.example/mcp", + server_id="s", + name="s", + transport="http", + url="https://up.example/mcp", static_headers={"Authorization": "Bearer configured"}, ), spec=ServerSpec(server_id="s", resource="https://up.example/mcp", config=selected), @@ -12800,16 +12841,24 @@ async def test_debug_reports_legacy_signing_and_non_http_transport(transport: Li from litellm.types.mcp_server.mcp_server_manager import MCPServer diagnostics = MCPAuthDiagnostics() - token = active_mcp_request_ctx_var.set(_mcp_request_ctx( - request=Request({"type": "http", MCP_AUTH_DIAGNOSTICS_SCOPE_KEY: diagnostics}), - )) + token = active_mcp_request_ctx_var.set( + _mcp_request_ctx( + request=Request({"type": "http", MCP_AUTH_DIAGNOSTICS_SCOPE_KEY: diagnostics}), + ) + ) try: server = MCPServer( - server_id="signed", name="signed", transport=transport, - url="https://up.example/mcp", auth_type="aws_sigv4", - aws_access_key_id="AKIDEXAMPLE", aws_secret_access_key="test-signing-secret", - aws_region_name="us-east-1", aws_service_name="execute-api", - command="python", args=["-c", "pass"], + server_id="signed", + name="signed", + transport=transport, + url="https://up.example/mcp", + auth_type="aws_sigv4", + aws_access_key_id="AKIDEXAMPLE", + aws_secret_access_key="test-signing-secret", + aws_region_name="us-east-1", + aws_service_name="execute-api", + command="python", + args=["-c", "pass"], ) client = await MCPServerManager()._create_mcp_client(server) if transport == "stdio": @@ -12828,12 +12877,16 @@ async def test_debug_reports_legacy_signing_and_non_http_transport(transport: Li async def test_temporary_server_discovery_reuses_resolved_metadata_without_publishing() -> None: manager: Final = MCPServerManager() server: Final = MCPServer( - server_id="temporary-oauth-discovery", name="temporary", url="https://idp.example.com/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.true_passthrough, + server_id="temporary-oauth-discovery", + name="temporary", + url="https://idp.example.com/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.true_passthrough, ) manager._set_oauth_discovery_deferred(server.server_id, True) metadata: Final = MCPOAuthMetadata( - authorization_url="https://idp.example.com/authorize", token_url="https://idp.example.com/token", + authorization_url="https://idp.example.com/authorize", + token_url="https://idp.example.com/token", registration_url="https://idp.example.com/register", ) with patch.object(manager, "_discover_oauth_metadata_for_server", AsyncMock(return_value=metadata)) as discovery: @@ -12853,13 +12906,18 @@ async def test_temporary_server_discovery_reuses_resolved_metadata_without_publi async def test_repeated_stale_oauth_discovery_is_bounded(auth_type: MCPAuth) -> None: manager: Final = MCPServerManager() server: Final = MCPServer( - server_id="repeated-stale", name="stale", url="https://idp.example.com/mcp", - transport=MCPTransport.http, auth_type=auth_type, oauth2_flow="authorization_code", + server_id="repeated-stale", + name="stale", + url="https://idp.example.com/mcp", + transport=MCPTransport.http, + auth_type=auth_type, + oauth2_flow="authorization_code", ) manager.registry[server.server_id] = server manager._set_oauth_discovery_deferred(server.server_id, True) metadata: Final = MCPOAuthMetadata( - authorization_url="https://idp.example.com/authorize", token_url="https://idp.example.com/token", + authorization_url="https://idp.example.com/authorize", + token_url="https://idp.example.com/token", ) with ( patch.object(manager, "_discover_oauth_metadata_for_server", AsyncMock(return_value=metadata)) as discovery, @@ -12879,13 +12937,20 @@ async def test_repeated_stale_oauth_discovery_is_bounded(auth_type: MCPAuth) -> async def test_stale_discovery_falls_back_to_resolved_registered_server() -> None: manager: Final = MCPServerManager() original: Final = MCPServer( - server_id="resolved-replacement", name="replacement", url="https://old.example.com/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.oauth2, oauth2_flow="authorization_code", + server_id="resolved-replacement", + name="replacement", + url="https://old.example.com/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.oauth2, + oauth2_flow="authorization_code", + ) + replacement: Final = original.model_copy( + update={ + "url": "https://new.example.com/mcp", + "authorization_url": "https://new.example.com/authorize", + "token_url": "https://new.example.com/token", + } ) - replacement: Final = original.model_copy(update={ - "url": "https://new.example.com/mcp", "authorization_url": "https://new.example.com/authorize", - "token_url": "https://new.example.com/token", - }) manager.registry[original.server_id] = replacement assert await manager._rejoin_oauth_metadata_discovery(original, retry_stale=False) is replacement @@ -12893,8 +12958,11 @@ async def test_stale_discovery_falls_back_to_resolved_registered_server() -> Non def test_stale_discovery_cannot_overwrite_new_registered_server() -> None: manager: Final = MCPServerManager() original: Final = MCPServer( - server_id="stale-publication", name="publication", url="https://old.example.com/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.oauth2, + server_id="stale-publication", + name="publication", + url="https://old.example.com/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.oauth2, ) manager._set_oauth_discovery_deferred(original.server_id, True) original_slot: Final = manager._oauth_discovery_slot(original.server_id) @@ -12910,9 +12978,13 @@ def test_stale_discovery_cannot_overwrite_new_registered_server() -> None: async def test_temporary_oauth_discovery_expires_without_more_requests() -> None: manager: Final = MCPServerManager() server: Final = MCPServer( - server_id="expiring-session", name="temporary", url="https://idp.example.com/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.true_passthrough, - authorization_url="https://idp.example.com/authorize", token_url="https://idp.example.com/token", + server_id="expiring-session", + name="temporary", + url="https://idp.example.com/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.true_passthrough, + authorization_url="https://idp.example.com/authorize", + token_url="https://idp.example.com/token", ) manager._set_oauth_discovery_deferred(server.server_id, True) resolved: Final = await manager.ensure_oauth_metadata_discovered(server) @@ -13013,7 +13085,9 @@ async def test_openapi_health_reports_size_limit_as_unknown_and_caches_failure(r result = await manager.health_check_server(server.server_id) cached = await manager.health_check_server(server.server_id) assert result.status == "unknown" - assert result.health_check_error == "OpenAPI specification probe refused: Response exceeds the configured size limit" + assert ( + result.health_check_error == "OpenAPI specification probe refused: Response exceeds the configured size limit" + ) assert cached.health_check_error == result.health_check_error assert cached.last_health_check == result.last_health_check assert route.call_count == 1 @@ -13025,8 +13099,11 @@ async def test_openapi_health_cancellation_does_not_poison_cache(respx_mock, mon monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True") manager = MCPServerManager() server = MCPServer( - server_id="cancelled-cache", name="cancelled-cache", transport=MCPTransport.http, - spec_path="https://93.184.216.34/cancelled-cache.json", auth_type=MCPAuth.none, + server_id="cancelled-cache", + name="cancelled-cache", + transport=MCPTransport.http, + spec_path="https://93.184.216.34/cancelled-cache.json", + auth_type=MCPAuth.none, ) manager.registry = {server.server_id: server} started = asyncio.Event() @@ -13084,7 +13161,11 @@ def _mcp_upstream(respond): auth=kwargs.get("auth") or self._resolved_auth or self._aws_auth, ) - with patch.object(MCPClient, "_create_httpx_client_factory", lambda self: functools.partial(make_client, self)): + with ( + patch.object( # test-quality-ok: respx cannot intercept httpx2; inject MockTransport through the client factory + MCPClient, "_create_httpx_client_factory", lambda self: functools.partial(make_client, self) + ) + ): yield @@ -13106,11 +13187,18 @@ class _DiscoveryUpstream: return httpx2.Response(202) self.requests = (*self.requests, (payload.method, request.headers.get("authorization", ""))) if payload.method == "initialize": - return httpx2.Response(200, json={ - "jsonrpc": "2.0", "id": payload.id, - "result": {"protocolVersion": "2025-03-26", "serverInfo": {"name": "discovery", "version": "1"}, - "capabilities": {} if self.outcome == "unsupported" else {"prompts": {}, "resources": {}}}, - }) + return httpx2.Response( + 200, + json={ + "jsonrpc": "2.0", + "id": payload.id, + "result": { + "protocolVersion": "2025-03-26", + "serverInfo": {"name": "discovery", "version": "1"}, + "capabilities": {} if self.outcome == "unsupported" else {"prompts": {}, "resources": {}}, + }, + }, + ) self.entered.set() await self.release.wait() if self.outcome == "failure": @@ -13118,12 +13206,15 @@ class _DiscoveryUpstream: if self.outcome == "cancelled": raise asyncio.CancelledError() if self.outcome == "rejected": - return httpx2.Response(200, json={"jsonrpc": "2.0", "id": payload.id, - "error": {"code": -32601, "message": "Unsupported"}}) + return httpx2.Response( + 200, json={"jsonrpc": "2.0", "id": payload.id, "error": {"code": -32601, "message": "Unsupported"}} + ) result: Final = { "prompts/list": {"prompts": [{"name": "example", "description": "original"}]}, "resources/list": {"resources": [{"name": "example", "uri": "test://example", "description": "original"}]}, - "resources/templates/list": {"resourceTemplates": [{"name": "example", "uriTemplate": "test://{name}", "description": "original"}]}, + "resources/templates/list": { + "resourceTemplates": [{"name": "example", "uriTemplate": "test://{name}", "description": "original"}] + }, "tools/list": {"tools": []}, }[payload.method] return httpx2.Response(200, json={"jsonrpc": "2.0", "id": payload.id, "result": result}) @@ -13134,7 +13225,9 @@ class _DiscoveryUpstream: def _discovery_server() -> MCPServer: - return MCPServer(server_id="discovery", name="discovery", url="https://discovery.example/mcp", transport=MCPTransport.http) + return MCPServer( + server_id="discovery", name="discovery", url="https://discovery.example/mcp", transport=MCPTransport.http + ) @pytest.mark.asyncio @@ -13145,8 +13238,11 @@ async def test_discovery_cache_reuses_raw_results_and_expires(kind: str) -> None clock: Final = _DiscoveryClock() manager: Final = MCPServerManager(discovery_clock=clock) upstream: Final = _DiscoveryUpstream() - operation: Final = {"prompts": manager.get_prompts_from_server, "resources": manager.get_resources_from_server, - "templates": manager.get_resource_templates_from_server}[kind] + operation: Final = { + "prompts": manager.get_prompts_from_server, + "resources": manager.get_resources_from_server, + "templates": manager.get_resource_templates_from_server, + }[kind] server: Final = _discovery_server() with _mcp_upstream(upstream.respond): first: Final = await operation(server, None) @@ -13174,8 +13270,11 @@ async def test_discovery_cache_empty_results_and_failures(kind: str, outcome: st manager: Final = MCPServerManager() upstream: Final = _DiscoveryUpstream() upstream.outcome = outcome - operation: Final = {"prompts": manager.get_prompts_from_server, "resources": manager.get_resources_from_server, - "templates": manager.get_resource_templates_from_server}[kind] + operation: Final = { + "prompts": manager.get_prompts_from_server, + "resources": manager.get_resources_from_server, + "templates": manager.get_resource_templates_from_server, + }[kind] with _mcp_upstream(upstream.respond): assert await operation(_discovery_server(), None) == [] assert await operation(_discovery_server(), None) == [] @@ -13200,9 +13299,20 @@ async def test_discovery_cache_isolates_forwarded_credentials_and_shares_static_ assert len(await manager.get_prompts_from_server(server, user)) == 1 assert upstream.initializes == 1 for credential in ("first-secret", "second-secret", "first-secret"): - assert len(await manager.get_prompts_from_server(server, first_user, extra_headers={"Authorization": credential})) == 1 + assert ( + len( + await manager.get_prompts_from_server( + server, first_user, extra_headers={"Authorization": credential} + ) + ) + == 1 + ) assert upstream.initializes == 3 - assert {auth for method, auth in upstream.requests if method == "prompts/list"} == {"", "first-secret", "second-secret"} + assert {auth for method, auth in upstream.requests if method == "prompts/list"} == { + "", + "first-secret", + "second-secret", + } @pytest.mark.asyncio @@ -13213,7 +13323,9 @@ async def test_discovery_cache_coalesces_and_survives_waiter_cancellation() -> N upstream: Final = _DiscoveryUpstream() upstream.release.clear() with _mcp_upstream(upstream.respond): - tasks: Final = tuple(asyncio.create_task(manager.get_prompts_from_server(_discovery_server(), None)) for _ in range(10)) + tasks: Final = tuple( + asyncio.create_task(manager.get_prompts_from_server(_discovery_server(), None)) for _ in range(10) + ) await asyncio.wait_for(upstream.entered.wait(), timeout=5) tasks[0].cancel() with pytest.raises(asyncio.CancelledError): @@ -13260,7 +13372,9 @@ async def test_discovery_cache_can_be_disabled(monkeypatch: pytest.MonkeyPatch) assert upstream.initializes == 2 -@pytest.mark.parametrize("value,expected", (("invalid", 60.0), ("nan", 60.0), ("inf", 60.0), ("-1", 60.0), ("12.5", 12.5))) +@pytest.mark.parametrize( + "value,expected", (("invalid", 60.0), ("nan", 60.0), ("inf", 60.0), ("-1", 60.0), ("12.5", 12.5)) +) def test_discovery_cache_ttl_validation(value: str, expected: float, monkeypatch: pytest.MonkeyPatch) -> None: from litellm.proxy._experimental.mcp_server.mcp_server_manager import _mcp_discovery_cache_ttl @@ -13378,9 +13492,15 @@ async def test_discovery_cache_tracks_resolved_credentials_across_workers() -> N source: Final = CredentialSource() managers: Final = (MCPServerManager(cred_provider=source), MCPServerManager(cred_provider=source)) server: Final = MCPServer( - server_id="discovery", name="discovery", url="https://discovery.example/mcp", transport=MCPTransport.http, - auth_type=MCPAuth.oauth2, oauth2_flow="authorization_code", client_id="discovery-client", - authorization_url="https://discovery.example/authorize", token_url="https://discovery.example/token", + server_id="discovery", + name="discovery", + url="https://discovery.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.oauth2, + oauth2_flow="authorization_code", + client_id="discovery-client", + authorization_url="https://discovery.example/authorize", + token_url="https://discovery.example/token", ) user: Final = UserAPIKeyAuth(user_id="same-user", api_key="same-key") upstream: Final = _DiscoveryUpstream() @@ -13398,11 +13518,15 @@ async def test_discovery_cache_tracks_resolved_credentials_across_workers() -> N with _mcp_upstream(respond): for manager in managers: - assert [item.name for item in await manager.get_prompts_from_server(server, user)] == ["discovery-account-a"] + assert [item.name for item in await manager.get_prompts_from_server(server, user)] == [ + "discovery-account-a" + ] assert upstream.initializes == 2 source.token = "token-b" for manager in managers: - assert [item.name for item in await manager.get_prompts_from_server(server, user)] == ["discovery-account-b"] + assert [item.name for item in await manager.get_prompts_from_server(server, user)] == [ + "discovery-account-b" + ] assert upstream.initializes == 4 source.token = None for manager in managers: @@ -13429,9 +13553,15 @@ async def test_discovery_resolves_stored_oauth_for_the_requesting_user() -> None store: Final = TokenStore() manager: Final = MCPServerManager(per_user_oauth_token_store=store) server: Final = MCPServer( - server_id="discovery", name="discovery", url="https://discovery.example/mcp", transport=MCPTransport.http, - auth_type=MCPAuth.oauth2, oauth2_flow="authorization_code", client_id="discovery-client", - authorization_url="https://discovery.example/authorize", token_url="https://discovery.example/token", + server_id="discovery", + name="discovery", + url="https://discovery.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.oauth2, + oauth2_flow="authorization_code", + client_id="discovery-client", + authorization_url="https://discovery.example/authorize", + token_url="https://discovery.example/token", ) user: Final = UserAPIKeyAuth(user_id="requesting-user") upstream: Final = _DiscoveryUpstream() @@ -13507,26 +13637,45 @@ async def test_discovery_cache_returns_oversized_results_without_retaining_them( class TestProtectedCredentialPreparation: @pytest.mark.asyncio - @pytest.mark.parametrize("auth_type,credential", [ - (MCPAuth.bearer_token, None), - (MCPAuth.bearer_token, "Bearer"), - (MCPAuth.api_key, None), - (MCPAuth.basic, "Basic"), - ]) + @pytest.mark.parametrize( + "auth_type,credential", + [ + (MCPAuth.bearer_token, None), + (MCPAuth.bearer_token, "Bearer"), + (MCPAuth.api_key, None), + (MCPAuth.basic, "Basic"), + ], + ) @pytest.mark.parametrize("dispatch", ["managed", "local"]) async def test_openapi_dispatch_rejects_unusable_effective_credentials( - self, tmp_path: Path, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch, - auth_type: MCPAuthType, credential: str | None, dispatch: str, + self, + tmp_path: Path, + respx_mock: MockRouter, + monkeypatch: pytest.MonkeyPatch, + auth_type: MCPAuthType, + credential: str | None, + dispatch: str, ) -> None: from litellm.proxy._experimental.mcp_server.server import _handle_local_mcp_tool from litellm.proxy._experimental.mcp_server.utils import add_server_prefix_to_name, get_server_prefix spec_path: Final = tmp_path / "openapi.json" - spec_path.write_text(json.dumps({"openapi": "3.0.0", "info": {"title": "Auth", "version": "1"}, - "paths": {"/echo": {"get": {"operationId": "echo"}}}})) + spec_path.write_text( + json.dumps( + { + "openapi": "3.0.0", + "info": {"title": "Auth", "version": "1"}, + "paths": {"/echo": {"get": {"operationId": "echo"}}}, + } + ) + ) server: Final = MCPServer( - server_id="dispatch-auth", name="dispatch-auth", url="https://upstream.example", - transport=MCPTransport.http, auth_type=auth_type, authentication_token=credential, + server_id="dispatch-auth", + name="dispatch-auth", + url="https://upstream.example", + transport=MCPTransport.http, + auth_type=auth_type, + authentication_token=credential, ) manager: Final = MCPServerManager() await manager._register_openapi_tools(str(spec_path), server, server.url) @@ -13549,14 +13698,21 @@ class TestProtectedCredentialPreparation: self, transport: MCPTransport, client_secret: str | None, subject: str | None ) -> None: server = MCPServer( - server_id="incomplete-obo", name="incomplete-obo", url="https://upstream.example/mcp", - transport=transport, auth_type=MCPAuth.oauth2_token_exchange, - client_id="gateway", client_secret=client_secret, - token_exchange_endpoint="https://idp.example/token", authentication_token="static-fallback", + server_id="incomplete-obo", + name="incomplete-obo", + url="https://upstream.example/mcp", + transport=transport, + auth_type=MCPAuth.oauth2_token_exchange, + client_id="gateway", + client_secret=client_secret, + token_exchange_endpoint="https://idp.example/token", + authentication_token="static-fallback", ) with pytest.raises(HTTPException) as exc: await MCPServerManager()._create_mcp_client( - server, mcp_auth_header="Bearer override", subject_token=subject, + server, + mcp_auth_header="Bearer override", + subject_token=subject, ) assert exc.value.status_code == (401 if subject is None else 500) assert "static-fallback" not in str(exc.value.detail) @@ -13569,8 +13725,11 @@ class TestProtectedCredentialPreparation: self, auth_type: MCPAuthType, credential: str | dict[str, str] | None ) -> None: server = MCPServer( - server_id="empty-static", name="empty-static", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=auth_type, + server_id="empty-static", + name="empty-static", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=auth_type, ) with pytest.raises(HTTPException) as exc: await MCPServerManager()._create_mcp_client(server, mcp_auth_header=credential) @@ -13578,16 +13737,22 @@ class TestProtectedCredentialPreparation: assert "credential" in str(exc.value.detail).lower() @pytest.mark.asyncio - @pytest.mark.parametrize("auth_type,headers", [ - (MCPAuth.api_key, {"X-API-Key": "key"}), - (MCPAuth.bearer_token, {"Authorization": "Bearer token"}), - ]) + @pytest.mark.parametrize( + "auth_type,headers", + [ + (MCPAuth.api_key, {"X-API-Key": "key"}), + (MCPAuth.bearer_token, {"Authorization": "Bearer token"}), + ], + ) async def test_static_auth_accepts_actual_forwarded_credential( self, auth_type: MCPAuthType, headers: dict[str, str] ) -> None: server = MCPServer( - server_id="header-static", name="header-static", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=auth_type, + server_id="header-static", + name="header-static", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=auth_type, ) client = await MCPServerManager()._create_mcp_client(server, extra_headers=headers) assert client._get_auth_headers() == headers @@ -13596,29 +13761,48 @@ class TestProtectedCredentialPreparation: @pytest.mark.parametrize("auth_type", [MCPAuth.oauth2_token_exchange]) async def test_openapi_protected_auth_rejects_missing_credentials(self, auth_type: MCPAuthType) -> None: server = MCPServer( - server_id="openapi-empty", name="openapi-empty", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=auth_type, + server_id="openapi-empty", + name="openapi-empty", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=auth_type, token_exchange_endpoint="https://idp.example/token", ) with pytest.raises(HTTPException) as exc: await MCPServerManager().resolve_openapi_upstream_auth( - mcp_server=server, oauth2_headers=None, raw_headers=None, mcp_auth_header=None, - user_api_key_auth=None, forwarded_headers=None, + mcp_server=server, + oauth2_headers=None, + raw_headers=None, + mcp_auth_header=None, + user_api_key_auth=None, + forwarded_headers=None, ) assert exc.value.status_code in (401, 500) @pytest.mark.asyncio - @pytest.mark.parametrize("auth_type,slot,value", [ - (MCPAuth.api_key, "X-API-Key", "token"), - (MCPAuth.authorization, "Authorization", "opaque-secret-value"), - (MCPAuth.authorization, "Authorization", "Bearer abc"), - (MCPAuth.authorization, "Authorization", "Custom abc"), - ]) + @pytest.mark.parametrize( + "auth_type,slot,value", + [ + (MCPAuth.api_key, "X-API-Key", "token"), + (MCPAuth.authorization, "Authorization", "opaque-secret-value"), + (MCPAuth.authorization, "Authorization", "Bearer abc"), + (MCPAuth.authorization, "Authorization", "Custom abc"), + ], + ) async def test_raw_static_credentials_are_forwarded_unchanged( - self, auth_type: MCPAuthType, slot: str, value: str, + self, + auth_type: MCPAuthType, + slot: str, + value: str, ) -> None: - server = MCPServer(server_id="raw-key", name="raw-key", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=auth_type, authentication_token=value) + server = MCPServer( + server_id="raw-key", + name="raw-key", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=auth_type, + authentication_token=value, + ) client = await MCPServerManager()._create_mcp_client(server) assert client._resolved_auth is not None request = httpx.Request("GET", server.url) @@ -13632,17 +13816,24 @@ class TestProtectedCredentialPreparation: @pytest.mark.parametrize("value", ["Bearer", "basic", "token", "ApiKey", " bEaReR ", "\tTOKEN\t"]) @pytest.mark.parametrize("source", ["configured", "caller", "forwarded"]) async def test_raw_authorization_rejects_bare_schemes_before_dispatch( - self, respx_mock: MockRouter, value: str, source: str, + self, + respx_mock: MockRouter, + value: str, + source: str, ) -> None: server: Final = MCPServer( - server_id="raw-empty", name="raw-empty", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.authorization, + server_id="raw-empty", + name="raw-empty", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.authorization, authentication_token=value if source == "configured" else None, ) destination: Final = respx_mock.route().respond(200) with pytest.raises(HTTPException, match="requires a usable upstream credential") as exc: await MCPServerManager()._create_mcp_client( - server, mcp_auth_header=value if source == "caller" else None, + server, + mcp_auth_header=value if source == "caller" else None, extra_headers={"Authorization": value} if source == "forwarded" else None, ) assert exc.value.status_code == 500 @@ -13650,9 +13841,15 @@ class TestProtectedCredentialPreparation: @pytest.mark.asyncio async def test_byok_flag_cannot_bypass_incomplete_obo(self) -> None: - server = MCPServer(server_id="obo-byok", name="obo-byok", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.oauth2_token_exchange, is_byok=True, - token_exchange_endpoint="https://idp.example/token") + server = MCPServer( + server_id="obo-byok", + name="obo-byok", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.oauth2_token_exchange, + is_byok=True, + token_exchange_endpoint="https://idp.example/token", + ) with pytest.raises(HTTPException) as exc: await MCPServerManager()._create_mcp_client(server, mcp_auth_header="Bearer override") assert exc.value.status_code == 401 @@ -13660,41 +13857,66 @@ class TestProtectedCredentialPreparation: @pytest.mark.asyncio @pytest.mark.parametrize("configured,override", [(None, "Bearer usable"), ("shared", "Bearer usable")]) async def test_bearer_override_remains_usable(self, configured: str | None, override: str) -> None: - server = MCPServer(server_id="override", name="override", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.bearer_token, authentication_token=configured) + server = MCPServer( + server_id="override", + name="override", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.bearer_token, + authentication_token=configured, + ) client = await MCPServerManager()._create_mcp_client(server, mcp_auth_header=override) assert client._get_auth_headers()["Authorization"] == override @pytest.mark.asyncio @pytest.mark.parametrize("token", [None, "shared"]) async def test_empty_injected_header_cannot_satisfy_protected_auth(self, token: str | None) -> None: - server = MCPServer(server_id="empty-header", name="empty-header", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.bearer_token, authentication_token=token) + server = MCPServer( + server_id="empty-header", + name="empty-header", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.bearer_token, + authentication_token=token, + ) with pytest.raises(HTTPException) as exc: await MCPServerManager()._create_mcp_client(server, extra_headers={"authorization": " "}) assert exc.value.status_code == 500 @pytest.mark.asyncio async def test_custom_slot_uses_its_actual_credential(self) -> None: - server = MCPServer(server_id="custom", name="custom", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.api_key, - upstream_token_header="X-Custom", authentication_token="key") + server = MCPServer( + server_id="custom", + name="custom", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.api_key, + upstream_token_header="X-Custom", + authentication_token="key", + ) client = await MCPServerManager()._create_mcp_client(server, extra_headers={"X-Trace": "trace"}) assert client._credential_slot == "X-Custom" assert await client.discovery_auth_fingerprint() @pytest.mark.asyncio - @pytest.mark.parametrize("static_headers,accepted", [ - ({"apikey": "static-key"}, True), - ({"apikey": ""}, False), - ({"X-Tenant": "tenant"}, True), - ]) + @pytest.mark.parametrize( + "static_headers,accepted", + [ + ({"apikey": "static-key"}, True), + ({"apikey": ""}, False), + ({"X-Tenant": "tenant"}, True), + ], + ) async def test_api_key_carried_by_static_header_passes_fail_closed_check( self, static_headers: dict[str, str], accepted: bool ) -> None: server: Final = MCPServer( - server_id="static-slot", name="static-slot", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.api_key, static_headers=static_headers, + server_id="static-slot", + name="static-slot", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.api_key, + static_headers=static_headers, ) if not accepted: with pytest.raises(HTTPException) as exc: @@ -13706,21 +13928,36 @@ class TestProtectedCredentialPreparation: assert all(request.headers[name] == value for name, value in static_headers.items()) @pytest.mark.asyncio - @pytest.mark.parametrize("static,forwarded,caller", [ - ({"X-API-Key": "static"}, {"x-api-key": "forwarded"}, None), - ({}, {"X-API-Key": "forwarded"}, None), - ({}, None, "ApiKey caller"), - ({"X-API-Key": "static"}, {"Authorization": ""}, None), - ]) + @pytest.mark.parametrize( + "static,forwarded,caller", + [ + ({"X-API-Key": "static"}, {"x-api-key": "forwarded"}, None), + ({}, {"X-API-Key": "forwarded"}, None), + ({}, None, "ApiKey caller"), + ({"X-API-Key": "static"}, {"Authorization": ""}, None), + ], + ) async def test_openapi_static_credentials_remain_supported( - self, respx_mock: MockRouter, monkeypatch: pytest.MonkeyPatch, - static: dict[str, str], forwarded: dict[str, str] | None, caller: str | None + self, + respx_mock: MockRouter, + monkeypatch: pytest.MonkeyPatch, + static: dict[str, str], + forwarded: dict[str, str] | None, + caller: str | None, ) -> None: from litellm.proxy._experimental.mcp_server.openapi_to_mcp_generator import ( - _request_auth_header, _request_extra_headers, create_tool_function, + _request_auth_header, + _request_extra_headers, + create_tool_function, ) + tool: Final = create_tool_function( - "/echo", "get", {}, "https://upstream.example", headers=static, auth_type=MCPAuth.api_key, + "/echo", + "get", + {}, + "https://upstream.example", + headers=static, + auth_type=MCPAuth.api_key, ) monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True") destination: Final = respx_mock.get("https://upstream.example/echo").respond(200, text="authenticated") @@ -13754,8 +13991,13 @@ class TestProtectedCredentialPreparation: self.closed = True auth = CancelledAuth() - server = MCPServer(server_id="cancel", name="cancel", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.api_key) + server = MCPServer( + server_id="cancel", + name="cancel", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.api_key, + ) client = MCPClient(server_url=server.url, auth_type=MCPAuth.api_key, resolved_auth=auth) with pytest.raises(asyncio.CancelledError): await prepare_mcp_client(server, client) @@ -13764,8 +14006,14 @@ class TestProtectedCredentialPreparation: @pytest.mark.asyncio @pytest.mark.parametrize("auth_type", [MCPAuth.basic, MCPAuth.token, MCPAuth.authorization]) async def test_other_static_schemes_reject_whitespace_credentials(self, auth_type: MCPAuthType) -> None: - server = MCPServer(server_id="blank-static", name="blank-static", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=auth_type, authentication_token=" ") + server = MCPServer( + server_id="blank-static", + name="blank-static", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=auth_type, + authentication_token=" ", + ) with pytest.raises(HTTPException) as exc: await MCPServerManager()._create_mcp_client(server) assert exc.value.status_code == 500 @@ -13773,8 +14021,13 @@ class TestProtectedCredentialPreparation: @pytest.mark.asyncio @pytest.mark.parametrize("header", ["Basic", "Basic @@@", "Other abc", "Basic QmFzaWM=", "Basic bm8tY29sb24="]) async def test_basic_headers_without_usable_credentials_reject(self, header: str) -> None: - server = MCPServer(server_id="bad-basic", name="bad-basic", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.basic) + server = MCPServer( + server_id="bad-basic", + name="bad-basic", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.basic, + ) with pytest.raises(HTTPException) as exc: await MCPServerManager()._create_mcp_client(server, extra_headers={"Authorization": header}) assert exc.value.status_code == 500 @@ -13783,34 +14036,48 @@ class TestProtectedCredentialPreparation: @pytest.mark.parametrize("value", ["Basic", "Basic ", "basic"]) @pytest.mark.parametrize("source", ["configured", "caller"]) async def test_basic_scheme_alone_is_not_a_credential(self, value: str, source: str) -> None: - server = MCPServer(server_id="basic-scheme", name="basic-scheme", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.basic, - authentication_token=value if source == "configured" else None) + server = MCPServer( + server_id="basic-scheme", + name="basic-scheme", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.basic, + authentication_token=value if source == "configured" else None, + ) with pytest.raises(HTTPException) as exc: await MCPServerManager()._create_mcp_client(server, mcp_auth_header=value if source == "caller" else None) assert exc.value.status_code == 500 @pytest.mark.asyncio - @pytest.mark.parametrize("auth_type,value,default_slot", [ - (MCPAuth.api_key, "fixture-key", "X-API-Key"), - (MCPAuth.bearer_token, "fixture-key", "Authorization"), - (MCPAuth.basic, "user:pass", "Authorization"), - (MCPAuth.token, "fixture-key", "Authorization"), - (MCPAuth.authorization, "fixture-key", "Authorization"), - ]) + @pytest.mark.parametrize( + "auth_type,value,default_slot", + [ + (MCPAuth.api_key, "fixture-key", "X-API-Key"), + (MCPAuth.bearer_token, "fixture-key", "Authorization"), + (MCPAuth.basic, "user:pass", "Authorization"), + (MCPAuth.token, "fixture-key", "Authorization"), + (MCPAuth.authorization, "fixture-key", "Authorization"), + ], + ) @pytest.mark.parametrize("source", ["configured", "caller"]) async def test_usable_credential_survives_an_empty_alternate_header( self, auth_type: MCPAuthType, value: str, default_slot: str, source: str ) -> None: server: Final = MCPServer( - server_id="alternate", name="alternate", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=auth_type, upstream_token_header="X-Custom", + server_id="alternate", + name="alternate", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=auth_type, + upstream_token_header="X-Custom", authentication_token=value if source == "configured" else None, ) empty_slot: Final = default_slot if source == "configured" else "X-Custom" selected_slot: Final = "X-Custom" if source == "configured" else default_slot client: Final = await MCPServerManager()._create_mcp_client( - server, mcp_auth_header=value if source == "caller" else None, extra_headers={empty_slot: ""}, + server, + mcp_auth_header=value if source == "caller" else None, + extra_headers={empty_slot: ""}, ) request: Final = await client.prepare_request_auth() assert request.headers[selected_slot] @@ -13819,8 +14086,12 @@ class TestProtectedCredentialPreparation: @pytest.mark.asyncio async def test_empty_custom_and_default_headers_do_not_satisfy_auth(self) -> None: server: Final = MCPServer( - server_id="both-empty", name="both-empty", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.api_key, upstream_token_header="X-Custom", + server_id="both-empty", + name="both-empty", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.api_key, + upstream_token_header="X-Custom", ) with pytest.raises(HTTPException) as exc: await MCPServerManager()._create_mcp_client(server, extra_headers={"X-Custom": "", "X-API-Key": ""}) @@ -13833,12 +14104,17 @@ class TestProtectedCredentialPreparation: self, custom_slot: str | None, source: str ) -> None: server: Final = MCPServer( - server_id="caller-auth", name="caller-auth", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.api_key, upstream_token_header=custom_slot, + server_id="caller-auth", + name="caller-auth", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.api_key, + upstream_token_header=custom_slot, ) headers: Final = {"Authorization": "Bearer caller-credential", "X-API-Key": ""} client: Final = await MCPServerManager()._create_mcp_client( - server, mcp_auth_header=headers if source == "caller" else None, + server, + mcp_auth_header=headers if source == "caller" else None, extra_headers=headers if source == "forwarded" else None, ) request: Final = await client.prepare_request_auth() @@ -13847,14 +14123,29 @@ class TestProtectedCredentialPreparation: assert custom_slot is None or custom_slot not in request.headers @pytest.mark.asyncio - @pytest.mark.parametrize("value", [ - "", " ", "Bearer", "Basic", "token", "ApiKey", - "Bearer Bearer", "ApiKey ApiKey", "token token", "bEaReR BEARER", "aPiKeY\tAPIKEY", - ]) + @pytest.mark.parametrize( + "value", + [ + "", + " ", + "Bearer", + "Basic", + "token", + "ApiKey", + "Bearer Bearer", + "ApiKey ApiKey", + "token token", + "bEaReR BEARER", + "aPiKeY\tAPIKEY", + ], + ) async def test_api_key_rejects_authorization_without_a_credential(self, value: str) -> None: server: Final = MCPServer( - server_id="caller-empty", name="caller-empty", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.api_key, + server_id="caller-empty", + name="caller-empty", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.api_key, ) with pytest.raises(HTTPException) as exc: await MCPServerManager()._create_mcp_client(server, mcp_auth_header={"Authorization": value}) @@ -13865,8 +14156,11 @@ class TestProtectedCredentialPreparation: @pytest.mark.parametrize("source", ["configured", "caller"]) async def test_basic_requires_a_username_password_separator(self, value: str, source: str) -> None: server: Final = MCPServer( - server_id="basic-pair", name="basic-pair", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.basic, + server_id="basic-pair", + name="basic-pair", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.basic, authentication_token=value if source == "configured" else None, ) with pytest.raises(HTTPException) as exc: @@ -13879,8 +14173,12 @@ class TestProtectedCredentialPreparation: import base64 server: Final = MCPServer( - server_id="basic-valid", name="basic-valid", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=MCPAuth.basic, authentication_token=value, + server_id="basic-valid", + name="basic-valid", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=MCPAuth.basic, + authentication_token=value, ) client: Final = await MCPServerManager()._create_mcp_client(server) request: Final = await client.prepare_request_auth() @@ -13889,17 +14187,27 @@ class TestProtectedCredentialPreparation: assert base64.b64decode(encoded) == value.encode() @pytest.mark.asyncio - @pytest.mark.parametrize("auth_type,value", [ - (MCPAuth.bearer_token, "Bearer"), (MCPAuth.bearer_token, "Bearer "), (MCPAuth.bearer_token, "bearer"), - (MCPAuth.token, "token"), (MCPAuth.token, "token "), (MCPAuth.token, "TOKEN"), - ]) + @pytest.mark.parametrize( + "auth_type,value", + [ + (MCPAuth.bearer_token, "Bearer"), + (MCPAuth.bearer_token, "Bearer "), + (MCPAuth.bearer_token, "bearer"), + (MCPAuth.token, "token"), + (MCPAuth.token, "token "), + (MCPAuth.token, "TOKEN"), + ], + ) @pytest.mark.parametrize("source", ["configured", "caller"]) async def test_static_scheme_only_input_cannot_hide_behind_rendered_prefix( self, auth_type: MCPAuthType, value: str, source: str ) -> None: server: Final = MCPServer( - server_id="empty-scheme", name="empty-scheme", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=auth_type, + server_id="empty-scheme", + name="empty-scheme", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=auth_type, authentication_token=value if source == "configured" else None, ) with pytest.raises(HTTPException) as exc: @@ -13907,17 +14215,24 @@ class TestProtectedCredentialPreparation: assert exc.value.status_code == 500 @pytest.mark.asyncio - @pytest.mark.parametrize("auth_type,value,expected", [ - (MCPAuth.bearer_token, "token", "Bearer token"), - (MCPAuth.bearer_token, "Bearertoken", "Bearer Bearertoken"), - (MCPAuth.token, "tokenish", "token tokenish"), - ]) + @pytest.mark.parametrize( + "auth_type,value,expected", + [ + (MCPAuth.bearer_token, "token", "Bearer token"), + (MCPAuth.bearer_token, "Bearertoken", "Bearer Bearertoken"), + (MCPAuth.token, "tokenish", "token tokenish"), + ], + ) async def test_static_credentials_that_resemble_schemes_remain_usable( self, auth_type: MCPAuthType, value: str, expected: str ) -> None: server: Final = MCPServer( - server_id="real-token", name="real-token", url="https://upstream.example/mcp", - transport=MCPTransport.http, auth_type=auth_type, authentication_token=value, + server_id="real-token", + name="real-token", + url="https://upstream.example/mcp", + transport=MCPTransport.http, + auth_type=auth_type, + authentication_token=value, ) client: Final = await MCPServerManager()._create_mcp_client(server) request: Final = await client.prepare_request_auth() @@ -13956,16 +14271,31 @@ async def test_request_selected_during_guardrail_runs_concurrently_with_tool(mon registry.register_tool("observer-execute", "Execute", {"type": "object"}, upstream) monkeypatch.setattr(tool_registry, "global_mcp_tool_registry", registry) manager = MCPServerManager() - manager.registry = {"observer": MCPServer( - server_id="observer", name="observer", server_name="observer", transport="http", - url="https://observer.example/mcp", spec_path="observer.json", auth_type="none", - )} + manager.registry = { + "observer": MCPServer( + server_id="observer", + name="observer", + server_name="observer", + transport="http", + url="https://observer.example/mcp", + spec_path="observer.json", + auth_type="none", + ) + } manager.tool_name_to_mcp_server_name_mapping = {"observer-execute": "observer"} - result = await asyncio.wait_for(manager.call_tool( - server_name="observer", name="execute", arguments={"text": "hello"}, - user_api_key_auth=UserAPIKeyAuth(), proxy_logging_obj=ProxyLogging(user_api_key_cache=DualCache()), - guardrail_context=MCPRequestContext.resolve_guardrail_context({"metadata": {"guardrails": ["observe"] if selected else []}}), - ), timeout=5) + result = await asyncio.wait_for( + manager.call_tool( + server_name="observer", + name="execute", + arguments={"text": "hello"}, + user_api_key_auth=UserAPIKeyAuth(), + proxy_logging_obj=ProxyLogging(user_api_key_cache=DualCache()), + guardrail_context=MCPRequestContext.resolve_guardrail_context( + {"metadata": {"guardrails": ["observe"] if selected else []}} + ), + ), + timeout=5, + ) assert tool_started.is_set() assert guardrail_started.is_set() is selected assert result.is_error is False