diff --git a/README.md b/README.md index 487d48bee..4b4f66057 100644 --- a/README.md +++ b/README.md @@ -1067,6 +1067,67 @@ For a complete example with separate Authorization Server and Resource Server im See [TokenVerifier](src/mcp/server/auth/provider.py) for more details on implementing token validation. +For a minimal bearer-token server, validate the `Authorization` header in a +`TokenVerifier`. The SDK applies the verifier to both SSE and Streamable HTTP +requests, and tools can read the accepted token from `get_access_token()`: + + +```python +"""Run from the repository root: +uv run examples/snippets/servers/bearer_auth_server.py +""" + +from pydantic import AnyHttpUrl + +from mcp.server.auth.middleware.auth_context import get_access_token +from mcp.server.auth.provider import AccessToken, TokenVerifier +from mcp.server.auth.settings import AuthSettings +from mcp.server.mcpserver import MCPServer + + +class StaticTokenVerifier(TokenVerifier): + """Accept a single bearer token for demonstration purposes.""" + + async def verify_token(self, token: str) -> AccessToken | None: + if token != "secret-token": + return None + + return AccessToken( + token=token, + client_id="demo-client", + scopes=["user"], + ) + + +mcp = MCPServer( + "Bearer auth demo", + token_verifier=StaticTokenVerifier(), + auth=AuthSettings( + issuer_url=AnyHttpUrl("https://auth.example.com"), + resource_server_url=AnyHttpUrl("http://localhost:8000"), + required_scopes=["user"], + ), +) + + +@mcp.tool() +async def whoami() -> str: + """Return the authenticated client id.""" + access_token = get_access_token() + if access_token is None: + # The auth middleware rejects unauthenticated requests before tools run. + raise ValueError("No access token found") + + return f"Authenticated as {access_token.client_id}" + + +if __name__ == "__main__": + mcp.run(transport="streamable-http") +``` + +_Full example: [examples/snippets/servers/bearer_auth_server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/bearer_auth_server.py)_ + + ### FastMCP Properties The FastMCP server instance accessible via `ctx.fastmcp` provides access to server configuration and metadata: diff --git a/README.v2.md b/README.v2.md index d0851c04e..62459bd1d 100644 --- a/README.v2.md +++ b/README.v2.md @@ -1058,6 +1058,67 @@ For a complete example with separate Authorization Server and Resource Server im See [TokenVerifier](src/mcp/server/auth/provider.py) for more details on implementing token validation. +For a minimal bearer-token server, validate the `Authorization` header in a +`TokenVerifier`. The SDK applies the verifier to both SSE and Streamable HTTP +requests, and tools can read the accepted token from `get_access_token()`: + + +```python +"""Run from the repository root: +uv run examples/snippets/servers/bearer_auth_server.py +""" + +from pydantic import AnyHttpUrl + +from mcp.server.auth.middleware.auth_context import get_access_token +from mcp.server.auth.provider import AccessToken, TokenVerifier +from mcp.server.auth.settings import AuthSettings +from mcp.server.mcpserver import MCPServer + + +class StaticTokenVerifier(TokenVerifier): + """Accept a single bearer token for demonstration purposes.""" + + async def verify_token(self, token: str) -> AccessToken | None: + if token != "secret-token": + return None + + return AccessToken( + token=token, + client_id="demo-client", + scopes=["user"], + ) + + +mcp = MCPServer( + "Bearer auth demo", + token_verifier=StaticTokenVerifier(), + auth=AuthSettings( + issuer_url=AnyHttpUrl("https://auth.example.com"), + resource_server_url=AnyHttpUrl("http://localhost:8000"), + required_scopes=["user"], + ), +) + + +@mcp.tool() +async def whoami() -> str: + """Return the authenticated client id.""" + access_token = get_access_token() + if access_token is None: + # The auth middleware rejects unauthenticated requests before tools run. + raise ValueError("No access token found") + + return f"Authenticated as {access_token.client_id}" + + +if __name__ == "__main__": + mcp.run(transport="streamable-http") +``` + +_Full example: [examples/snippets/servers/bearer_auth_server.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/bearer_auth_server.py)_ + + ### MCPServer Properties The MCPServer server instance accessible via `ctx.mcp_server` provides access to server configuration and metadata: diff --git a/examples/snippets/servers/bearer_auth_server.py b/examples/snippets/servers/bearer_auth_server.py new file mode 100644 index 000000000..b8bb03e59 --- /dev/null +++ b/examples/snippets/servers/bearer_auth_server.py @@ -0,0 +1,50 @@ +"""Run from the repository root: +uv run examples/snippets/servers/bearer_auth_server.py +""" + +from pydantic import AnyHttpUrl + +from mcp.server.auth.middleware.auth_context import get_access_token +from mcp.server.auth.provider import AccessToken, TokenVerifier +from mcp.server.auth.settings import AuthSettings +from mcp.server.mcpserver import MCPServer + + +class StaticTokenVerifier(TokenVerifier): + """Accept a single bearer token for demonstration purposes.""" + + async def verify_token(self, token: str) -> AccessToken | None: + if token != "secret-token": + return None + + return AccessToken( + token=token, + client_id="demo-client", + scopes=["user"], + ) + + +mcp = MCPServer( + "Bearer auth demo", + token_verifier=StaticTokenVerifier(), + auth=AuthSettings( + issuer_url=AnyHttpUrl("https://auth.example.com"), + resource_server_url=AnyHttpUrl("http://localhost:8000"), + required_scopes=["user"], + ), +) + + +@mcp.tool() +async def whoami() -> str: + """Return the authenticated client id.""" + access_token = get_access_token() + if access_token is None: + # The auth middleware rejects unauthenticated requests before tools run. + raise ValueError("No access token found") + + return f"Authenticated as {access_token.client_id}" + + +if __name__ == "__main__": + mcp.run(transport="streamable-http")