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 48c09f6e456..fa76af3dca8 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 @@ -911,6 +911,100 @@ class TestMCPServerManager: ] assert scopes == ["read", "write"] + @pytest.mark.asyncio + async def test_fetch_oauth_metadata_from_resource_parses_strict_rfc9728_payload( + self, + ): + """A fully spec-compliant RFC 9728 payload (with ``resource``) takes the + strict ProtectedResourceMetadata path, which normalizes the URLs.""" + manager = MCPServerManager() + + mock_response = MagicMock() + mock_response.json.return_value = { + "resource": "https://protected.example.com/mcp", + "authorization_servers": ["https://auth.example.com"], + "scopes_supported": ["read", "write"], + } + mock_response.raise_for_status = MagicMock() + + mock_client = MagicMock() + mock_client.get = AsyncMock(return_value=mock_response) + + with patch( + "litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client", + return_value=mock_client, + ): + servers, scopes = await manager._fetch_oauth_metadata_from_resource( + "https://protected.example.com/.well-known/oauth-protected-resource/mcp", + "https://protected.example.com/mcp", + ) + + assert servers == ["https://auth.example.com/"] + assert scopes == ["read", "write"] + + @pytest.mark.asyncio + async def test_discovery_payloads_that_are_not_json_objects_yield_no_metadata( + self, + ): + manager = MCPServerManager() + + mock_response = MagicMock() + mock_response.json.return_value = ["not", "an", "object"] + mock_response.raise_for_status = MagicMock() + + mock_client = MagicMock() + mock_client.get = AsyncMock(return_value=mock_response) + + with patch( + "litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client", + return_value=mock_client, + ): + servers, scopes = await manager._fetch_oauth_metadata_from_resource( + "https://protected.example.com/.well-known/oauth-protected-resource", + "https://protected.example.com/mcp", + ) + + assert servers == [] + assert scopes is None + assert manager._oauth_metadata_from_asm_payload(["not", "an", "object"]) is None + + @pytest.mark.asyncio + async def test_fetch_single_authorization_server_metadata_parses_strict_rfc8414_payload( + self, + ): + """A fully spec-compliant RFC 8414 payload (with ``issuer``) takes the + strict OAuthMetadata path and maps all endpoint fields.""" + manager = MCPServerManager() + + mock_response = MagicMock() + mock_response.json.return_value = { + "issuer": "https://provider.example.com", + "authorization_endpoint": "https://provider.example.com/authorize", + "token_endpoint": "https://provider.example.com/token", + "registration_endpoint": "https://provider.example.com/register", + "response_types_supported": ["code"], + "scopes_supported": ["mcp.read"], + } + mock_response.raise_for_status = MagicMock() + + mock_client = MagicMock() + mock_client.get = AsyncMock(return_value=mock_response) + + with patch( + "litellm.proxy._experimental.mcp_server.mcp_server_manager.get_async_httpx_client", + return_value=mock_client, + ): + result = await manager._fetch_single_authorization_server_metadata( + "https://provider.example.com", + "https://provider.example.com", + ) + + assert result is not None + assert result.authorization_url == "https://provider.example.com/authorize" + assert result.token_url == "https://provider.example.com/token" + assert result.registration_url == "https://provider.example.com/register" + assert result.scopes == ["mcp.read"] + @pytest.mark.asyncio async def test_descovery_metadata_probes_well_known_when_server_does_not_challenge( self,