Narrow OAuth resource URL fix

Author: kinocode3

examples/clients/simple-auth-client/tests/test_oauth_resource_url.py

@@ -0,0 +1,57 @@
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+import anyio
+import pytest
+from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
+from mcp.client.auth import OAuthClientProvider
+from mcp.shared.message import SessionMessage
+
+from mcp_simple_auth_client import main as client_module
+from mcp_simple_auth_client.main import SimpleAuthClient
+
+
+@pytest.mark.anyio
+async def test_oauth_client_preserves_the_complete_connection_url(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The example passes the opaque MCP endpoint unchanged to its OAuth provider."""
+    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(SimpleAuthClient, "_run_session", record_session)
+
+    await SimpleAuthClient(resource_url, transport_type="sse").connect()
+
+    assert sessions == 1
+    assert [str(provider.context.server_url) for provider in providers] == [resource_url]

examples/servers/simple-auth/README.md

@@ -38,18 +38,8 @@ uv run mcp-simple-auth-rs --port=8001 --auth-server=http://localhost:9000 --tran
 
 ```
 
-The resource identifier defaults to the selected transport endpoint: `/mcp` for
-Streamable HTTP and `/sse` for SSE. If a proxy or mounted application exposes a
-different public URL, pass the complete endpoint explicitly:
-
-```bash
-uv run mcp-simple-auth-rs --port=8001 --auth-server=http://localhost:9000 \
-  --resource-server-url=https://gateway.example.com/services/time/mcp
-```
-
-Configure the proxy to forward the corresponding public well-known path (for
-this example, `/.well-known/oauth-protected-resource/services/time/mcp`) to the
-resource-server application as well.
+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`:
 

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

@@ -99,11 +99,6 @@ async def get_time() -> dict[str, Any]:
 @click.command()
 @click.option("--port", default=8001, help="Port to listen on")
 @click.option("--auth-server", default="http://localhost:9000", help="Authorization Server URL")
-@click.option(
-    "--resource-server-url",
-    envvar="MCP_RESOURCE_SERVER_URL",
-    help="Complete public MCP endpoint URL (defaults to the selected transport path)",
-)
 @click.option(
     "--transport",
     default="streamable-http",
@@ -118,7 +113,6 @@ async def get_time() -> dict[str, Any]:
 def main(
     port: int,
     auth_server: str,
-    resource_server_url: str | None,
     transport: Literal["sse", "streamable-http"],
     oauth_strict: bool,
 ) -> int:
@@ -140,7 +134,7 @@ def main(
         # Create settings
         host = "localhost"
         transport_path = "/sse" if transport == "sse" else "/mcp"
-        server_url = resource_server_url or f"http://{host}:{port}{transport_path}"
+        server_url = f"http://{host}:{port}{transport_path}"
         settings = ResourceServerSettings(
             host=host,
             port=port,
@@ -151,7 +145,7 @@ def main(
         )
     except ValueError as e:
         logger.error(f"Configuration error: {e}")
-        logger.error("Make sure to provide valid Authorization and Resource Server URLs")
+        logger.error("Make sure to provide a valid Authorization Server URL")
         return 1
 
     try:

examples/servers/simple-auth/tests/test_resource_server_urls.py

@@ -0,0 +1,42 @@
+from typing import Literal
+
+import pytest
+from click.testing import CliRunner
+from mcp_simple_auth import server
+
+from mcp.server.mcpserver.server import MCPServer
+
+
+@pytest.mark.parametrize(
+    ("transport", "endpoint"),
+    [("streamable-http", "/mcp"), ("sse", "/sse")],
+)
+def test_selected_transport_uses_one_resource_path(
+    monkeypatch: pytest.MonkeyPatch,
+    transport: Literal["sse", "streamable-http"],
+    endpoint: str,
+) -> None:
+    """The example advertises and serves the selected transport path."""
+    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}]

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:8000/mcp"),  # This server's MCP endpoint
+        resource_server_url=AnyHttpUrl("http://127.0.0.1:8000/mcp"),  # This server's MCP endpoint
         required_scopes=["user"],
     ),
 )

pyproject.toml

@@ -175,9 +175,10 @@ executionEnvironments = [
     { root = "tests", extraPaths = [
         ".",
         "examples",
-        "examples/clients/simple-auth-client",
-        "examples/servers/simple-auth",
     ], 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/mcpserver/server.py

@@ -1122,20 +1122,21 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no
         required_scopes: list[str] = []
 
         # Set up auth if configured
-        if self.settings.auth:
+        if self.settings.auth:  # pragma: no cover
             required_scopes = self.settings.auth.required_scopes or []
-            assert self._token_verifier is not None
-
-            middleware = [
-                # extract auth info from request (but do not require it)
-                Middleware(
-                    AuthenticationMiddleware,
-                    backend=BearerAuthBackend(self._token_verifier),
-                ),
-                # Add the auth context middleware to store
-                # authenticated user in a contextvar
-                Middleware(AuthContextMiddleware),
-            ]
+
+            # Add auth middleware if token verifier is available
+            if self._token_verifier:
+                middleware = [
+                    # extract auth info from request (but do not require it)
+                    Middleware(
+                        AuthenticationMiddleware,
+                        backend=BearerAuthBackend(self._token_verifier),
+                    ),
+                    # Add the auth context middleware to store
+                    # authenticated user in a contextvar
+                    Middleware(AuthContextMiddleware),
+                ]
 
             # Add auth endpoints if auth server provider is configured
             if self._auth_server_provider:
@@ -1153,7 +1154,7 @@ async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no
                 )
 
         # When auth is configured, require authentication
-        if self.settings.auth:
+        if self._token_verifier:  # pragma: no cover
             # Determine resource metadata URL
             resource_metadata_url = None
             if self.settings.auth and self.settings.auth.resource_server_url:
@@ -1197,7 +1198,7 @@ async def sse_endpoint(request: Request) -> Response:  # pragma: no cover
                 )
             )
         # Add protected resource metadata endpoint if configured as RS
-        if self.settings.auth and self.settings.auth.resource_server_url:
+        if self.settings.auth and self.settings.auth.resource_server_url:  # pragma: no cover
             from mcp.server.auth.routes import create_protected_resource_routes
 
             routes.extend(