style(mcp): satisfy lint and type budgets for the SDK 2 port

Format the ported files, annotate mutable wire payloads, give the e2e
OAuth client the SDK 2 httpx2/AuthorizationCodeResult API, tighten the
transport-streams alias to the two-stream SDK 2 shape, and add a
test-quality reason for the MockTransport factory injection.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

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)

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",

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"]

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.

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)

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."

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": {

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"

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

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()