Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
61 changes: 61 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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()`:

<!-- snippet-source examples/snippets/servers/bearer_auth_server.py -->
```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)_
<!-- /snippet-source -->

### FastMCP Properties

The FastMCP server instance accessible via `ctx.fastmcp` provides access to server configuration and metadata:
Expand Down
61 changes: 61 additions & 0 deletions README.v2.md
Original file line number Diff line number Diff line change
Expand Up @@ -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()`:

<!-- snippet-source examples/snippets/servers/bearer_auth_server.py -->
```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)_
<!-- /snippet-source -->

### MCPServer Properties

The MCPServer server instance accessible via `ctx.mcp_server` provides access to server configuration and metadata:
Expand Down
50 changes: 50 additions & 0 deletions examples/snippets/servers/bearer_auth_server.py
Original file line number Diff line number Diff line change
@@ -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")
Loading