Skip to content

authorization.md

Latest commit

 

History

History
150 lines (111 loc) · 4.69 KB

authorization.md

File metadata and controls

150 lines (111 loc) · 4.69 KB

Authorization

MCP HTTP transports authenticate via httpx.Auth. The SDK provides two implementations that plug into the same auth parameter:

  • BearerAuth — a minimal two-method provider for API keys, gateway-managed tokens, service accounts, or any scenario where the token comes from an external pipeline.
  • OAuthClientProvider — full OAuth 2.1 authorization-code flow with PKCE, Protected Resource Metadata discovery (RFC 9728), dynamic client registration, and automatic token refresh.

Both are httpx.Auth subclasses. Pass either to Client(url, auth=...), streamable_http_client(url, auth=...), or directly to httpx.AsyncClient(auth=...).

Bearer tokens

For a static token (API key, pre-provisioned credential):

from mcp.client import Client
from mcp.client.auth import BearerAuth

async with Client("https://api.example.com/mcp", auth=BearerAuth("my-api-key")) as client:
    tools = await client.list_tools()

For a dynamic token (environment variable, cache, external service), pass a callable — sync or async:

import os
from mcp.client.auth import BearerAuth

auth = BearerAuth(lambda: os.environ.get("MCP_TOKEN"))

token() is called before every request, so the callable can return a freshly rotated value each time. Keep it fast — return a cached value and refresh in the background rather than blocking on network calls.

Handling 401

By default, BearerAuth raises UnauthorizedError immediately on 401. To refresh credentials and retry once, pass an on_unauthorized handler:

from mcp.client.auth import BearerAuth, UnauthorizedContext

token_cache = TokenCache()

async def refresh(ctx: UnauthorizedContext) -> None:
    # ctx.response.headers["WWW-Authenticate"] has scope/resource_metadata hints
    await token_cache.invalidate()

auth = BearerAuth(token_cache.get, on_unauthorized=refresh)

After on_unauthorized returns, token() is called again and the request is retried once. If the retry also gets 401, UnauthorizedError is raised. Retry state is scoped per-request — a failed retry on one request does not block retries on subsequent requests.

To abort without retrying (for example, when interactive user action is required), raise from the handler:

async def signal_host(ctx: UnauthorizedContext) -> None:
    ui.show_reauth_prompt()
    raise UnauthorizedError("User action required before retry")

Subclassing

For more complex providers, subclass BearerAuth and override token() and on_unauthorized():

from mcp.client.auth import BearerAuth, UnauthorizedContext

class MyAuth(BearerAuth):
    async def token(self) -> str | None:
        return await self._store.get_access_token()

    async def on_unauthorized(self, context: UnauthorizedContext) -> None:
        await self._store.refresh()

OAuth 2.1

For the full OAuth authorization-code flow with PKCE — including Protected Resource Metadata discovery, authorization server metadata discovery, dynamic client registration, and automatic token refresh — use OAuthClientProvider:

import httpx
from mcp.client.auth import OAuthClientProvider, TokenStorage
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import OAuthClientMetadata

auth = OAuthClientProvider(
    server_url="https://api.example.com",
    client_metadata=OAuthClientMetadata(
        client_name="My MCP Client",
        redirect_uris=["http://localhost:3000/callback"],
        grant_types=["authorization_code", "refresh_token"],
        response_types=["code"],
    ),
    storage=my_token_storage,
    redirect_handler=open_browser,
    callback_handler=wait_for_callback,
)

async with streamable_http_client("https://api.example.com/mcp", auth=auth) as (read, write):
    ...

See examples/snippets/clients/oauth_client.py for a complete working example.

Non-interactive grants

For machine-to-machine authentication without a browser redirect, use the extensions in mcp.client.auth.extensions:

  • ClientCredentialsOAuthProvider — client_credentials grant with client ID and secret
  • PrivateKeyJWTOAuthProvider — client_credentials with private_key_jwt client authentication (RFC 7523)

Custom httpx.Auth

Any httpx.Auth implementation works. To combine authentication with custom HTTP settings (headers, timeouts, proxies), configure an httpx.AsyncClient directly:

import httpx
from mcp.client.streamable_http import streamable_http_client

http_client = httpx.AsyncClient(
    auth=my_auth,
    headers={"X-Custom": "value"},
    timeout=httpx.Timeout(60.0),
)

async with http_client:
    async with streamable_http_client(url, http_client=http_client) as (read, write):
        ...