updated auth examples

Author: ihrpr

README.md

@@ -744,30 +744,59 @@ Authentication can be used by servers that want to expose tools accessing protec
 
 MCP servers can use authentication by providing an implementation of the `TokenVerifier` protocol:
 
+<!-- snippet-source examples/snippets/servers/oauth_server.py -->
 ```python
-from mcp import FastMCP
-from mcp.server.auth.provider import TokenVerifier, TokenInfo
+"""
+Run from the repository root:
+    uv run examples/snippets/servers/oauth_server.py
+"""
+
+from pydantic import AnyHttpUrl
+
+from mcp.server.auth.provider import AccessToken, TokenVerifier
 from mcp.server.auth.settings import AuthSettings
+from mcp.server.fastmcp import FastMCP
+
 
+class SimpleTokenVerifier(TokenVerifier):
+    """Simple token verifier for demonstration."""
 
-class MyTokenVerifier(TokenVerifier):
-    # Implement token validation logic (typically via token introspection)
-    async def verify_token(self, token: str) -> TokenInfo:
-        # Verify with your authorization server
-        ...
+    async def verify_token(self, token: str) -> AccessToken | None:
+        pass  # This is where you would implement actual token validation
 
 
+# Create FastMCP instance as a Resource Server
 mcp = FastMCP(
-    "My App",
-    token_verifier=MyTokenVerifier(),
+    "Weather Service",
+    # Token verifier for authentication
+    token_verifier=SimpleTokenVerifier(),
+    # Auth settings for RFC 9728 Protected Resource Metadata
     auth=AuthSettings(
-        issuer_url="https://auth.example.com",
-        resource_server_url="http://localhost:3001",
-        required_scopes=["mcp:read", "mcp:write"],
+        issuer_url=AnyHttpUrl("https://auth.example.com"),  # Authorization Server URL
+        resource_server_url=AnyHttpUrl("http://localhost:3001"),  # This server's URL
+        required_scopes=["user"],
     ),
 )
+
+
+@mcp.tool()
+async def get_weather(city: str = "London") -> dict[str, str]:
+    """Get weather data for a city"""
+    return {
+        "city": city,
+        "temperature": "22°C",
+        "condition": "Partly cloudy",
+        "humidity": "65%",
+    }
+
+
+if __name__ == "__main__":
+    mcp.run(transport="streamable-http")
 ```
 
+_Full example: [examples/snippets/servers/oauth_server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/oauth_server.py)_
+<!-- /snippet-source -->
+
 For a complete example with separate Authorization Server and Resource Server implementations, see [`examples/servers/simple-auth/`](examples/servers/simple-auth/).
 
 **Architecture:**
