OAuth クライアント

一部の MCP サーバーは保護されています。トークンなしでリクエストを送ると、401 Unauthorized が返ってきます。

そのトークンを手に入れる手段が OAuthClientProvider です。これは MCP のオブジェクトではまったくありません。httpx2.Auth、つまり「すべてのリクエストに何かを施す」ための httpx2 標準のフックです。これを httpx2.AsyncClient に取り付け、そのクライアントを Streamable HTTP トランスポートに渡せば、あとは気にする必要がありません。

このページはクライアント側の話です。自分のサーバーにトークンを要求させる方法は 認可 で扱います。

プロバイダー

from urllib.parse import parse_qs, urlparse

import httpx2
from pydantic import AnyUrl

from mcp import Client
from mcp.client.auth import AuthorizationCodeResult, OAuthClientProvider
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthClientMetadata, OAuthToken


class InMemoryTokenStorage:
    def __init__(self) -> None:
        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 open_browser(authorization_url: str) -> None:
    print(f"Visit: {authorization_url}")


async def wait_for_callback() -> AuthorizationCodeResult:
    redirect_url = input("Paste the URL you were redirected to: ")
    params = parse_qs(urlparse(redirect_url).query)
    return AuthorizationCodeResult(
        code=params["code"][0],
        state=params["state"][0],
        iss=params["iss"][0] if "iss" in params else None,
    )


oauth = OAuthClientProvider(
    server_url="http://localhost:8001/mcp",
    client_metadata=OAuthClientMetadata(
        client_name="Bookshop Agent",
        redirect_uris=[AnyUrl("http://localhost:3030/callback")],
        scope="user",
    ),
    storage=InMemoryTokenStorage(),
    redirect_handler=open_browser,
    callback_handler=wait_for_callback,
)


async def main() -> None:
    async with httpx2.AsyncClient(auth=oauth, follow_redirects=True) as http_client:
        transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client)
        async with Client(transport) as client:
            result = await client.list_tools()
            print([tool.name for tool in result.tools])

渡すものは 4 つです。

• server_url：接続先の MCP エンドポイント。プロバイダーはそれ以外のすべてをここから発見します。
• client_metadata：認可サーバーの「アプリケーションを登録する」フォームに入力するような内容。
• storage：実行と実行のあいだにトークンを保管しておく場所。
• redirect_handler と callback_handler：人間が関わる 2 つの場面。

ファイル内のほかの箇所には OAuth は一切登場しません。main() がトークンを目にすることはありません。

クライアントメタデータ

OAuthClientMetadata は、本物の RFC 7591 登録ドキュメントを Pydantic モデルにしたものです。

設定するフィールドは 3 つです。残りはデフォルト値が埋めてくれます。grant_types は最初から ["authorization_code", "refresh_token"]、response_types は最初から ["code"] で、これはまさにこのプロバイダーが実行するフローです。

redirect_uris
  Field required [type=missing, input_value={'client_name': 'Bookshop Agent'}, input_type=dict]

トークンストレージ

TokenStorage は 4 つの非同期メソッドを持つ Protocol です。何かを継承する必要はありません。メソッドを書けば、どんなクラスでもトークンストアになります。

• get_tokens / set_tokens は OAuthToken（アクセストークン、リフレッシュトークン、有効期限、スコープ）を保持します。
• get_client_info / set_client_info は、プロバイダーが登録したときに認可サーバーが発行した OAuthClientInformationFull（client_id を含む）を保持します。

上のインメモリ版はちゃんと動きます。ただしプロセスが終了するとすべてを忘れるので、次の実行では一連の手順を最初からやり直すことになります。ファイルやプラットフォームのキーリングに永続化すれば、次の実行は何も聞かれずに済みます。

2 つのハンドラー

認可コードフローで人間が必要になるのはちょうど一度だけです。誰かがサインインして「許可」をクリックしなければなりません。

• redirect_handler は、完全に組み立て済みの認可 URL を引数に await されます。client_id、redirect_uri、state、PKCE チャレンジはすでにその中に入っています。やるべきことはブラウザーをそこへ向かわせることだけです。デスクトップアプリなら webbrowser.open を呼び、このファイルでは表示するだけです。
• 次に callback_handler が await されます。ユーザーが redirect_uri に戻ってくるまで待ち、そのリダイレクトのクエリパラメーターを AuthorizationCodeResult として返します。

