docs: complete concepts, authorization, and low-level server pages

Author: felixweinberger

docs/authorization.md

@@ -1,5 +1,363 @@
 # Authorization
 
-!!! warning "Under Construction"
+This page covers how the MCP Python SDK implements OAuth 2.1 authorization for protecting MCP servers and authenticating clients.
 
-    This page is currently being written. Check back soon for complete documentation.
+## OAuth 2.1 Overview
+
+MCP uses [OAuth 2.1](https://modelcontextprotocol.io/specification/2025-06-18/basic/authorization) to authorize access to protected servers. The architecture follows a separation between Authorization Servers and Resource Servers:
+
+- **Authorization Server (AS)**: Handles OAuth flows, user authentication, and token issuance.
+- **Resource Server (RS)**: Your MCP server, which validates tokens issued by the AS and serves protected resources.
+- **Client**: Discovers the AS through [RFC 9728](https://datatracker.ietf.org/doc/html/rfc9728) (Protected Resource Metadata), obtains tokens, and uses them with the MCP server.
+
+The typical flow is:
+
+1. The client connects to the MCP server and receives a `401 Unauthorized` response.
+2. The client discovers the Authorization Server via Protected Resource Metadata.
+3. The client performs an OAuth flow (authorization code with PKCE, or client credentials) to obtain an access token.
+4. The client includes the access token in subsequent requests to the MCP server.
+5. The MCP server validates the token and serves the request.
+
+## Server-Side Authorization
+
+### TokenVerifier
+
+The simplest way to add authentication to a FastMCP server is by providing a `TokenVerifier` implementation. The server acts as a Resource Server that validates bearer tokens issued by a separate Authorization Server:
+
+```python
+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 MyTokenVerifier(TokenVerifier):
+    """Verify tokens issued by your Authorization Server."""
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        # Validate the token (e.g., verify JWT signature, check expiry,
+        # call an introspection endpoint, etc.)
+        # Return an AccessToken if valid, or None to reject.
+        if is_valid(token):
+            return AccessToken(
+                token=token,
+                client_id="client-123",
+                scopes=["user"],
+            )
+        return None
+
+
+mcp = FastMCP(
+    "Protected Server",
+    token_verifier=MyTokenVerifier(),
+    auth=AuthSettings(
+        issuer_url=AnyHttpUrl("https://auth.example.com"),
+        resource_server_url=AnyHttpUrl("http://localhost:8000"),
+        required_scopes=["user"],
+    ),
+)
+
+
+@mcp.tool()
+async def get_secret_data() -> str:
+    """This tool requires authentication."""
+    return "secret data"
+
+
+if __name__ == "__main__":
+    mcp.run(transport="streamable-http")
+```
+
+The `TokenVerifier` protocol requires a single method:
+
+- `verify_token(token: str) -> AccessToken | None`: Receives the raw bearer token string. Return an `AccessToken` with the token's metadata if valid, or `None` to reject the request.
+
+### AuthSettings
+
+`AuthSettings` configures the server's Protected Resource Metadata (RFC 9728), which tells clients how to discover the Authorization Server:
+
+```python
+from mcp.server.auth.settings import AuthSettings, ClientRegistrationOptions
+
+auth = AuthSettings(
+    # The OAuth Authorization Server that issues tokens for this server
+    issuer_url=AnyHttpUrl("https://auth.example.com"),
+    # This server's URL (used as the resource identifier)
+    resource_server_url=AnyHttpUrl("https://api.example.com"),
+    # Scopes required for accessing this server
+    required_scopes=["read", "write"],
+    # Optional: documentation URL
+    service_documentation_url=AnyHttpUrl("https://docs.example.com"),
+)
+```
+
+Key settings:
+
+| Setting | Description |
+|---------|-------------|
+| `issuer_url` | The Authorization Server URL that issues tokens |
+| `resource_server_url` | This MCP server's URL, used as the resource identifier |
+| `required_scopes` | Scopes that must be present in the access token |
+| `service_documentation_url` | Optional URL to documentation for the API |
+| `client_registration_options` | Optional settings for dynamic client registration |
+| `revocation_options` | Optional settings for token revocation |
+
+### OAuthAuthorizationServerProvider
+
+For servers that need to act as both the Authorization Server and the Resource Server, implement the `OAuthAuthorizationServerProvider` protocol. This is more complex than the `TokenVerifier` approach and is typically only needed when you want a self-contained auth solution without an external AS:
+
+```python
+from mcp.server.auth.provider import (
+    AccessToken,
+    AuthorizationCode,
+    AuthorizationParams,
+    OAuthAuthorizationServerProvider,
+    RefreshToken,
+)
+from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
+
+
+class MyAuthProvider(OAuthAuthorizationServerProvider):
+    async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
+        """Look up a registered client by ID."""
+        ...
+
+    async def register_client(self, client_info: OAuthClientInformationFull) -> None:
+        """Register a new OAuth client (dynamic client registration)."""
+        ...
+
+    async def authorize(
+        self, client: OAuthClientInformationFull, params: AuthorizationParams
+    ) -> str:
+        """Handle authorization and return a redirect URL."""
+        ...
+
+    async def load_authorization_code(
+        self, client: OAuthClientInformationFull, authorization_code: str
+    ) -> AuthorizationCode | None:
+        """Load a previously issued authorization code."""
+        ...
+
+    async def exchange_authorization_code(
+        self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
+    ) -> OAuthToken:
+        """Exchange an authorization code for tokens."""
+        ...
+
+    async def load_access_token(self, token: str) -> AccessToken | None:
+        """Validate and load an access token."""
+        ...
+
+    async def load_refresh_token(
+        self, client: OAuthClientInformationFull, refresh_token: str
+    ) -> RefreshToken | None:
+        """Load a refresh token."""
+        ...
+
+    async def exchange_refresh_token(
+        self, client: OAuthClientInformationFull, refresh_token: RefreshToken
+    ) -> OAuthToken:
+        """Exchange a refresh token for new tokens."""
+        ...
+
+    async def revoke_token(
+        self, client: OAuthClientInformationFull, token: str, token_type_hint: str | None = None
+    ) -> None:
+        """Revoke an access or refresh token."""
+        ...
+```
+
+## Client-Side Authorization
+
+### OAuthClientProvider
+
+`OAuthClientProvider` is an `httpx.Auth` subclass that handles the full OAuth flow for MCP clients. It manages Protected Resource Metadata discovery, client registration, authorization code exchange with PKCE, token storage, and automatic token refresh:
+
+```python
+import asyncio
+from urllib.parse import parse_qs, urlparse
+
+import httpx
+from pydantic import AnyUrl
+
+from mcp import ClientSession
+from mcp.client.auth import OAuthClientProvider, TokenStorage
+from mcp.client.streamable_http import streamable_http_client
+from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken
+
+
+class InMemoryTokenStorage(TokenStorage):
+    """Simple in-memory token storage."""
+
+    def __init__(self):
+        self.tokens: OAuthToken | None = None
+        self.client_info: OAuthClientInformationFull | None = None
+
+    async def get_tokens(self) -> OAuthToken | None:
+        return self.tokens
+
+    async def set_tokens(self, tokens: OAuthToken) -> None:
+        self.tokens = tokens
+
+    async def get_client_info(self) -> OAuthClientInformationFull | None:
+        return self.client_info
+
+    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
+        self.client_info = client_info
+
+
+async def handle_redirect(auth_url: str) -> None:
+    """Handle the authorization redirect (e.g., open browser)."""
+    print(f"Visit this URL to authorize: {auth_url}")
+
+
+async def handle_callback() -> tuple[str, str | None]:
+    """Handle the callback after authorization."""
+    callback_url = input("Paste the callback URL: ")
+    params = parse_qs(urlparse(callback_url).query)
+    return params["code"][0], params.get("state", [None])[0]
+
+
+async def main():
+    oauth_auth = OAuthClientProvider(
+        server_url="http://localhost:8000",
+        client_metadata=OAuthClientMetadata(
+            client_name="My 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 httpx.AsyncClient(auth=oauth_auth, follow_redirects=True) as http_client:
+        async with streamable_http_client(
+            "http://localhost:8000/mcp", http_client=http_client
+        ) as (read, write, _):
+            async with ClientSession(read, write) as session:
+                await session.initialize()
+                tools = await session.list_tools()
+                print(f"Tools: {[t.name for t in tools.tools]}")
+
+
+asyncio.run(main())
+```
+
+The `OAuthClientProvider` constructor takes these parameters:
+
+| Parameter | Description |
+|-----------|-------------|
+| `server_url` | The MCP server URL |
+| `client_metadata` | OAuth client metadata (name, redirect URIs, grant types, scopes) |
+| `storage` | A `TokenStorage` implementation for persisting tokens and client info |
+| `redirect_handler` | Async callback invoked with the authorization URL (open a browser) |
+| `callback_handler` | Async callback that returns the authorization code from the callback URL |
+| `timeout` | Timeout for the OAuth flow in seconds (default: 300) |
+
+### TokenStorage
+
+The `TokenStorage` protocol defines how tokens and client registration info are persisted. Implement this to store tokens in a database, file system, or any other backend:
+
+```python
+from mcp.client.auth import TokenStorage
+from mcp.shared.auth import OAuthClientInformationFull, OAuthToken
+
+
+class FileTokenStorage(TokenStorage):
+    async def get_tokens(self) -> OAuthToken | None:
+        """Load tokens from disk."""
+        ...
+
+    async def set_tokens(self, tokens: OAuthToken) -> None:
+        """Save tokens to disk."""
+        ...
+
+    async def get_client_info(self) -> OAuthClientInformationFull | None:
+        """Load client registration info from disk."""
+        ...
+
+    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
+        """Save client registration info to disk."""
+        ...
+```
+
+The storage is called during the OAuth flow to persist and retrieve tokens and client information, enabling token reuse across sessions.
+
+## Client Credentials Extension
+
+For service-to-service authentication where no user interaction is needed, use `ClientCredentialsOAuthProvider`. This implements the OAuth 2.1 client credentials grant:
+
+```python
+from mcp.client.auth import TokenStorage
+from mcp.client.auth.extensions.client_credentials import ClientCredentialsOAuthProvider
+
+
+provider = ClientCredentialsOAuthProvider(
+    server_url="https://api.example.com",
+    storage=my_token_storage,
+    client_id="my-service-client-id",
+    client_secret="my-service-client-secret",
+    scopes="read write",
+)
+```
+
+This provider skips dynamic client registration and the authorization code flow entirely. It directly exchanges the client credentials for an access token at the token endpoint.
+
+### PrivateKeyJWTOAuthProvider
+
+For client credentials authentication using JWT assertions (RFC 7523), use `PrivateKeyJWTOAuthProvider`. This is common in workload identity federation scenarios:
+
+```python
+from mcp.client.auth.extensions.client_credentials import (
+    PrivateKeyJWTOAuthProvider,
+    SignedJWTParameters,
+)
+
+# Option 1: SDK-signed JWT (for testing or simple setups)
+jwt_params = SignedJWTParameters(
+    issuer="my-client-id",
+    subject="my-client-id",
+    signing_key=private_key_pem,
+)
+
+provider = PrivateKeyJWTOAuthProvider(
+    server_url="https://api.example.com",
+    storage=my_token_storage,
+    client_id="my-client-id",
+    assertion_provider=jwt_params.create_assertion_provider(),
+)
+```
+
+```python
+# Option 2: Workload identity federation (production)
+async def get_workload_identity_token(audience: str) -> str:
+    # Fetch JWT from your identity provider (GCP, AWS IAM, Azure AD)
+    return await fetch_token_from_identity_provider(audience=audience)
+
+provider = PrivateKeyJWTOAuthProvider(
+    server_url="https://api.example.com",
+    storage=my_token_storage,
+    client_id="my-client-id",
+    assertion_provider=get_workload_identity_token,
+)
+```
+
+Both `ClientCredentialsOAuthProvider` and `PrivateKeyJWTOAuthProvider` are used the same way as `OAuthClientProvider` -- pass them as the `auth` parameter to an `httpx.AsyncClient`:
+
+```python
+import httpx
+from mcp import ClientSession
+from mcp.client.streamable_http import streamable_http_client
+
+async with httpx.AsyncClient(auth=provider, follow_redirects=True) as http_client:
+    async with streamable_http_client(
+        "https://api.example.com/mcp", http_client=http_client
+    ) as (read, write, _):
+        async with ClientSession(read, write) as session:
+            await session.initialize()
+            # Use the session normally
+```