Fix OAuth examples to use complete resource URLs

docs/run/authorization.md

@@ -29,7 +29,7 @@ The SDK has no opinion about what a valid token looks like. You tell it, by impl
 `AuthSettings` is the public face of your resource server:
 
 * `issuer_url`: the authorization server that issues your tokens.
-* `resource_server_url`: the public URL of this MCP endpoint. It names *which* resource a token is for, and it's where the discovery document lives.
+* `resource_server_url`: the complete public URL of this MCP endpoint, including its path (for example, `/mcp`). It names *which* resource a token is for, and it's where the discovery document lives. Use the externally visible URL when a proxy or mounted application changes the public path; the SDK does not infer it from the internal route.
 * `required_scopes`: every token must carry all of them.
 
 !!! tip

examples/clients/simple-auth-client/mcp_simple_auth_client/main.py

@@ -214,7 +214,7 @@ async def _default_redirect_handler(authorization_url: str) -> None:
             # Create OAuth authentication handler using the new interface
             # Use client_metadata_url to enable CIMD when the server supports it
             oauth_auth = OAuthClientProvider(
-                server_url=self.server_url.replace("/mcp", ""),
+                server_url=self.server_url,
                 client_metadata=OAuthClientMetadata.model_validate(client_metadata_dict),
                 storage=InMemoryTokenStorage(),
                 redirect_handler=_default_redirect_handler,

examples/servers/simple-auth/README.md

@@ -38,6 +38,16 @@ uv run mcp-simple-auth-rs --port=8001 --auth-server=http://localhost:9000 --tran
 
 ```
 
+The resource identifier follows the selected transport endpoint: `/mcp` for
+Streamable HTTP and `/sse` for SSE.
+
+For SSE, both the transport and protected-resource metadata use `/sse`:
+
+```bash
+uv run mcp-simple-auth-rs --port=8001 --auth-server=http://localhost:9000 --transport=sse
+curl http://localhost:8001/.well-known/oauth-protected-resource/sse
+```
+
 ### Step 3: Test with Client
 
 ```bash
@@ -53,12 +63,12 @@ MCP_SERVER_PORT=8001 MCP_TRANSPORT_TYPE=streamable-http uv run mcp-simple-auth-c
 **Client → Resource Server:**
 
 ```bash
-curl http://localhost:8001/.well-known/oauth-protected-resource
+curl http://localhost:8001/.well-known/oauth-protected-resource/mcp
 ```
 
 ```json
 {
-  "resource": "http://localhost:8001",
+  "resource": "http://localhost:8001/mcp",
   "authorization_servers": ["http://localhost:9000"]
 }
 ```
@@ -119,7 +129,7 @@ This ensures existing MCP servers (which could optionally act as Authorization S
 
 ```bash
 # Test Resource Server discovery endpoint (new architecture)
-curl -v http://localhost:8001/.well-known/oauth-protected-resource
+curl -v http://localhost:8001/.well-known/oauth-protected-resource/mcp
 
 # Test Authorization Server metadata
 curl -v http://localhost:9000/.well-known/oauth-authorization-server

examples/servers/simple-auth/mcp_simple_auth/server.py

@@ -110,7 +110,12 @@ async def get_time() -> dict[str, Any]:
     is_flag=True,
     help="Enable RFC 8707 resource validation",
 )
-def main(port: int, auth_server: str, transport: Literal["sse", "streamable-http"], oauth_strict: bool) -> int:
+def main(
+    port: int,
+    auth_server: str,
+    transport: Literal["sse", "streamable-http"],
+    oauth_strict: bool,
+) -> int:
     """Run the MCP Resource Server.
 
     This server:
@@ -128,7 +133,8 @@ def main(port: int, auth_server: str, transport: Literal["sse", "streamable-http
 
         # Create settings
         host = "localhost"
-        server_url = f"http://{host}:{port}/mcp"
+        transport_path = "/sse" if transport == "sse" else "/mcp"
+        server_url = f"http://{host}:{port}{transport_path}"
         settings = ResourceServerSettings(
             host=host,
             port=port,
@@ -148,8 +154,16 @@ def main(port: int, auth_server: str, transport: Literal["sse", "streamable-http
         logger.info(f"🚀 MCP Resource Server running on {settings.server_url}")
         logger.info(f"🔑 Using Authorization Server: {settings.auth_server_url}")
 
-        # Run the server - this should block and keep running
-        mcp_server.run(transport=transport, host=host, port=port)
+        # Keep the advertised resource path and the listening route in lockstep.
+        if transport == "sse":
+            mcp_server.run(transport="sse", host=host, port=port, sse_path=transport_path)
+        else:
+            mcp_server.run(
+                transport="streamable-http",
+                host=host,
+                port=port,
+                streamable_http_path=transport_path,
+            )
         logger.info("Server stopped")
         return 0
     except Exception:

examples/snippets/clients/oauth_client.py

@@ -59,7 +59,7 @@ async def handle_callback() -> AuthorizationCodeResult:
 async def main():
     """Run the OAuth client example."""
     oauth_auth = OAuthClientProvider(
-        server_url="http://localhost:8001",
+        server_url="http://localhost:8001/mcp",
         client_metadata=OAuthClientMetadata(
             client_name="Example MCP Client",
             redirect_uris=[AnyUrl("http://localhost:3000/callback")],

examples/snippets/servers/oauth_server.py

@@ -24,7 +24,7 @@ async def verify_token(self, token: str) -> AccessToken | None:
     # Auth settings for RFC 9728 Protected Resource Metadata
     auth=AuthSettings(
         issuer_url=AnyHttpUrl("https://auth.example.com"),  # Authorization Server URL
-        resource_server_url=AnyHttpUrl("http://localhost:3001"),  # This server's URL
+        resource_server_url=AnyHttpUrl("http://127.0.0.1:8000/mcp"),  # This server's MCP endpoint
         required_scopes=["user"],
     ),
 )

pyproject.toml

@@ -176,6 +176,9 @@ executionEnvironments = [
         ".",
         "examples",
     ], reportUnusedFunction = false, reportPrivateUsage = false },
+    { root = "examples/clients/simple-auth-client", extraPaths = [
+        "examples/clients/simple-auth-client",
+    ], reportUnusedFunction = false },
     { root = "examples/stories", extraPaths = [
         "examples",
     ], reportUnusedFunction = false },

src/mcp/server/auth/settings.py

@@ -37,6 +37,7 @@ class AuthSettings(BaseModel):
     # Resource Server settings (when operating as RS only)
     resource_server_url: AnyHttpUrl | None = Field(
         ...,
-        description="The URL of the MCP server to be used as the resource identifier "
-        "and base route to look up OAuth Protected Resource Metadata.",
+        description="The complete externally visible URL of the MCP endpoint, including "
+        "any path prefix and transport path. Used as the resource identifier and to locate "
+        "OAuth Protected Resource Metadata.",
     )

tests/examples/simple_auth/conftest.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+import importlib
+import sys
+from collections.abc import Callable
+from pathlib import Path
+from types import ModuleType
+
+import pytest
+
+
+@pytest.fixture
+def load_example_module() -> Callable[[Path, str], ModuleType]:
+    """Import a workspace example without requiring it in the root test environment."""
+
+    def load(package_root: Path, module_name: str) -> ModuleType:
+        original_path = sys.path.copy()
+        try:
+            sys.path.insert(0, str(package_root))
+            return importlib.import_module(module_name)
+        finally:
+            sys.path[:] = original_path
+
+    return load

tests/examples/simple_auth/test_oauth_resource_url.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator, Callable
+from contextlib import asynccontextmanager
+from pathlib import Path
+from types import ModuleType
+from typing import Protocol, cast
+
+import anyio
+import pytest
+from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+
+from mcp.client.auth import OAuthClientProvider
+from mcp.shared.message import SessionMessage
+
+CLIENT_ROOT = Path(__file__).parents[3] / "examples" / "clients" / "simple-auth-client"
+
+
+class SimpleAuthClient(Protocol):
+    def __init__(
+        self,
+        server_url: str,
+        transport_type: str = "streamable-http",
+        client_metadata_url: str | None = None,
+    ) -> None: ...
+
+    async def connect(self) -> None: ...
+
+
+class ClientModule(Protocol):
+    SimpleAuthClient: type[SimpleAuthClient]
+
+
+@pytest.mark.anyio
+async def test_oauth_client_preserves_the_complete_connection_url(
+    monkeypatch: pytest.MonkeyPatch,
+    load_example_module: Callable[[Path, str], ModuleType],
+) -> None:
+    """The example passes the opaque MCP endpoint unchanged to its OAuth provider."""
+    client_module = cast(ClientModule, load_example_module(CLIENT_ROOT, "mcp_simple_auth_client.main"))
+    resource_url = "https://mcp.example.com/prefix/mcp?tenant=mcp"
+    providers: list[OAuthClientProvider] = []
+    sessions = 0
+
+    class FakeCallbackServer:
+        def __init__(self, port: int) -> None:
+            assert port == 3030
+
+        def start(self) -> None:
+            pass
+
+    @asynccontextmanager
+    async def fake_sse_client(
+        *, url: str, auth: OAuthClientProvider, timeout: float
+    ) -> AsyncIterator[
+        tuple[MemoryObjectReceiveStream[SessionMessage | Exception], MemoryObjectSendStream[SessionMessage]]
+    ]:
+        assert url == resource_url
+        assert timeout == 60.0
+        providers.append(auth)
+        read_send, read_receive = anyio.create_memory_object_stream[SessionMessage | Exception](1)
+        write_send, write_receive = anyio.create_memory_object_stream[SessionMessage](1)
+        async with read_send, read_receive, write_send, write_receive:
+            yield read_receive, write_send
+
+    async def record_session(
+        self: SimpleAuthClient,
+        read_stream: MemoryObjectReceiveStream[SessionMessage | Exception],
+        write_stream: MemoryObjectSendStream[SessionMessage],
+    ) -> None:
+        nonlocal sessions
+        sessions += 1
+
+    monkeypatch.setattr(client_module, "CallbackServer", FakeCallbackServer)
+    monkeypatch.setattr(client_module, "sse_client", fake_sse_client)
+    monkeypatch.setattr(client_module.SimpleAuthClient, "_run_session", record_session)
+
+    await client_module.SimpleAuthClient(resource_url, transport_type="sse").connect()
+
+    assert sessions == 1
+    assert [str(provider.context.server_url) for provider in providers] == [resource_url]

tests/examples/simple_auth/test_resource_server_urls.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+from types import ModuleType
+from typing import Literal, Protocol, cast
+
+import pytest
+from click import Command
+from click.testing import CliRunner
+
+from mcp.server.mcpserver.server import MCPServer
+
+SERVER_ROOT = Path(__file__).parents[3] / "examples" / "servers" / "simple-auth"
+
+
+class ServerModule(Protocol):
+    main: Command
+
+
+@pytest.mark.parametrize(
+    ("transport", "endpoint"),
+    [("streamable-http", "/mcp"), ("sse", "/sse")],
+)
+def test_selected_transport_uses_one_resource_path(
+    monkeypatch: pytest.MonkeyPatch,
+    load_example_module: Callable[[Path, str], ModuleType],
+    transport: Literal["sse", "streamable-http"],
+    endpoint: str,
+) -> None:
+    """The example advertises and serves the selected transport path."""
+    server = cast(ServerModule, load_example_module(SERVER_ROOT, "mcp_simple_auth.server"))
+    created: list[MCPServer] = []
+    run_arguments: list[dict[str, object]] = []
+
+    def record_run(
+        self: MCPServer,
+        transport: Literal["stdio", "sse", "streamable-http"] = "stdio",
+        *,
+        host: str = "127.0.0.1",
+        port: int = 8000,
+        **kwargs: object,
+    ) -> None:
+        created.append(self)
+        run_arguments.append({"transport": transport, "host": host, "port": port, **kwargs})
+
+    monkeypatch.setattr(MCPServer, "run", record_run)
+    result = CliRunner().invoke(server.main, ["--port", "8123", "--transport", transport])
+
+    assert result.exit_code == 0, result.output
+    auth = created[0].settings.auth
+    assert auth is not None
+    assert str(auth.resource_server_url) == f"http://localhost:8123{endpoint}"
+    path_argument = "sse_path" if transport == "sse" else "streamable_http_path"
+    assert run_arguments == [{"transport": transport, "host": "localhost", "port": 8123, path_argument: endpoint}]