@@ -1556,53 +1585,100 @@ This ensures your client UI shows the most user-friendly names that servers prov
 
 The SDK includes [authorization support](https://modelcontextprotocol.io/specification/2025-03-26/basic/authorization) for connecting to protected MCP servers:
 
+<!-- snippet-source examples/snippets/clients/oauth_client.py -->
 ```python
+"""
+Before running, specify running MCP RS server URL.
+To spin up RS server locally, see
+    examples/servers/simple-auth/README.md
+
+cd to the `examples/snippets` directory and run:
+    uv run oauth-client
+"""
+
+import asyncio
+from urllib.parse import parse_qs, urlparse
+
+from pydantic import AnyUrl
+
+from mcp import ClientSession
 from mcp.client.auth import OAuthClientProvider, TokenStorage
-from mcp.client.session import ClientSession
 from mcp.client.streamable_http import streamablehttp_client
 from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
 
 
-class CustomTokenStorage(TokenStorage):
-    """Simple in-memory token storage implementation."""
+class InMemoryTokenStorage(TokenStorage):
+    """Demo In-memory token storage implementation."""
+
+    def __init__(self):
+        self.tokens: OAuthToken | None = None
+        self.client_info: OAuthClientInformationFull | None = None
 
     async def get_tokens(self) -> OAuthToken | None:
-        pass
+        """Get stored tokens."""
+        return self.tokens
 
     async def set_tokens(self, tokens: OAuthToken) -> None:
-        pass
+        """Store tokens."""
+        self.tokens = tokens
 
     async def get_client_info(self) -> OAuthClientInformationFull | None:
-        pass
+        """Get stored client information."""
+        return self.client_info
 
     async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
-        pass
+        """Store client information."""
+        self.client_info = client_info
+
+
+async def handle_redirect(auth_url: str) -> None:
+    print(f"Visit: {auth_url}")
+
+
+async def handle_callback() -> tuple[str, str | None]:
+    callback_url = input("Paste callback URL: ")
+    params = parse_qs(urlparse(callback_url).query)
+    return params["code"][0], params.get("state", [None])[0]
 
 
 async def main():
-    # Set up OAuth authentication
+    """Run the OAuth client example."""
     oauth_auth = OAuthClientProvider(
-        server_url="https://api.example.com",
+        server_url="http://localhost:8001",
         client_metadata=OAuthClientMetadata(
-            client_name="My Client",
-            redirect_uris=["http://localhost:3000/callback"],
+            client_name="Example MCP Client",
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
             grant_types=["authorization_code", "refresh_token"],
             response_types=["code"],
+            scope="user",
         ),
-        storage=CustomTokenStorage(),
-        redirect_handler=lambda url: print(f"Visit: {url}"),
-        callback_handler=lambda: ("auth_code", None),
+        storage=InMemoryTokenStorage(),
+        redirect_handler=handle_redirect,
+        callback_handler=handle_callback,
     )
 
-    # Use with streamable HTTP client
-    async with streamablehttp_client(
-        "https://api.example.com/mcp", auth=oauth_auth
-    ) as (read, write, _):
+    async with streamablehttp_client("http://localhost:8001/mcp", auth=oauth_auth) as (read, write, _):
         async with ClientSession(read, write) as session:
             await session.initialize()
-            # Authenticated session ready
+
+            tools = await session.list_tools()
+            print(f"Available tools: {[tool.name for tool in tools.tools]}")
+
+            resources = await session.list_resources()
+            print(f"Available resources: {[r.uri for r in resources.resources]}")
+
+
+def run():
+    asyncio.run(main())
+
+
+if __name__ == "__main__":
+    run()
 ```
 
+_Full example: [examples/snippets/clients/oauth_client.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/oauth_client.py)_
+<!-- /snippet-source -->
+
 For a complete working example, see [`examples/clients/simple-auth-client/`](examples/clients/simple-auth-client/).
 
 ### MCP Primitives

examples/snippets/clients/oauth_client.py

@@ -0,0 +1,87 @@
+"""
+Before running, specify running MCP RS server URL.
+To spin up RS server locally, see
+    examples/servers/simple-auth/README.md
+
+cd to the `examples/snippets` directory and run:
+    uv run oauth-client
+"""
+
+import asyncio
+from urllib.parse import parse_qs, urlparse
+
+from pydantic import AnyUrl
+
+from mcp import ClientSession
+from mcp.client.auth import OAuthClientProvider, TokenStorage
+from mcp.client.streamable_http import streamablehttp_client
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
+
+
+class InMemoryTokenStorage(TokenStorage):
+    """Demo In-memory token storage implementation."""
+
+    def __init__(self):
+        self.tokens: OAuthToken | None = None
+        self.client_info: OAuthClientInformationFull | None = None
+
+    async def get_tokens(self) -> OAuthToken | None:
+        """Get stored tokens."""
+        return self.tokens
+
+    async def set_tokens(self, tokens: OAuthToken) -> None:
+        """Store tokens."""
+        self.tokens = tokens
+
+    async def get_client_info(self) -> OAuthClientInformationFull | None:
+        """Get stored client information."""
+        return self.client_info
+
+    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
+        """Store client information."""
+        self.client_info = client_info
+
+
+async def handle_redirect(auth_url: str) -> None:
+    print(f"Visit: {auth_url}")
+
+
+async def handle_callback() -> tuple[str, str | None]:
+    callback_url = input("Paste callback URL: ")
+    params = parse_qs(urlparse(callback_url).query)
+    return params["code"][0], params.get("state", [None])[0]
+
+
+async def main():
+    """Run the OAuth client example."""
+    oauth_auth = OAuthClientProvider(
+        server_url="http://localhost:8001",
+        client_metadata=OAuthClientMetadata(
+            client_name="Example MCP Client",
+            redirect_uris=[AnyUrl("http://localhost:3000/callback")],
+            grant_types=["authorization_code", "refresh_token"],
+            response_types=["code"],
+            scope="user",
+        ),
+        storage=InMemoryTokenStorage(),
+        redirect_handler=handle_redirect,
+        callback_handler=handle_callback,
+    )
+
+    async with streamablehttp_client("http://localhost:8001/mcp", auth=oauth_auth) as (read, write, _):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+
+            tools = await session.list_tools()
+            print(f"Available tools: {[tool.name for tool in tools.tools]}")
+
+            resources = await session.list_resources()
+            print(f"Available resources: {[r.uri for r in resources.resources]}")
+
+
+def run():
+    asyncio.run(main())
+
+
+if __name__ == "__main__":
+    run()

examples/snippets/pyproject.toml

@@ -20,3 +20,4 @@ client = "clients.stdio_client:main"
 completion-client = "clients.completion_client:main"
 direct-execution-server = "servers.direct_execution:main"
 display-utilities-client = "clients.display_utilities:main"
+oauth-client = "clients.oauth_client:run"

examples/snippets/servers/oauth_server.py

@@ -0,0 +1,46 @@
+"""
+Run from the repository root:
+    uv run examples/snippets/servers/oauth_server.py
+"""
+
+from pydantic import AnyHttpUrl
+
+from mcp.server.auth.provider import AccessToken, TokenVerifier
+from mcp.server.auth.settings import AuthSettings
+from mcp.server.fastmcp import FastMCP
+
+
+class SimpleTokenVerifier(TokenVerifier):
+    """Simple token verifier for demonstration."""
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        pass  # This is where you would implement actual token validation
+
+
+# Create FastMCP instance as a Resource Server
+mcp = FastMCP(
+    "Weather Service",
+    # Token verifier for authentication
+    token_verifier=SimpleTokenVerifier(),
+    # 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
+        required_scopes=["user"],
+    ),
+)
+
+
+@mcp.tool()
+async def get_weather(city: str = "London") -> dict[str, str]:
+    """Get weather data for a city"""
+    return {
+        "city": city,
+        "temperature": "22°C",
+        "condition": "Partly cloudy",
+        "humidity": "65%",
+    }
+
+
+if __name__ == "__main__":
+    mcp.run(transport="streamable-http")