FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Improve OAuth protected resource metadata URL construction per RFC 9728 by shulkx · Pull Request #1407 · modelcontextprotocol/python-sdk · GitHub

This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ class ResourceServerSettings(BaseSettings):
# Server settings
host: str = "localhost"
port: int = 8001
server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:8001")
server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:8001/mcp")

# Authorization Server settings
auth_server_url: AnyHttpUrl = AnyHttpUrl("http://localhost:9000")
Expand Down Expand Up @@ -137,7 +137,7 @@ def main(port: int, auth_server: str, transport: Literal["sse", "streamable-http

# Create settings
host = "localhost"
server_url = f"http://{host}:{port}"
server_url = f"http://{host}:{port}/mcp"
settings = ResourceServerSettings(
host=host,
port=port,
Expand Down
28 changes: 27 additions & 1 deletion src/mcp/server/auth/routes.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from collections.abc import Awaitable, Callable
from typing import Any
from urllib.parse import urlparse

from pydantic import AnyHttpUrl
from starlette.middleware.cors import CORSMiddleware
Expand Down Expand Up @@ -186,6 +187,25 @@ def build_metadata(
return metadata


def build_resource_metadata_url(resource_server_url: AnyHttpUrl) -> AnyHttpUrl:
"""
Build RFC 9728 compliant protected resource metadata URL.

Inserts /.well-known/oauth-protected-resource between host and resource path
as specified in RFC 9728 §3.1.

Args:
resource_server_url: The resource server URL (e.g., https://example.com/mcp)

Returns:
The metadata URL (e.g., https://example.com/.well-known/oauth-protected-resource/mcp)
"""
parsed = urlparse(str(resource_server_url))
# Handle trailing slash: if path is just "/", treat as empty
resource_path = parsed.path if parsed.path != "/" else ""
return AnyHttpUrl(f"{parsed.scheme}://{parsed.netloc}/.well-known/oauth-protected-resource{resource_path}")


def create_protected_resource_routes(
resource_url: AnyHttpUrl,
authorization_servers: list[AnyHttpUrl],
Expand Down Expand Up @@ -218,9 +238,15 @@ def create_protected_resource_routes(

handler = ProtectedResourceMetadataHandler(metadata)

# RFC 9728 §3.1: Register route at /.well-known/oauth-protected-resource + resource path
metadata_url = build_resource_metadata_url(resource_url)
# Extract just the path part for route registration
parsed = urlparse(str(metadata_url))
well_known_path = parsed.path

return [
Route(
"/.well-known/oauth-protected-resource",
well_known_path,
endpoint=cors_middleware(handler.handle, ["GET", "OPTIONS"]),
methods=["GET", "OPTIONS"],
)
Expand Down
38 changes: 13 additions & 25 deletions src/mcp/server/fastmcp/server.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -824,11 +824,10 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):
# Determine resource metadata URL
resource_metadata_url = None
if self.settings.auth and self.settings.auth.resource_server_url:
from pydantic import AnyHttpUrl
from mcp.server.auth.routes import build_resource_metadata_url

resource_metadata_url = AnyHttpUrl(
str(self.settings.auth.resource_server_url).rstrip("/") + "/.well-known/oauth-protected-resource"
)
# Build compliant metadata URL for WWW-Authenticate header
resource_metadata_url = build_resource_metadata_url(self.settings.auth.resource_server_url)

# Auth is enabled, wrap the endpoints with RequireAuthMiddleware
routes.append(
Expand Down Expand Up @@ -937,11 +936,10 @@ def streamable_http_app(self) -> Starlette:
# Determine resource metadata URL
resource_metadata_url = None
if self.settings.auth and self.settings.auth.resource_server_url:
from pydantic import AnyHttpUrl
from mcp.server.auth.routes import build_resource_metadata_url

resource_metadata_url = AnyHttpUrl(
str(self.settings.auth.resource_server_url).rstrip("/") + "/.well-known/oauth-protected-resource"
)
# Build compliant metadata URL for WWW-Authenticate header
resource_metadata_url = build_resource_metadata_url(self.settings.auth.resource_server_url)

routes.append(
Route(
Expand All @@ -960,23 +958,13 @@ def streamable_http_app(self) -> Starlette:

# Add protected resource metadata endpoint if configured as RS
if self.settings.auth and self.settings.auth.resource_server_url:
from mcp.server.auth.handlers.metadata import ProtectedResourceMetadataHandler
from mcp.server.auth.routes import cors_middleware
from mcp.shared.auth import ProtectedResourceMetadata

protected_resource_metadata = ProtectedResourceMetadata(
resource=self.settings.auth.resource_server_url,
authorization_servers=[self.settings.auth.issuer_url],
scopes_supported=self.settings.auth.required_scopes,
)
routes.append(
Route(
"/.well-known/oauth-protected-resource",
endpoint=cors_middleware(
ProtectedResourceMetadataHandler(protected_resource_metadata).handle,
["GET", "OPTIONS"],
),
methods=["GET", "OPTIONS"],
from mcp.server.auth.routes import create_protected_resource_routes

routes.extend(
create_protected_resource_routes(
resource_url=self.settings.auth.resource_server_url,
authorization_servers=[self.settings.auth.issuer_url],
scopes_supported=self.settings.auth.required_scopes,
)
)

Expand Down
154 changes: 150 additions & 4 deletions tests/server/auth/test_protected_resource.py
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
from pydantic import AnyHttpUrl
from starlette.applications import Starlette

from mcp.server.auth.routes import create_protected_resource_routes
from mcp.server.auth.routes import build_resource_metadata_url, create_protected_resource_routes


@pytest.fixture
Expand Down Expand Up @@ -36,10 +36,11 @@ async def test_client(test_app: Starlette):


@pytest.mark.anyio
async def test_metadata_endpoint(test_client: httpx.AsyncClient):
"""Test the OAuth 2.0 Protected Resource metadata endpoint."""
async def test_metadata_endpoint_with_path(test_client: httpx.AsyncClient):
"""Test the OAuth 2.0 Protected Resource metadata endpoint for path-based resource."""

response = await test_client.get("/.well-known/oauth-protected-resource")
# For resource with path "/resource", metadata should be accessible at the path-aware location
response = await test_client.get("/.well-known/oauth-protected-resource/resource")
assert response.json() == snapshot(
{
"resource": "https://example.com/resource",
Expand All @@ -50,3 +51,148 @@ async def test_metadata_endpoint(test_client: httpx.AsyncClient):
"bearer_methods_supported": ["header"],
}
)


@pytest.mark.anyio
async def test_metadata_endpoint_root_path_returns_404(test_client: httpx.AsyncClient):
"""Test that root path returns 404 for path-based resource."""

# Root path should return 404 for path-based resources
response = await test_client.get("/.well-known/oauth-protected-resource")
assert response.status_code == 404


@pytest.fixture
def root_resource_app():
"""Fixture to create protected resource routes for root-level resource."""

# Create routes for a resource without path component
protected_resource_routes = create_protected_resource_routes(
resource_url=AnyHttpUrl("https://example.com"),
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
scopes_supported=["read"],
resource_name="Root Resource",
)

app = Starlette(routes=protected_resource_routes)
return app


@pytest.fixture
async def root_resource_client(root_resource_app: Starlette):
"""Fixture to create an HTTP client for the root resource app."""
async with httpx.AsyncClient(
transport=httpx.ASGITransport(app=root_resource_app), base_url="https://mcptest.com"
) as client:
yield client


@pytest.mark.anyio
async def test_metadata_endpoint_without_path(root_resource_client: httpx.AsyncClient):
"""Test metadata endpoint for root-level resource."""

# For root resource, metadata should be at standard location
response = await root_resource_client.get("/.well-known/oauth-protected-resource")
assert response.status_code == 200
assert response.json() == snapshot(
{
"resource": "https://example.com/",
"authorization_servers": ["https://auth.example.com/"],
"scopes_supported": ["read"],
"resource_name": "Root Resource",
"bearer_methods_supported": ["header"],
}
)


class TestMetadataUrlConstruction:
"""Test URL construction utility function."""

def test_url_without_path(self):
"""Test URL construction for resource without path component."""
resource_url = AnyHttpUrl("https://example.com")
result = build_resource_metadata_url(resource_url)
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"

def test_url_with_path_component(self):
"""Test URL construction for resource with path component."""
resource_url = AnyHttpUrl("https://example.com/mcp")
result = build_resource_metadata_url(resource_url)
assert str(result) == "https://example.com/.well-known/oauth-protected-resource/mcp"

def test_url_with_trailing_slash_only(self):
"""Test URL construction for resource with trailing slash only."""
resource_url = AnyHttpUrl("https://example.com/")
result = build_resource_metadata_url(resource_url)
# Trailing slash should be treated as empty path
assert str(result) == "https://example.com/.well-known/oauth-protected-resource"

@pytest.mark.parametrize(
"resource_url,expected_url",
[
("https://example.com", "https://example.com/.well-known/oauth-protected-resource"),
("https://example.com/", "https://example.com/.well-known/oauth-protected-resource"),
("https://example.com/mcp", "https://example.com/.well-known/oauth-protected-resource/mcp"),
("http://localhost:8001/mcp", "http://localhost:8001/.well-known/oauth-protected-resource/mcp"),
],
)
def test_various_resource_configurations(self, resource_url: str, expected_url: str):
"""Test URL construction with various resource configurations."""
result = build_resource_metadata_url(AnyHttpUrl(resource_url))
assert str(result) == expected_url


class TestRouteConsistency:
"""Test consistency between URL generation and route registration."""

def test_route_path_matches_metadata_url(self):
"""Test that route path matches the generated metadata URL."""
resource_url = AnyHttpUrl("https://example.com/mcp")

# Generate metadata URL
metadata_url = build_resource_metadata_url(resource_url)

# Create routes
routes = create_protected_resource_routes(
resource_url=resource_url,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)

# Extract path from metadata URL
from urllib.parse import urlparse

metadata_path = urlparse(str(metadata_url)).path

# Verify consistency
assert len(routes) == 1
assert routes[0].path == metadata_path

@pytest.mark.parametrize(
"resource_url,expected_path",
[
("https://example.com", "/.well-known/oauth-protected-resource"),
("https://example.com/", "/.well-known/oauth-protected-resource"),
("https://example.com/mcp", "/.well-known/oauth-protected-resource/mcp"),
],
)
def test_consistent_paths_for_various_resources(self, resource_url: str, expected_path: str):
"""Test that URL generation and route creation are consistent."""
resource_url_obj = AnyHttpUrl(resource_url)

# Test URL generation
metadata_url = build_resource_metadata_url(resource_url_obj)
from urllib.parse import urlparse

url_path = urlparse(str(metadata_url)).path

# Test route creation
routes = create_protected_resource_routes(
resource_url=resource_url_obj,
authorization_servers=[AnyHttpUrl("https://auth.example.com")],
)
route_path = routes[0].path

# Both should match expected path
assert url_path == expected_path
assert route_path == expected_path
assert url_path == route_path
Loading

Back | FazBrowse Home | New Git URL