実際のクライアントは、input() を呼ぶ代わりにリダイレクト URI 上で小さなローカル HTTP サーバーを動かします。形はまったく同じです。リダイレクトを受け取り、code、state、iss を返します。

Client へ

main() を見てください。プロバイダーは httpx2 クライアントに載り、httpx2 クライアントは streamable_http_client(url, http_client=...) に入り、そのトランスポートが Client に入ります。

streamable_http_client には auth= キーワードがありません。HTTP レベルのもの（認証、ヘッダー、タイムアウト、プロキシ）はすべて、持ち込む httpx2.AsyncClient に設定します。このレイヤー構成については クライアントのトランスポート を参照してください。

プロバイダーがやってくれること

Client が初めてリクエストを送ると、サーバーは 401 を返します。そこからプロバイダーが引き継ぎます。

1. 発見。 WWW-Authenticate ヘッダーを読み、サーバーの Protected Resource Metadata を /.well-known/oauth-protected-resource から取得します。そこからこのリソースを保護している認可サーバーを知り、「その」サーバーのメタデータを取得します。
2. 登録。 ストレージに何もなければ、OAuthClientMetadata を使って動的に登録し、結果を保存します。
3. 認可。 PKCE のペアと state を生成し、認可 URL を組み立て、redirect_handler を await します。続いて、コードを受け取るために callback_handler を await します。
4. 交換。 コードを OAuthToken と引き換えて保存し、元のリクエストを Authorization: Bearer ... 付きで再送します。

それ以降は静かになります。トークンはストレージから取り出され、期限切れのアクセストークンはリフレッシュトークンで更新されます。そのどれもうまくいかないときだけ、フローをもう一度実行します。

これらを自分で書く必要はまったくありませんでした。残るキーワード引数は 2 つ（client_metadata_url と validate_resource_url）で、このファイルではどちらも不要です。知っておく価値があるのは client_metadata_url のほうで、下に専用のセクションがあります。

試してみる

このドキュメントの例のほとんどは、インメモリの Client(server) で確認できます。これは違います。このフローの要点は HTTP の 401 であり、インメモリのクライアントとサーバーのあいだには HTTP がありません。

リポジトリには実際に動くバージョンが同梱されています。examples/servers/simple-auth/ はスタンドアロンの認可サーバーと保護された MCP サーバーを動かし、examples/clients/simple-auth-client/ はこのページのクライアントを小さな CLI に育てたものです。その README に 2 つのコマンドが載っています。サーバーを起動し、それに対してクライアントを実行すれば、4 つのステップが進んでいくのを見られます。

Client ID Metadata Documents

仕様の 2026-07-28 改訂では、動的クライアント登録が非推奨になり、代わりに Client ID Metadata Documents（CIMD）が推奨されます。出会う認可サーバーごとに新しい登録を POST する代わりに、クライアントは自分自身についての JSON ドキュメントを 1 つ、安定した HTTPS URL で公開します。そしてその URL がそのまま client_id になります。ドキュメントを取得するのは認可サーバーで、プロバイダーはそれに一切触れません。

SDK はすでにこれに対応しています。プロバイダーを構築するときに URL を client_metadata_url= として渡してください。認可サーバーのメタデータが client_id_metadata_document_supported: true を公表していれば、プロバイダーは /register リクエストを完全に省きます。URL が client_id としてフローに入り、client_secret はありません。サーバーがそれを公表していない場合（まだ大半がそうです）、または URL を渡さなかった場合、プロバイダーは何も言わずに動的登録にフォールバックし、上の説明どおりにすべてが動きます。保存済みの client_info は、依然としてそのどちらよりも優先されます。

URL は HTTPS で、ルート以外のパスを持っている必要があります。それ以外は、ネットワーク通信が起こる前の構築時点で ValueError になります。同梱の examples/clients/simple-auth-client/ は、これを MCP_CLIENT_METADATA_URL 環境変数として受け取ります。

マシン間通信

