refactor(mcp): build agent 365 protected resource metadata immutably to satisfy the type discipline gate

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

litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py

@@ -5,13 +5,14 @@ import secrets
 import time
 from collections.abc import Callable, Mapping
 from datetime import datetime, timezone
-from typing import TYPE_CHECKING, Any, Final, Literal, Optional
+from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypedDict
 from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse
 
 import httpx
 from fastapi import APIRouter, Depends, Form, HTTPException, Request
 from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
 from pydantic import BaseModel, ConfigDict, Field, SecretStr, ValidationError
+from typing_extensions import ReadOnly
 
 from litellm._logging import verbose_logger
 from litellm.caching.in_memory_cache import InMemoryCache
@@ -2375,7 +2376,7 @@ async def _build_oauth_protected_resource_response(
     request: Request,
     mcp_server_name: str | None,
     use_standard_pattern: bool,
-) -> dict:
+) -> Mapping[str, object]:
     """
     Build OAuth protected resource response with the appropriate URL pattern.
 
@@ -2481,11 +2482,12 @@ async def _build_oauth_protected_resource_response(
 
     agent_365_issuers: Final = agent_365_authorization_servers(mcp_server, None) if mcp_server else ()
     if mcp_server is not None and agent_365_issuers:
-        return {
-            "authorization_servers": list(agent_365_issuers),
+        agent_365_metadata: Final[_ProtectedResourceMetadata] = {
+            "authorization_servers": agent_365_issuers,
             "resource": resource_url,
-            "scopes_supported": list(mcp_server.scopes or ()),
+            "scopes_supported": tuple(mcp_server.scopes or ()),
         }
+        return agent_365_metadata
 
     if explicitly_named and mcp_server is not None and mcp_server.advertises_gateway_authorization_server:
         return {
@@ -2506,6 +2508,12 @@ async def _build_oauth_protected_resource_response(
     }
 
 
+class _ProtectedResourceMetadata(TypedDict):
+    authorization_servers: ReadOnly[tuple[str, ...]]
+    resource: ReadOnly[str]
+    scopes_supported: ReadOnly[tuple[str, ...]]
+
+
 def _obo_protected_resource_response(mcp_server: MCPServer | None, resource_url: str) -> dict | None:
     """The OBO (token_exchange) PRM, or None when this server is not OBO / no issuer is configured.
 

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -1779,7 +1779,7 @@ class MCPServerManager:
             "gmail_send_email": "zapier_mcp_server",
         }
         """
-        self._listed_tools_by_server_id: dict[str, dict[str, MCPTool]] = {}
+        self._listed_tools_by_server_id: dict[str, Mapping[str, MCPTool]] = {}  # mutable-ok: refreshed per tools/list
         self._upstream_initialize_instructions_by_server_id: dict[str, str] = {}
         # Per-server monotonic timestamp of last upstream prefetch attempt (success,
         # empty result, or failure). Used to throttle re-probes for servers that do
@@ -4244,7 +4244,7 @@ class MCPServerManager:
                 # time producing "test_petstore-test_petstore-getinventory".
                 prefix: Final = get_server_prefix(server)
                 sep: Final = MCP_TOOL_PREFIX_SEPARATOR
-                bare_tools: Final = [
+                bare_tools: Final = [  # mutable-ok: returned through the list[MCPTool] listing contract
                     (
                         t.model_copy(update={"name": t.name[len(prefix) + len(sep) :]})
                         if t.name.startswith(f"{prefix}{sep}")
@@ -4252,7 +4252,7 @@ class MCPServerManager:
                     )
                     for t in tools
                 ]
-                self._listed_tools_by_server_id[server.server_id] = {t.name: t for t in bare_tools}
+                self._listed_tools_by_server_id[server.server_id] = MappingProxyType({t.name: t for t in bare_tools})
                 return tools if add_prefix else bare_tools
             else:
                 tools = await self._fetch_tools_with_timeout(client, server.name)
@@ -5134,7 +5134,7 @@ class MCPServerManager:
             for spelling in iter_known_tool_name_spellings(original_name, server):
                 self.tool_name_to_mcp_server_name_mapping[spelling] = prefix
 
-        self._listed_tools_by_server_id[server.server_id] = {tool.name: tool for tool in tools}
+        self._listed_tools_by_server_id[server.server_id] = MappingProxyType({tool.name: tool for tool in tools})
         verbose_logger.info("Successfully fetched %s tools from server %s", len(prefixed_tools), server.name)
         return prefixed_tools
 

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

@@ -10,6 +10,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from fastapi import HTTPException
+from fastapi.encoders import jsonable_encoder
 
 from litellm.types.mcp import MCPAuth
 
@@ -7148,7 +7149,7 @@ async def _agent_365_gated_prm(scopes):
 @pytest.mark.asyncio
 async def test_agent_365_gated_server_prm_names_the_entra_tenant(agent_365_guardrail):
     response = await _agent_365_gated_prm(scopes=["api://gateway-app/access_as_user"])
-    assert response == {
+    assert jsonable_encoder(response) == {
         "authorization_servers": ["https://login.microsoftonline.com/tenant-abc/v2.0"],
         "resource": "https://litellm.example.com/mcp/tools",
         "scopes_supported": ["api://gateway-app/access_as_user"],