夜間ジョブ、CI のステップ、別のサービス。ブラウザーはなく、「許可」をクリックする人もいません。これが クライアントクレデンシャル グラントです。client_id と client_secret はすでに手元にあり、トークンエンドポイントがフローのすべてです。

ClientCredentialsOAuthProvider は同じ httpx2.Auth で、人間がいないだけです。

import httpx2

from mcp import Client
from mcp.client.auth.extensions.client_credentials import ClientCredentialsOAuthProvider
from mcp.client.streamable_http import streamable_http_client
from mcp.shared.auth import OAuthClientInformationFull, OAuthToken


class InMemoryTokenStorage:
    def __init__(self) -> None:
        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


oauth = ClientCredentialsOAuthProvider(
    server_url="http://localhost:8001/mcp",
    storage=InMemoryTokenStorage(),
    client_id="reporting-agent",
    client_secret="...",
    scope="user",
)


async def main() -> None:
    async with httpx2.AsyncClient(auth=oauth, follow_redirects=True) as http_client:
        transport = streamable_http_client("http://localhost:8001/mcp", http_client=http_client)
        async with Client(transport) as client:
            result = await client.list_tools()
            print([tool.name for tool in result.tools])

変わった点は次のとおりです。

• OAuthClientMetadata もハンドラーもありません。client_id と client_secret を渡すと、プロバイダーはそれらを中心に最小限の client_credentials 登録を組み立て、動的登録を完全に省きます。
• scope はスペース区切りの文字列で、OAuth の通信上の形式です。
• その先はすべて同じです。同じ TokenStorage、同じ httpx2.AsyncClient(auth=...)、同じ streamable_http_client です。

デフォルトでは、シークレットはトークンリクエストの HTTP Basic 認証として送られます（client_secret_basic）。代わりにフォームボディに入れるには、token_endpoint_auth_method="client_secret_post" を渡してください。認可サーバーによっては、2 つのうち片方しか受け付けません。

人間がいない状況はもう 1 つあります。クライアントが企業に属していて、どの MCP サーバーに到達してよいかをユーザーではなくその企業のアイデンティティプロバイダーが決める場合です。これは独自の信頼モデルを持つ別のグラントで、専用のページ アイデンティティアサーション があります。

失敗したとき

OAuth フローがうまくいかないと、プロバイダーは mcp.client.auth の OAuthFlowError を送出します。これには 2 つのサブクラスがあります。OAuthRegistrationError は、登録の結果として使えるクライアントが得られなかったことを意味します。認可サーバーが登録を拒否したか、登録はされたもののこのフローでは使えないクレデンシャル（たとえば実装していない認証方式）だった場合です。OAuthTokenError は、トークンを取得できなかったことを意味します。トークンエンドポイントに拒否されたか、保存済みのクライアントレコードにこのクライアントが適用できない認証方式が含まれていた場合で、後者は送信されずにトークンリクエストの組み立て中に報告されます。except OAuthFlowError: 1 つで、発見、登録、認可、交換のすべてをカバーできます。

すべてがフローエラーというわけではありません。ネットワークが失敗することもあります。それらは通常の httpx2 の例外で、手を加えられずにそのまま通り抜けます。

まとめ

• OAuthClientProvider は httpx2.Auth です。httpx2.AsyncClient に載せ、それを streamable_http_client(url, http_client=...) に渡せば、Client は OAuth が行われたことを知ることすらありません。
• 渡すものは 4 つです。サーバーの URL、OAuthClientMetadata、TokenStorage、そしてリダイレクト／コールバックのハンドラーのペアです。
• TokenStorage は Protocol です。非同期メソッドが 4 つで、基底クラスはありません。トークンだけでなく client_info も永続化してください。
• 発見、登録（動的、または Client ID Metadata Document 経由）、PKCE、state と iss のチェック、トークンの更新はプロバイダーの仕事であり、呼び出し側の仕事ではありません。
• ClientCredentialsOAuthProvider は人間がいない版です。client_id と client_secret だけで、ハンドラーもブラウザーも要りません。
• OAuth の失敗はすべて OAuthFlowError です。OAuthRegistrationError と OAuthTokenError がそのサブクラスです。

このハンドシェイクのもう半分、つまり「サーバー」にトークンを要求させる方法は 認可 で扱います。