Chuk MCP Client Oauth

楚克mcp客户端oauth

一个简单、安全的OAuth 2.0客户端库，用于连接到MCP（模型上下文协议）服务器。

非常适合那些希望在MCP应用程序中添加OAuth身份验证而无需与OAuth复杂性搏斗的开发人员。

   

______________________________________________________________________

🎯 这是什么？

这个图书馆使它 非常简单 通过启用OAuth的MCP服务器进行身份验证。无论您是在构建需要连接到MCP服务器的CLI工具、web应用程序还是服务，此库都能为您处理所有OAuth复杂性。

什么是MCP OAuth？

MCP（模型上下文协议）服务器可以使用OAuth 2.0来控制谁可以访问它们。把它想象成登录GitHub或谷歌——但对于AI/LLM服务来说。

作为客户端开发人员，您需要：

1. 🔐 验证 -从服务器获取权限
2. 💾 存储代币 -确保凭据安全
3. 🔄 刷新令牌 -保持会话活跃
4. 🔧 使用令牌 -在API请求中包含它们

这个图书馆为你做这一切。

OAuth 2.1和MCP合规性

此库实现了：

• ✅ OAuth 2.1最佳实践 -授权码+PKCE，无遗留授权
• ✅ MCP授权规范 -受保护资源元数据发现（RFC 9728）
• ✅ 资源指标 -令牌绑定以防止重用（RFC 8707）
• ✅ WWW身份验证回退 -从401/403条回复中发现
• ✅ 安全令牌存储 -操作系统密钥链、加密文件、HashiCorp保险库
• ✅ 自动令牌刷新 -透明地处理过期
• 🔄 设备代码流 -v0.2.0版本提供无头环境

标准合规性：

• OAuth 2.1草案 -现代OAuth最佳实践
• RFC 9728 -受保护资源元数据
• RFC 8707 -资源指标
• RFC 8414 -授权服务器元数据发现
• RFC 7591 -动态客户端注册
• RFC 7636 - PKCE

______________________________________________________________________

🚀 快速入门（5分钟）

安装

使用 uv （推荐）：

uv add chuk-mcp-client-oauth

或者使用pip：

pip install chuk-mcp-client-oauth

30秒最小示例

import asyncio
from chuk_mcp_client_oauth import OAuthHandler

async def main():
    handler = OAuthHandler()  # Auto keychain/credential manager or encrypted file

    # Authenticate (opens browser once, then caches tokens)
    await handler.ensure_authenticated_mcp(
        server_name="notion",
        server_url="https://mcp.notion.com/mcp",
        scopes=["read", "write"],
    )

    # Get ready-to-use headers for any HTTP/SSE/WebSocket call
    headers = await handler.prepare_headers_for_mcp_server(
        "notion",
        "https://mcp.notion.com/mcp"
    )
    print(headers["Authorization"][:30], "...")

asyncio.run(main())

就是这样！ 后续运行使用缓存令牌，无需浏览器。看 完成MCP会话 完整的JSON-RPC+SSE示例。

______________________________________________________________________

您的第一个OAuth流（完整示例）

import asyncio
from chuk_mcp_client_oauth import OAuthHandler

async def main():
    # Create handler - it auto-configures secure storage
    handler = OAuthHandler()

    # Authenticate with a server (opens browser once)
    tokens = await handler.ensure_authenticated_mcp(
        server_name="notion-mcp",
        server_url="https://mcp.notion.com/mcp",
        scopes=["read", "write"]
    )

    print(f"✅ Authenticated! Token: {tokens.access_token[:20]}...")

    # Next time you run this, it uses cached tokens (no browser)
    # Headers are ready to use in your HTTP requests
    headers = await handler.prepare_headers_for_mcp_server(
        server_name="notion-mcp",
        server_url="https://mcp.notion.com/mcp"
    )

    print(f"🔑 Authorization header: {headers['Authorization'][:30]}...")

asyncio.run(main())

使用macOS钥匙串（显式）：

import asyncio
from chuk_mcp_client_oauth import OAuthHandler, TokenManager, TokenStoreBackend

async def main():
    # Explicitly use macOS Keychain for token storage
    # NOTE: 'keyring' library is automatically installed on macOS/Windows
    # No password needed - uses macOS Keychain Access
    token_manager = TokenManager(backend=TokenStoreBackend.KEYCHAIN)
    handler = OAuthHandler(token_manager=token_manager)

    # Authenticate with a server (tokens stored in macOS Keychain)
    tokens = await handler.ensure_authenticated_mcp(
        server_name="notion-mcp",
        server_url="https://mcp.notion.com/mcp",
        scopes=["read", "write"]
    )

    print(f"✅ Authenticated! Token stored in macOS Keychain")
    print(f"🔑 Access Token: {tokens.access_token[:20]}...")

    # You can verify this in Keychain Access app:
    # 1. Open Keychain Access
    # 2. Search for "chuk-oauth"
    # 3. You'll see "notion-mcp" entry under the "chuk-oauth" service

asyncio.run(main())

使用令牌-完整的MCP示例：

现在，让我们使用这些令牌与Notion MCP进行实际交互，列出可用的工具：

import asyncio
import uuid
from chuk_mcp_client_oauth import OAuthHandler, parse_sse_json

async def list_notion_tools():
    """Complete example: Authenticate and list Notion MCP tools."""
    handler = OAuthHandler()
    server_name = "notion-mcp"
    server_url = "https://mcp.notion.com/mcp"

    # Authenticate (uses cached tokens if available)
    print("🔐 Authenticating with Notion MCP...")
    tokens = await handler.ensure_authenticated_mcp(
        server_name=server_name,
        server_url=server_url,
        scopes=["read", "write"]
    )
    print(f"✅ Authenticated! Token: {tokens.access_token[:20]}...")

    # Now use the tokens to make authenticated requests
    session_id = str(uuid.uuid4())

    # Step 1: Initialize MCP session
    print("\n📋 Initializing MCP session...")
    init_response = await handler.authenticated_request(
        server_name=server_name,
        server_url=server_url,
        url=server_url,
        method="POST",
        json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"roots": {"listChanged": True}},
                "clientInfo": {"name": "quickstart-example", "version": "1.0.0"}
            }
        },
        headers={
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json"
        },
        timeout=60.0  # MCP initialization can be slow
    )

    # Extract session ID from response header
    session_id = init_response.headers.get('mcp-session-id', session_id)
    print(f"   ✅ Session initialized: {session_id[:16]}...")

    # Step 2: Send initialized notification
    await handler.authenticated_request(
        server_name=server_name,
        server_url=server_url,
        url=server_url,
        method="POST",
        json={"jsonrpc": "2.0", "method": "notifications/initialized"},
        headers={
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json",
            "Mcp-Session-Id": session_id
        },
        timeout=30.0
    )

    # Step 3: List tools (this is where we use the Bearer token!)
    print("\n🔧 Listing available tools...")
    tools_response = await handler.authenticated_request(
        server_name=server_name,
        server_url=server_url,
        url=server_url,
        method="POST",
        json={"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}},
        headers={
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json",
            "Mcp-Session-Id": session_id
            # Note: Authorization: Bearer  is automatically added!
        },
        timeout=30.0
    )

    # Parse SSE response (MCP servers often return text/event-stream)
    content_type = tools_response.headers.get('content-type', '')
    if 'text/event-stream' in content_type:
        data = parse_sse_json(tools_response.text.strip().splitlines())
    else:
        data = tools_response.json()

    # Display the tools
    if "result" in data and "tools" in data["result"]:
        tools = data["result"]["tools"]
        print(f"\n📦 Found {len(tools)} Notion tools:")
        for tool in tools[:5]:  # Show first 5
            print(f"   • {tool.get('name', 'Unknown')}")
            if 'description' in tool:
                desc = tool['description']
                print(f"     {desc[:80]}{'...' if len(desc) > 80 else ''}")
        if len(tools) > 5:
            print(f"   ... and {len(tools) - 5} more")

    print("\n✅ Complete! Your Bearer token was automatically used in all requests.")
    print(f"   The library added: Authorization: Bearer {tokens.access_token[:20]}...")
    print("   to every HTTP request above.")

asyncio.run(list_notion_tools())

输出：

🔐 Authenticating with Notion MCP...
✅ Authenticated! Token: 282c6a79-d66f-402e-a...

📋 Initializing MCP session...
   ✅ Session initialized: d6b130b8684f5ee9...

🔧 Listing available tools...

📦 Found 15 Notion tools:
   • notion-search
     Perform a search over: - "internal": Semantic search over Notion workspace and c...
   • notion-fetch
     Retrieves details about a Notion entity (page or database) by URL or ID.
Provide...
   • notion-create-pages
     ## Overview
Creates one or more Notion pages, with the specified properties and ...
   • notion-update-page
     ## Overview
Update a Notion page's properties or content.
## Properties
Notion p...
   • notion-move-pages
     Move one or more Notion pages or databases to a new parent.
   ... and 10 more

✅ Complete! Your Bearer token was automatically used in all requests.
   The library added: Authorization: Bearer 282c6a79-d66f-402e-a...
   to every HTTP request above.

幕后发生了什么：

每个HTTP请求都包含您的Bearer令牌：

POST /mcp HTTP/1.1
Host: mcp.notion.com
Authorization: Bearer 282c6a79-d66f-402e-a8f4-27b1c5d3e6f7...
Accept: application/json, text/event-stream
Content-Type: application/json
Mcp-Session-Id: d6b130b8684f5ee9...

{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}

这 authenticated_request() 方法：

1. ✅ 已检索缓存的令牌（无需重新身份验证）
2. ✅ 添加 Authorization: Bearer 每个请求的标题
3. ✅ 自动解析SSE响应
4. ✅ 如果服务器返回401，则会刷新令牌

使用自定义服务名称（针对您的应用程序）：

import asyncio
from chuk_mcp_client_oauth import OAuthHandler, TokenManager, TokenStoreBackend

async def main():
    # Use your own application name for keychain entries
    token_manager = TokenManager(
        backend=TokenStoreBackend.KEYCHAIN,
        service_name="my-awesome-app"  # Custom service name
    )
    handler = OAuthHandler(token_manager=token_manager)

    tokens = await handler.ensure_authenticated_mcp(
        server_name="notion-mcp",
        server_url="https://mcp.notion.com/mcp",
        scopes=["read", "write"]
    )

    print(f"✅ Authenticated! Token stored under 'my-awesome-app' service")

    # In Keychain Access, search for "my-awesome-app" instead of "chuk-oauth"
    # This helps organize tokens for your specific application

asyncio.run(main())

平台特定令牌存储：

• macOS: keyring 已自动安装→ 使用macOS钥匙串（无需密码）
• 视窗: keyring 已自动安装→ 使用Windows凭据管理器（无需密码）
• Linux：安装 pip install chuk-mcp-client-oauth[linux] → 使用特勤局（GNOME/KDE）
• 所有平台：如果平台后端不可用，则回退到加密文件存储

就是这样！ 库处理：

• ✅ OAuth服务器发现
• ✅ 动态客户端注册
• ✅ 打开浏览器征求用户同意
• ✅ 接收回拨
• ✅ 用代码交换代币
• ✅ 安全地存储令牌
• ✅ 在后续运行中重复使用令牌
• ✅ 刷新过期的令牌

每次跑步都会发生什么：

• 首次运行：打开浏览器进行身份验证→ 将令牌保存到存储中
• 二轮：加载缓存的令牌→ 无需浏览器
• 清除令牌后重新运行：再次打开浏览器（如第一次运行）

快速参考：清除令牌以重新运行快速启动

# Method 1: Using CLI (works for all storage backends)
uvx chuk-mcp-client-oauth clear notion-mcp

# Method 2: macOS Keychain (if using Keychain storage)
security delete-generic-password -s "chuk-oauth" -a "notion-mcp"

# Method 3: Delete encrypted file (if using file storage)
rm ~/.chuk_oauth/tokens/notion-mcp.enc
rm ~/.chuk_oauth/tokens/notion-mcp_client.json

# After clearing, run the quickstart again - browser will open

______________________________________________________________________

🧠 了解MCP OAuth（客户端视角）

OAuth流程（实际发生的事情）

当您使用MCP服务器进行身份验证时，幕后会发生以下情况：

1. 🔍 DISCOVERY
   Your app asks: "Server, how do I authenticate with you?"
   Server responds: "Here are my OAuth endpoints and capabilities"

2. 📝 REGISTRATION
   Your app: "I'd like to register as a client"
   Server: "OK, here's your client_id"

3. 🌐 AUTHORIZATION
   Your app opens browser: "User, please approve this app"
   User clicks "Allow"
   Browser redirects back with a code

4. 🎟️ TOKEN EXCHANGE
   Your app: "Here's the code, give me tokens"
   Server: "Here's your access_token and refresh_token"

5. 💾 STORAGE
   Your app saves tokens to secure storage (Keychain/etc)

6. ✅ AUTHENTICATED
   Your app can now make API requests with the token

此库自动化 所有这些步骤.

关键概念

访问令牌 -就像一个临时密码，证明你已获得授权

• 用于每个API请求
• 一段时间后过期（例如1小时）
• 格式： Bearer

刷新令牌 -类似于“获取新密码”令牌

• 用于在新访问令牌过期时获取新访问令牌
• 寿命长（天/周）
• 安全存储

范围 -您正在请求哪些权限

• 示例： ["read", "write"], ["notion:read"]
• 服务器决定授予什么

PKCE -增强安全性，防止代币被盗

• 此库自动处理
• 你不用想

发现 -客户端如何查找OAuth配置

• 符合MCP（RFC 9728）：受保护的资源元数据位于 /.well-known/oauth-protected-resource

- 指向授权服务器元数据 - 包括用于令牌绑定的资源标识符

• 回退（遗留）：直接AS发现 /.well-known/oauth-authorization-server
• WWW身份验证回退：来自401/403响应标头的PRM URL
• 此库通过回退支持自动发现

资源指示器（RFC 8707） -令牌绑定到特定资源

• 令牌绑定到特定的MCP服务器资源
• 防止跨不同资源的令牌重用
• 自动包含在令牌请求中

______________________________________________________________________

📊 流程图

身份验证码+PKCE（桌面/CLI，带浏览器）

这是 主流 此库用于交互式应用程序：

┌──────────────────┐        ┌──────────────┐         ┌──────────────────────┐        ┌───────────────┐
│  MCP Client      │        │  User        │         │  OAuth 2.1 Server    │        │  MCP Server    │
│  (CLI / Agent)   │        │  Browser     │         │  (Auth + Token)      │        │               │
└──┬───────────────┘        └──────┬───────┘         └──────────┬───────────┘        └───────┬───────┘
   │ 1) GET /.well-known/oauth-protected-resource (RFC 9728)   │                             │
   ├────────────────────────────────────────────────────────────────────────────────────────▶│
   │                                                           │ 2) PRM: resource ID,        │
   │◀────────────────────────────────────────────────────────────────────────────────────────┤    AS URLs
   │                                                           │                             │
   │ 3) GET AS metadata from PRM.authorization_servers[0]     │                             │
   ├──────────────────────────────────────────────────────────▶│                             │
   │                                                           │ 4) AS metadata: endpoints   │
   │◀───────────────────────────────────────────────────────────┤                            │
   │                                                           │                             │
   │ 5) Build Auth URL (PKCE: code_challenge)                  │                             │
   │ 6) Open browser ----------------------------------------▶ │                             │
   │                                                           │ 7) User login + consent     │
   │                                                           │◀────────────────────────────┤
   │                                                           │ 8) Redirect with ?code=...  │
   │◀───────────────────────────────────────────────────────────┤  to http://127.0.0.1:PORT   │
   │ 9) Local redirect handler captures code + state           │                             │
   │ 10) POST /token (code + code_verifier + resource=MCP_URL) │                             │
   ├──────────────────────────────────────────────────────────▶│                             │
   │                                                           │ 11) access_token + refresh  │
   │◀───────────────────────────────────────────────────────────┤     (bound to resource)    │
   │ 12) Store tokens securely (keyring / pluggable)           │                             │
   │                                                           │                             │
   │ 13) Connect to MCP with Authorization: Bearer      │                             │
   ├────────────────────────────────────────────────────────────────────────────────────────▶│
   │                                                           │                             │ 14) Session OK
   │◀────────────────────────────────────────────────────────────────────────────────────────┤
   │                                                           │                             │
   │ 15) (When expired) POST /token (refresh_token + resource=MCP_URL)                       │
   ├──────────────────────────────────────────────────────────▶│                             │
   │                                                           │ 16) New access/refresh      │
   │◀───────────────────────────────────────────────────────────┤     -> update secure store │
   │                                                           │                             │

传说：

• PKCE: code_challenge = SHA256(code_verifier) （经授权发送）， code_verifier （以令牌形式发送）
• PRM：受保护的资源元数据（RFC 9728）-符合MCP的发现
• 资源指标: resource= 参数将令牌绑定到特定的MCP服务器（RFC 8707）
• 令牌存储在操作系统密钥链（或可插拔的安全后端）中
• MCP请求携带 Authorization: Bearer

符合MCP的发现流（RFC 9728）

图书馆实施 MCP指定的发现流 自动回退：

🔍 Discovery Attempt 1: Protected Resource Metadata (MCP-Compliant)
   ┌─────────────────────────────────────────────────────────────┐
   │ GET /.well-known/oauth-protected-resource                   │
   │ → Returns: {                                                │
   │     "resource": "https://mcp.notion.com/mcp",               │
   │     "authorization_servers": [                              │
   │       "https://auth.notion.com/.well-known/oauth-as"        │
   │     ]                                                       │
   │   }                                                         │
   └─────────────────────────────────────────────────────────────┘
                              ↓
   ┌─────────────────────────────────────────────────────────────┐
   │ GET https://auth.notion.com/.well-known/oauth-as            │
   │ → Returns AS metadata (authorization_endpoint, etc.)        │
   └─────────────────────────────────────────────────────────────┘

❌ If PRM fails (404/500):

🔍 Discovery Attempt 2: Direct AS Discovery (Fallback)
   ┌─────────────────────────────────────────────────────────────┐
   │ GET /.well-known/oauth-authorization-server                 │
   │ → Returns AS metadata directly                              │
   └─────────────────────────────────────────────────────────────┘

❌ If both fail, check WWW-Authenticate header:

🔍 Discovery Attempt 3: WWW-Authenticate Fallback
   ┌─────────────────────────────────────────────────────────────┐
   │ On 401/403 response:                                        │
   │ WWW-Authenticate: Bearer                                    │
   │   resource_metadata="https://mcp.example.com/.well-known/..." │
   │ → Extract PRM URL and try again                             │
   └─────────────────────────────────────────────────────────────┘

为什么这很重要：

• ✅ 符合MCP规范：遵循模型上下文协议授权规范
• ✅ 令牌绑定：资源指示器阻止跨服务器重用令牌
• ✅ 向后兼容：退回到旧服务器的遗留发现
• ✅ 自动：库透明地处理所有发现方法

设备代码流（无头TTY/SSH代理）

即将发布v0.2.0版本 -非常适合仅限SSH的盒子、CI运行器和后台代理。

计划API：

import asyncio
from chuk_mcp_client_oauth import OAuthHandler

async def main():
    handler = OAuthHandler()

    # Device code flow for headless environments
    await handler.ensure_authenticated_mcp_device(
        server_name="notion",
        server_url="https://mcp.notion.com/mcp",
        scopes=["read", "write"],
        prompt=lambda code, url: print(f"🔐 Go to {url} and enter code: {code}")
    )

    # Rest is identical to auth code flow
    headers = await handler.prepare_headers_for_mcp_server(
        "notion",
        "https://mcp.notion.com/mcp"
    )

asyncio.run(main())

使用案例：

• 仅SSH服务器
• CI/CD管道
• 后台代理
• 共享/无头环境

流程图：

┌──────────────────┐                          ┌──────────────────────┐                          ┌───────────────┐
│  MCP Client      │                          │  OAuth 2.1 Server    │                          │  MCP Server    │
│  (Headless)      │                          │  (Device + Token)    │                          │               │
└──┬───────────────┘                          └──────────┬───────────┘                          └───────┬───────┘
   │ 1) POST /device_authorization (client_id, scope)            │                                         │
   ├────────────────────────────────────────────────────────────▶│                                         │
   │                                                            │ 2) device_code, user_code, verify_uri   │
   │◀────────────────────────────────────────────────────────────┤    expires_in, interval                 │
   │ 3) Show: "Go to VERIFY_URI and enter USER_CODE"            │                                         │
   │                                                            │                                         │
   │                (User on any device)                         │                                         │
   │                         ┌──────────────┐                    │                                         │
   │                         │  User        │ 4) Visit verify URI│                                         │
   │                         │  Browser     │ ◀──────────────────▶│                                         │
   │                         └──────┬───────┘ 5) Enter user code │                                         │
   │                                                            │ 6) Consent + login done                  │
   │                                                            │                                         │
   │ 7) Poll POST /token (device_code, grant_type=device_code)  │                                         │
   ├────────────────────────────────────────────────────────────▶│                                         │
   │ (repeat every `interval` seconds until authorized)         │                                         │
   │◀────────────────────────────────────────────────────────────┤ 8) access_token + refresh               │
   │ 9) Store tokens securely                                   │                                         │
   │ 10) Connect MCP: Authorization: Bearer              │                                         │
   ├─────────────────────────────────────────────────────────────────────────────────────────────────────▶│
   │                                                                                                     │ 11) Session OK
   │◀─────────────────────────────────────────────────────────────────────────────────────────────────────┤
   │ 12) Refresh on expiry → POST /token (refresh_token)         │                                        │
   ├────────────────────────────────────────────────────────────▶│                                        │
   │◀────────────────────────────────────────────────────────────┤ New tokens → update store              │

何时使用设备代码流：

• 仅SSH环境 -目标计算机上没有可用的浏览器
• CI/CD管道 -自动构建需要OAuth，无需交互式登录
• 后台代理 -服务在没有用户交互的情况下运行
• 共享/无头服务器 -多用户，无需桌面环境

令牌如何附加到MCP请求

白板视图： 客户端进行发现，执行OAuth（身份验证代码+PKCE或设备代码），安全存储令牌，并自动附加 Authorization: Bearer 到 每次MCP握手和请求，需要时静静地提神。

HTTP请求：

GET /mcp/api/resources HTTP/1.1
Host: mcp.example.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
Content-Type: application/json

服务器发送事件（SSE）：

GET /mcp/events HTTP/1.1
Host: mcp.example.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
Accept: text/event-stream
Connection: keep-alive

WebSocket：

GET /mcp/ws HTTP/1.1
Host: mcp.example.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
Upgrade: websocket
Connection: Upgrade

______________________________________________________________________

🔍 OAuth发现（应用程序如何查找OAuth端点）

什么是OAuth发现？

MCP服务器在 众所周知的URL。这就像一个菜单，告诉你的应用程序：

• “这是您获得授权的地方”
• “这是您交换代币代码的地方”
• “以下是我支持的内容（PKCE、刷新令牌等）”

符合MCP的发现（先这样做）

根据MCP规范，客户端必须通过受保护的资源元数据（RFC 9728）发现OAuth端点：

步骤1：发现受保护资源元数据（PRM）

# MCP-compliant discovery starts here
GET /.well-known/oauth-protected-resource

PRM响应示例：

{
  "resource": "https://mcp.notion.com/mcp",
  "authorization_servers": [
    "https://mcp.notion.com/.well-known/oauth-authorization-server"
  ],
  "scopes_supported": ["read", "write"],
  "bearer_methods_supported": ["header"]
}

关键PRM字段：

• resource -资源标识符（在 resource= 令牌请求的参数）
• authorization_servers -接下来要获取的AS元数据URL数组

步骤2：获取授权服务器元数据

# Follow the URL from PRM's authorization_servers[0]
GET 

AS元数据响应示例：

{
  "issuer": "https://mcp.notion.com",
  "authorization_endpoint": "https://mcp.notion.com/authorize",
  "token_endpoint": "https://mcp.notion.com/token",
  "registration_endpoint": "https://mcp.notion.com/register",
  "revocation_endpoint": "https://mcp.notion.com/token",
  "response_types_supported": ["code"],
  "response_modes_supported": ["query"],
  "grant_types_supported": ["authorization_code", "refresh_token"],
  "code_challenge_methods_supported": ["plain", "S256"],
  "token_endpoint_auth_methods_supported": ["client_secret_basic", "client_secret_post", "none"]
}

关键AS元数据字段：

• authorization_endpoint -用户批准您的应用程序的地方
• token_endpoint -您在哪里用代码兑换代币
• registration_endpoint -在哪里注册为客户
• code_challenge_methods_supported -支持PKCE（S256=SHA-256）

步骤3：在令牌请求中包含资源指示符

请求令牌时，请包括 resource PRM（RFC 8707）中的参数：

POST /token HTTP/1.1
Host: mcp.notion.com
Content-Type: application/x-www-form-urlencoded

grant_type=authorization_code
&code=AUTH_CODE
&redirect_uri=http://localhost:8080/callback
&client_id=CLIENT_ID
&code_verifier=CODE_VERIFIER
&resource=https://mcp.notion.com/mcp

这将令牌绑定到特定的MCP资源，防止令牌在不同服务器之间重用。

WWW身份验证回退

如果PRM发现失败，MCP服务器 应该 （根据MCP规范惯例）通过以下方式在401/403响应中包含PRM URL WWW-Authenticate 头球

备注：The resource_metadata 参数是 MCP特定惯例，不是核心RFC 6750（承载令牌使用）的一部分。它扩展了标准的承载身份验证方案，以从错误响应中启用OAuth发现，如模型上下文协议授权规范中所述。

HTTP/1.1 401 Unauthorized
WWW-Authenticate: Bearer realm="mcp",
  resource_metadata="https://mcp.notion.com/.well-known/oauth-protected-resource"

标题格式示例：

WWW-Authenticate: Bearer resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource"

WWW-Authenticate: Bearer realm="mcp", error="invalid_token",
  resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource"

客户应当：

1. 解析 resource_metadata 标题中的URL
2. 从该URL获取PRM文档
3. 继续正常的发现流程（上述步骤2）

传统回退（非MCP服务器）

对于 向后兼容 对于不实现PRM发现的服务器，库将退回到直接AS发现：

# Legacy OAuth servers (pre-MCP)
GET /.well-known/oauth-authorization-server

发现优先级：

1. ✅ 第一：尝试PRM /.well-known/oauth-protected-resource （符合MCP标准）
2. ✅ 第二：检查 WWW-Authenticate 401/403响应的标题
3. ✅ 第三：退回到直接AS发现（遗留兼容性）

这个图书馆如何使用Discovery

当您拨打电话时：

tokens = await handler.ensure_authenticated_mcp(
    server_name="notion-mcp",
    server_url="https://mcp.notion.com/mcp",
    scopes=["read", "write"]
)

幕后（符合MCP的流程）：

1. PRM发现：胎儿 https://mcp.notion.com/.well-known/oauth-protected-resource
2. 提取资源：保存 resource 令牌绑定标识符（RFC 8707）
3. AS发现：从获取AS元数据 authorization_servers[0] 统一资源定位符
4. 解析：提取物 authorization_endpoint, token_endpoint等等。
5. 验证：检查是否支持PKCE
6. 缓存：保存配置以备将来使用
7. 令牌请求：包括 resource= 所有令牌请求中的参数
8. 继续：将发现的端点用于OAuth流

后备方案：如果PRM发现失败，则退回到直接AS发现，以实现旧服务器兼容性。

手动发现（高级）

您还可以手动发现端点：

import asyncio
from chuk_mcp_client_oauth import MCPOAuthClient

async def discover_endpoints():
    client = MCPOAuthClient(
        server_url="https://mcp.notion.com/mcp",
        redirect_uri="http://localhost:8080/callback"
    )

    # Discover OAuth configuration
    metadata = await client.discover_authorization_server()

    # Now you can inspect the discovered endpoints
    print(f"Authorization URL: {metadata.authorization_endpoint}")
    print(f"Token URL: {metadata.token_endpoint}")
    print(f"Registration URL: {metadata.registration_endpoint}")
    print(f"Supported scopes: {metadata.scopes_supported}")
    print(f"PKCE methods: {metadata.code_challenge_methods_supported}")

# Run the async function
asyncio.run(discover_endpoints())

使用curl测试Discovery

您可以测试服务器是否支持符合MCP的OAuth发现：

# Step 1: Test PRM discovery (MCP-compliant)
curl https://mcp.notion.com/.well-known/oauth-protected-resource

# Expected response:
# {
#   "resource": "https://mcp.notion.com/mcp",
#   "authorization_servers": ["https://mcp.notion.com/.well-known/oauth-authorization-server"],
#   "scopes_supported": ["read", "write"]
# }

# Step 2: Test AS discovery (from PRM's authorization_servers[0])
curl https://mcp.notion.com/.well-known/oauth-authorization-server

# Expected response: AS metadata with endpoints

# Test your own MCP server
curl https://your-server.com/.well-known/oauth-protected-resource

预期答复：

• PRM：JSON格式 resource, authorization_servers, scopes_supported
• AS元数据：JSON格式 authorization_endpoint, token_endpoint等等。

常见错误：

• 404 Not Found 在PRM上-服务器可能不符合MCP（库将回退到直接AS发现）
• 404 Not Found 两者都有-服务器根本不支持OAuth发现
• Connection refused -服务器URL不正确
• Invalid JSON -服务器配置了错误的OAuth
• {"error":"invalid_token"} -发现终结点的保护不正确（应该是公共的）

测试WWW身份验证回退：

# Make an unauthenticated request to a protected endpoint
curl -i https://mcp.example.com/mcp

# Look for header:
# WWW-Authenticate: Bearer resource_metadata="https://mcp.example.com/.well-known/oauth-protected-resource"

发现规范

MCP OAuth发现如下：

• RFC 9728 -受保护资源元数据（PRM）-主要发现方法
• RFC 8414 -OAuth 2.0授权服务器元数据-从PRM进行二次发现
• RFC 8707 -资源指示器-令牌绑定 resource= 参数

PRM（/.wearned/oauth保护的资源）必须具有：

• resource -资源标识符（用于令牌请求）
• authorization_servers -AS元数据URL数组

AS元数据（/.wearned/oauth授权服务器）必须具有：

• issuer -服务器标识符
• authorization_endpoint -向何处发送用户
• token_endpoint -哪里可以获得代币

应具有（对于MCP）：

• registration_endpoint -动态客户端注册（RFC 7591）
• code_challenge_methods_supported: ["S256"] -PKCE支持
• revocation_endpoint -令牌撤销（RFC 7009）

检查服务器是否支持MCP OAuth的示例：

import asyncio
import httpx

async def check_mcp_oauth_support(server_url: str) -> bool:
    """Check if a server supports MCP-compliant OAuth."""
    # Step 1: Check PRM discovery (MCP-compliant)
    prm_url = f"{server_url}/.well-known/oauth-protected-resource"

    try:
        async with httpx.AsyncClient() as client:
            # Try PRM discovery first
            prm_response = await client.get(prm_url)

            if prm_response.status_code != 200:
                print(f"⚠️  No PRM support (falling back to legacy discovery)")
                # Try legacy AS discovery
                as_url = f"{server_url}/.well-known/oauth-authorization-server"
                as_response = await client.get(as_url)

                if as_response.status_code != 200:
                    print(f"❌ No OAuth support at all")
                    return False

                print("✅ Server supports legacy OAuth (not MCP-compliant)")
                return True

            prm = prm_response.json()

            # Check required PRM fields
            if "resource" not in prm or "authorization_servers" not in prm:
                print("❌ Invalid PRM document")
                return False

            # Step 2: Check AS metadata from PRM
            as_url = prm["authorization_servers"][0]
            as_response = await client.get(as_url)

            if as_response.status_code != 200:
                print(f"❌ AS metadata not available")
                return False

            as_config = as_response.json()

            # Check required AS metadata fields
            required = ["authorization_endpoint", "token_endpoint"]
            if not all(field in as_config for field in required):
                print("❌ Missing required OAuth endpoints")
                return False

            # Check for PKCE support
            if "S256" not in as_config.get("code_challenge_methods_supported", []):
                print("⚠️  PKCE not supported (less secure)")

            # Check for dynamic registration
            if "registration_endpoint" not in as_config:
                print("⚠️  No dynamic registration (manual setup required)")

            print("✅ Server supports MCP-compliant OAuth")
            print(f"   Resource: {prm['resource']}")
            print(f"   Auth: {as_config['authorization_endpoint']}")
            print(f"   Token: {as_config['token_endpoint']}")
            return True

    except Exception as e:
        print(f"❌ Discovery failed: {e}")
        return False

# Usage
asyncio.run(check_mcp_oauth_support("https://mcp.notion.com/mcp"))

______________________________________________________________________

📦 安装选项

# Basic installation
# - macOS: Automatically includes keyring for Keychain support
# - Windows: Automatically includes keyring for Credential Manager support
# - Linux: Uses encrypted file storage by default
uv add chuk-mcp-client-oauth

# Linux with Secret Service support (GNOME/KDE)
uv add chuk-mcp-client-oauth --extra linux

# With HashiCorp Vault support
uv add chuk-mcp-client-oauth --extra vault

# All optional features
uv add chuk-mcp-client-oauth --extra all

# Development installation (includes testing tools)
git clone https://github.com/chrishayuk/chuk-mcp-client-oauth.git
cd chuk-mcp-client-oauth
uv sync --all-extras

平台特定依赖关系：

• macOS/Windows: keyring 自动安装（无需任何操作）
• Linux：添加 [linux] 额外用于特勤局支持，否则使用加密文件
• 企业：添加 [vault] HashiCorp Vault集成的额外功能

您的平台上安装了什么：

______________________________________________________________________

💡 使用示例

示例1：带令牌管理的CLI工具

import asyncio
from chuk_mcp_client_oauth import OAuthHandler

async def connect_to_server(server_name: str, server_url: str):
    """Connect to an MCP server with OAuth."""
    handler = OAuthHandler()

    # First run: Opens browser for auth
    # Subsequent runs: Uses cached tokens
    tokens = await handler.ensure_authenticated_mcp(
        server_name=server_name,
        server_url=server_url,
        scopes=["read", "write"]
    )

    if tokens.is_expired():
        print("⚠️  Token expired, refreshing...")
        # Automatic refresh happens in ensure_authenticated_mcp

    return tokens

# Usage
tokens = asyncio.run(connect_to_server("notion-mcp", "https://mcp.notion.com/mcp"))
print(f"Connected! Token expires in {tokens.expires_in} seconds")

示例2：具有多个服务器的Web应用程序

from chuk_mcp_client_oauth import OAuthHandler

class MCPClient:
    def __init__(self):
        self.handler = OAuthHandler()
        self.servers = {}

    async def add_server(self, name: str, url: str):
        """Add and authenticate with a server."""
        tokens = await self.handler.ensure_authenticated_mcp(
            server_name=name,
            server_url=url,
            scopes=["read", "write"]
        )
        self.servers[name] = url
        return tokens

    async def call_server(self, name: str, endpoint: str):
        """Make authenticated API call."""
        import httpx

        # Get headers with valid token
        headers = await self.handler.prepare_headers_for_mcp_server(
            server_name=name,
            server_url=self.servers[name]
        )

        # Make request
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.servers[name]}{endpoint}",
                headers=headers
            )
            return response.json()

# Usage
mcp = MCPClient()
await mcp.add_server("notion", "https://mcp.notion.com/mcp")
await mcp.add_server("github", "https://mcp.github.com/mcp")

data = await mcp.call_server("notion", "/api/pages")

示例3：低级控制

import asyncio
from chuk_mcp_client_oauth import MCPOAuthClient

async def manual_oauth_flow():
    """Full control over the OAuth process."""
    client = MCPOAuthClient(
        server_url="https://mcp.example.com",
        redirect_uri="http://localhost:8080/callback"
    )

    # Step 1: Discover OAuth endpoints
    metadata = await client.discover_authorization_server()
    print(f"📍 Auth URL: {metadata.authorization_endpoint}")
    print(f"📍 Token URL: {metadata.token_endpoint}")

    # Step 2: Register as a client
    client_info = await client.register_client(
        client_name="My Awesome App",
        redirect_uris=["http://localhost:8080/callback"]
    )
    print(f"📝 Client ID: {client_info['client_id']}")

    # Step 3: Authorize (opens browser)
    tokens = await client.authorize(scopes=["read", "write"])
    print(f"🎟️ Access Token: {tokens.access_token[:20]}...")

    # Step 4: Use the token
    headers = {"Authorization": tokens.get_authorization_header()}

    # Step 5: Refresh when needed
    if tokens.is_expired():
        new_tokens = await client.refresh_token(tokens.refresh_token)
        print(f"🔄 Refreshed: {new_tokens.access_token[:20]}...")

    return tokens

# Run the async function
asyncio.run(manual_oauth_flow())

______________________________________________________________________

🗄️ 令牌存储（安全和自动）

存储工作原理

库会自动将令牌存储在 最安全的位置 对于您的平台：

存储目录

默认情况下，令牌存储在：

~/.chuk_oauth/tokens/

对于加密文件存储：

• 每个服务器都有自己的加密文件： .enc
• 文件使用AES-256加密
• 客户端注册存储为： _client.json
• 加密盐存储为： .salt
• 您可以设置自定义密码或让它自动生成

目录结构示例：

$ ls -la ~/.chuk_oauth/tokens/
total 24
drwx------  5 user  staff  160 Nov  1 12:38 .
drwxr-xr-x  4 user  staff  128 Nov  1 12:38 ..
-rw-------  1 user  staff   16 Nov  1 12:38 .salt
-rw-------  1 user  staff  132 Nov  1 12:38 notion-mcp_client.json
-rw-------  1 user  staff  504 Nov  1 12:38 notion-mcp.enc

检查和清算代币

检查存储了哪些令牌：

# List all stored tokens
uvx chuk-mcp-client-oauth list

# Get details for a specific server (safely redacted)
uvx chuk-mcp-client-oauth get notion-mcp

清除令牌以重新运行演示：

# Option 1: Clear specific server tokens (recommended)
uvx chuk-mcp-client-oauth clear notion-mcp

# Option 2: Logout and revoke with server (best practice)
uvx chuk-mcp-client-oauth logout notion-mcp --url https://mcp.notion.com/mcp

# Option 3: Manual deletion of encrypted files
rm ~/.chuk_oauth/tokens/notion-mcp.enc
rm ~/.chuk_oauth/tokens/notion-mcp_client.json

# Option 4: Delete all tokens
rm -rf ~/.chuk_oauth/

对于macOS钥匙链存储：

使用钥匙扣访问应用程序（GUI）：

1. Open "Keychain Access" app (in Applications > Utilities)
2. Make sure "login" keychain is selected (left sidebar)
3. Search for "chuk-oauth" in the search box (top right)
4. You'll see entries like "notion-mcp" under the service "chuk-oauth"
5. Right-click on the entry → "Delete"
6. Confirm deletion

使用命令行：

# Delete specific server token (e.g., notion-mcp)
security delete-generic-password -s "chuk-oauth" -a "notion-mcp"

# List all tokens stored by this library
security find-generic-password -s "chuk-oauth"

# Search for all entries (shows details)
security find-generic-password -s "chuk-oauth" -g

# Delete all tokens for this library (careful!)
# First, list them to see what you'll delete
security dump-keychain | grep -A 5 "chuk-oauth"
# Then delete each one individually using the account name

示例：从Keychain中删除概念mcp令牌

# Method 1: Using security command
security delete-generic-password -s "chuk-oauth" -a "notion-mcp"

# Method 2: Using the CLI tool (recommended - also clears client registration)
uvx chuk-mcp-client-oauth clear notion-mcp

# Verify it's deleted
security find-generic-password -s "chuk-oauth" -a "notion-mcp"
# Should return: "The specified item could not be found in the keychain."

macOS钥匙链故障排除：

如果在Keychain Access中找不到令牌：

# 1. Check if tokens are actually in Keychain
security find-generic-password -s "chuk-oauth"

# 2. If empty, check if using encrypted file storage instead
ls -la ~/.chuk_oauth/tokens/

# 3. Check which backend is being used
# Run your app and it should log which storage backend it's using

常见问题：

• 在钥匙扣访问应用程序中找不到：确保您在“登录”钥匙链中搜索，而不是在“系统”中搜索
• “找不到”错误：令牌可能已被删除，或正在使用文件存储
• 权限不足：您可能需要在“系统偏好设置”>“隐私与安全”中允许终端/应用程序访问钥匙串

存储示例

自动检测（推荐）

from chuk_mcp_client_oauth import TokenManager

# Automatically uses the best backend for your platform
manager = TokenManager()

# Save tokens
manager.save_tokens("my-server", tokens)

# Load tokens (returns None if not found)
tokens = manager.load_tokens("my-server")

# Check if tokens exist and are valid
if manager.has_valid_tokens("my-server"):
    print("✅ Tokens are valid")

# Delete tokens
manager.delete_tokens("my-server")

显式后端选择

from chuk_mcp_client_oauth import TokenManager, TokenStoreBackend

# Use macOS Keychain
manager = TokenManager(backend=TokenStoreBackend.KEYCHAIN)

# Use encrypted files with custom password
manager = TokenManager(
    backend=TokenStoreBackend.ENCRYPTED_FILE,
    password="my-super-secret-password-123"
)

# Use HashiCorp Vault
manager = TokenManager(
    backend=TokenStoreBackend.VAULT,
    vault_url="https://vault.company.com",
    vault_token="s.xyz123...",
    vault_mount_point="secret",
    vault_path_prefix="mcp-oauth"
)

自定义存储目录

from pathlib import Path

manager = TokenManager(
    backend=TokenStoreBackend.ENCRYPTED_FILE,
    token_dir=Path("/secure/custom/path/tokens"),
    password="my-password"
)

存储安全功能

1. 平台原生安全

- macOS钥匙链：受系统钥匙链访问控制保护 - Windows：受Windows帐户凭据保护 - Linux：受特勤局守护进程保护

1. 加密

- 加密文件存储使用AES-256-GCM - 使用PBKDF2从密码导出的密钥 - 每个令牌文件都有唯一的盐和IV

1. 访问控制

- 使用模式0600创建的文件（仅限所有者读/写） - 使用模式0700创建的令牌目录（仅限所有者访问）

1. 令牌元数据

- 创建时间戳 - 过期跟踪 - 范围信息 - 自动清理过期令牌

检查可用后端

from chuk_mcp_client_oauth import TokenStoreFactory

# Get list of available backends on this system
available = TokenStoreFactory.get_available_backends()
print("Available backends:", available)
# Example output: [TokenStoreBackend.KEYCHAIN, TokenStoreBackend.ENCRYPTED_FILE]

# Get the auto-detected backend
detected = TokenStoreFactory._detect_backend()
print(f"Auto-detected backend: {detected}")
# Example output: TokenStoreBackend.KEYCHAIN (on macOS)

存储最佳做法

发展：

# Use auto-detection for simplicity
manager = TokenManager()

生产（单用户）：

# Use platform-native storage
manager = TokenManager(backend=TokenStoreBackend.AUTO)

生产（多用户服务器）：

# Use Vault for centralized secret management
manager = TokenManager(
    backend=TokenStoreBackend.VAULT,
    vault_url=os.environ["VAULT_URL"],
    vault_token=os.environ["VAULT_TOKEN"]
)

测试：

# Use encrypted files in temp directory
import tempfile
manager = TokenManager(
    backend=TokenStoreBackend.ENCRYPTED_FILE,
    token_dir=Path(tempfile.mkdtemp()),
    password="test-password"
)

______________________________________________________________________

🛠️ CLI工具（快速测试）

该库包括一个用于测试OAuth流的CLI工具。你可以用它来运行 uvx （无需安装）或在本地安装：

使用uvx（推荐-无需安装）

# Authenticate with a server
uvx chuk-mcp-client-oauth auth notion-mcp https://mcp.notion.com/mcp

# List all stored tokens
uvx chuk-mcp-client-oauth list

# Get token details (safely redacted)
uvx chuk-mcp-client-oauth get notion-mcp

# Test connection
uvx chuk-mcp-client-oauth test notion-mcp

# Logout and revoke tokens with server (recommended)
uvx chuk-mcp-client-oauth logout notion-mcp --url https://mcp.notion.com/mcp

# Clear tokens locally only (no server notification)
uvx chuk-mcp-client-oauth clear notion-mcp

使用已安装的CLI

# Install the package first
uv add chuk-mcp-client-oauth

# Then use the chuk-mcp-client-oauth command
chuk-mcp-client-oauth auth notion-mcp https://mcp.notion.com/mcp
chuk-mcp-client-oauth list
chuk-mcp-client-oauth get notion-mcp
chuk-mcp-client-oauth test notion-mcp
chuk-mcp-client-oauth logout notion-mcp --url https://mcp.notion.com/mcp
chuk-mcp-client-oauth clear notion-mcp

使用示例目录

# Or run from examples directory
uv run examples/oauth_cli.py auth notion-mcp https://mcp.notion.com/mcp

输出示例：

============================================================
Authenticating with notion-mcp
============================================================
Server URL: https://mcp.notion.com/mcp
Scopes: read, write (default)

🔐 Starting OAuth flow...
This will open your browser for authorization.

✅ Authentication successful!
Access Token: 282c6a79-d66f-402e-a...********************...w7q85t
Token Type: Bearer
Expires In: 3600 seconds

💾 Tokens saved to secure storage
Storage Backend: KeychainTokenStore

______________________________________________________________________

💻 CLI工具

该库包括一个用于管理OAuth令牌和与MCP服务器交互的命令行界面：

快速开始

# Using uvx (no installation required)
uvx chuk-mcp-client-oauth --help

# Authenticate with an MCP server
uvx chuk-mcp-client-oauth auth notion-mcp https://mcp.notion.com/mcp

# List available tools from an MCP server
uvx chuk-mcp-client-oauth tools notion-mcp https://mcp.notion.com/mcp

# List all servers with tokens
uvx chuk-mcp-client-oauth list

# Get token for a specific server
uvx chuk-mcp-client-oauth get notion-mcp

# Test connection
uvx chuk-mcp-client-oauth test notion-mcp

# Logout (revoke tokens)
uvx chuk-mcp-client-oauth logout notion-mcp --url https://mcp.notion.com/mcp

# Clear tokens locally
uvx chuk-mcp-client-oauth clear notion-mcp

CLI命令

列出MCP工具

这 tools 命令可以很容易地发现MCP服务器提供了什么：

uvx chuk-mcp-client-oauth tools notion-mcp https://mcp.notion.com/mcp

输出：

============================================================
Listing Tools for notion-mcp
============================================================
🔐 Authenticating...
✅ Authenticated

📋 Initializing MCP session...
✅ Session initialized: 7b3c8d2f...

📨 Sending initialized notification...
✅ Notification sent

🔧 Listing available tools...

📦 Found 15 tools:

   • create_page
     Create a new page in Notion

   • search
     Search across your Notion workspace

   • get_page
     Retrieve a specific page by ID

   ... and 12 more

此命令：

• 使用MCP服务器进行身份验证（如果可用，则使用缓存令牌）
• 按照协议初始化正确的MCP会话
• 发送所需 initialized 通知
• 列出所有可用工具及其说明
• 非常适合发现MCP服务器在不编写代码的情况下可以做什么

______________________________________________________________________

📚 工作示例

该库包括完整的工作示例：

1.经过身份验证的请求(authenticated_requests.py) ✅ 新

它显示了什么： 在SSE支持下完成经过身份验证的请求

uv run examples/authenticated_requests.py

演示：

• ✅ httpbin.org示例 -REST API身份验证
• ✅ 完整的概念MCP示例 -支持SSE的完整MCP会话
• 401自动刷新令牌
• SSE（服务器发送事件）响应解析
• MCP会话初始化和工具列表
• 带有身份验证的自定义标头
• 手动401处理
• 错误场景
• 代币生命周期说明

互动示例：

1. httpbin.org REST API（工作演示）
2. 完成Notion MCP会话（列出了15个工具）
3. 使用JSON-RPC的自定义标头
4. 手动401处理
5. 错误处理场景
6. 代币生命周期说明

2.基本的MCP OAuth(basic_mcp_oauth.py)

它显示了什么： 从头开始完成OAuth流程

uv run examples/basic_mcp_oauth.py
# Or with custom server
uv run examples/basic_mcp_oauth.py https://your-mcp-server.com/mcp

3.OAuth处理程序(oauth_handler_example.py)

它显示了什么： 具有令牌缓存的高级API

uv run examples/oauth_handler_example.py

演示：

• 带Notion的MCP OAuth
• 令牌缓存和重用
• 令牌验证
• 标题准备

4.令牌存储(token_storage_example.py)

它显示了什么： 不同的存储后端

uv run examples/token_storage_example.py

演示：

• 自动检测
• 加密文件存储
• 钥匙链集成
• 保险库集成

5.CLI工具(oauth_cli.py)

它显示了什么： 完整的令牌管理工具

uv run examples/oauth_cli.py --help

所有示例均为 功能齐全 并用真实的MCP服务器（Notion MCP）进行了测试。

______________________________________________________________________

🔧 API 参考

快速参考

______________________________________________________________________

OAuthHandler（高级API）

建议用于大多数用例。

from chuk_mcp_client_oauth import OAuthHandler

handler = OAuthHandler(token_manager=None)  # None = auto-detect storage

方法：

• ensure_authenticated_mcp(server_name, server_url, scopes=None)

使用MCP服务器进行身份验证（如果可用，则使用缓存令牌）

  tokens = await handler.ensure_authenticated_mcp(
      server_name="my-server",
      server_url="https://mcp.example.com/mcp",
      scopes=["read", "write"]
  )

• prepare_headers_for_mcp_server(server_name, server_url, scopes=None)

准备好使用具有授权的HTTP标头

  headers = await handler.prepare_headers_for_mcp_server(
      server_name="my-server",
      server_url="https://mcp.example.com/mcp"
  )
  # Use in requests: httpx.get(url, headers=headers)

• get_authorization_header(server_name)

仅获取Authorization标头值

  auth = handler.get_authorization_header("my-server")
  # Returns: "Bearer "

• clear_tokens(server_name)

从缓存和存储中删除令牌（仅限本地）

  handler.clear_tokens("my-server")

• logout(server_name, server_url=None)

注销并撤销服务器令牌（RFC 7009）

  # Revoke tokens with server (recommended)
  await handler.logout(
      server_name="my-server",
      server_url="https://mcp.example.com/mcp"
  )

  # Clear tokens locally only (no server notification)
  await handler.logout("my-server")

备注：何时 server_url 如果提供，图书馆将：

1. 尝试撤销服务器的刷新和访问令牌 1. 从内存缓存中清除令牌 1. 从安全存储中删除令牌 1. 删除客户端注册

如果撤销失败（网络错误，服务器不支持），令牌仍将在本地清除。

MCPOAuthClient（低级别API）

用于对OAuth流进行高级控制。

from chuk_mcp_client_oauth import MCPOAuthClient

client = MCPOAuthClient(
    server_url="https://mcp.example.com/mcp",
    redirect_uri="http://localhost:8080/callback"
)

方法：

• discover_authorization_server() -RFC 8414发现
• register_client(client_name, redirect_uris) -RFC 7591注册
• authorize(scopes) -使用PKCE的完整授权流
• refresh_token(refresh_token) -获取新的访问令牌
• revoke_token(token, token_type_hint=None) -使用服务器撤销令牌（RFC 7009）

客户，

管理安全令牌存储。

from chuk_mcp_client_oauth import TokenManager, TokenStoreBackend

manager = TokenManager(
    backend=TokenStoreBackend.AUTO,  # or KEYCHAIN, VAULT, etc
    token_dir=None,  # custom directory (for ENCRYPTED_FILE)
    password=None,   # password (for ENCRYPTED_FILE)
)

方法：

• save_tokens(server_name, tokens) -安全地存储令牌
• load_tokens(server_name) -检索存储的令牌（如果未找到，则返回None）
• has_valid_tokens(server_name) -检查是否存在有效令牌
• delete_tokens(server_name) -删除令牌

OAuthTokens（令牌对象）

表示OAuth令牌。

tokens = OAuthTokens(
    access_token="...",
    token_type="Bearer",
    expires_in=3600,
    refresh_token="...",
    scope="read write"
)

方法：

• get_authorization_header() -退货 "Bearer "
• is_expired() -检查令牌是否已过期
• to_dict() -转换为词典

______________________________________________________________________

🔐 安全特性

内置安全护栏

1. 仅环回重定向URI（RFC 8252）

- 默认重定向URI: http://127.0.0.1:/callback - 用途 127.0.0.1 （不是 localhost)防止DNS重新绑定攻击 - 随机端口选择防止端口劫持 - 除非明确允许，否则拒绝自定义主机（仅限高级使用）

1. TLS强制

- 公共API做 不 暴露a verify=False 应急出口 - 所有OAuth端点都必须使用HTTPS（回调的环回除外） - 对于使用自定义CA进行开发，请传递自定义 httpx.AsyncClient 使用受信任的CA捆绑包

1. 刷新令牌绑定

- 刷新仅发送给已发现的令牌 token_endpoint 为了 同一发行人+资源 - 绑定到PRM resource 标识符（RFC 8707） - 防止不同MCP服务器之间的令牌重用

1. PKCE实施（RFC 7636）

- 具有S256（SHA-256）的PKCE 总是 用于授权码流 - 代码验证器从不写入磁盘（仅在流期间写入内存） - 状态参数（256位熵）验证回调真实性 - 防止授权码拦截攻击

1. 令牌存储加密

- 平台原生安全存储（macOS钥匙链、Windows凭据管理器、Linux特勤局） - 回退：使用PBKDF2-HMAC-SHA256的AES-256-GCM加密（600000次迭代） - 使用模式0600创建的文件（仅限所有者读/写） - 每个代币文件都有独特的盐和静脉注射

1. 自动过期跟踪

- 跟踪令牌过期时间戳 - 使用前验证令牌 - 令牌过期时自动刷新 - 无明文存储-所有令牌都加密或存储在安全的操作系统存储中

1. 范围验证

- 确保请求的范围与授予的范围匹配 - 防止范围升级攻击

______________________________________________________________________

📊 支持矩阵

OAuth流程和功能

平台和存储

MCP集成

如何将令牌附加到MCP请求：

# The library adds this header to all MCP HTTP requests:
headers = {
    "Authorization": f"Bearer {access_token}",
    "Content-Type": "application/json",
    "Accept": "application/json, text/event-stream",
    "Mcp-Session-Id": ""  # For MCP session requests
}

# For SSE responses (common in MCP), the library can parse:
# Response format:
#   event: message
#   data: {"jsonrpc":"2.0","result":{...}}
#
# Automatically parsed to JSON with parse_sse_response()

使用自动刷新发出经过身份验证的请求

图书馆提供 authenticated_request() 它处理完整的请求生命周期，包括401响应的自动令牌刷新和SSE（服务器发送事件）响应解析：

import asyncio
from chuk_mcp_client_oauth import OAuthHandler

async def main():
    handler = OAuthHandler()

    # Make authenticated JSON-RPC request to MCP server
    # Supports both JSON and SSE response formats
    response = await handler.authenticated_request(
        server_name="notion-mcp",
        server_url="https://mcp.notion.com/mcp",
        url="https://mcp.notion.com/mcp",
        method="POST",
        json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list",
            "params": {}
        },
        headers={
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json",
            "Mcp-Session-Id": ""
        },
        timeout=30.0  # Optional timeout in seconds
    )

    print(f"Status: {response.status_code}")

    # Parse response - automatically handles both JSON and SSE formats
    if 'text/event-stream' in response.headers.get('content-type', ''):
        # SSE response - parse it
        data = parse_sse_response(response.text)
    else:
        # Regular JSON response
        data = response.json()

    print(f"Response: {data}")

asyncio.run(main())

完整的MCP会话示例：

import asyncio
import uuid
from chuk_mcp_client_oauth import OAuthHandler

async def mcp_session_example():
    handler = OAuthHandler()
    server_name = "notion-mcp"
    server_url = "https://mcp.notion.com/mcp"
    session_id = str(uuid.uuid4())

    # Step 1: Initialize MCP session
    init_response = await handler.authenticated_request(
        server_name=server_name,
        server_url=server_url,
        url=server_url,
        method="POST",
        json={
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"roots": {"listChanged": True}},
                "clientInfo": {"name": "my-app", "version": "1.0.0"}
            }
        },
        headers={
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json"
        },
        timeout=60.0  # MCP initialization can be slow
    )

    # Extract session ID from response header
    session_id = init_response.headers.get('mcp-session-id', session_id)
    print(f"Session initialized: {session_id}")

    # Step 2: Send initialized notification
    await handler.authenticated_request(
        server_name=server_name,
        server_url=server_url,
        url=server_url,
        method="POST",
        json={"jsonrpc": "2.0", "method": "notifications/initialized"},
        headers={
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json",
            "Mcp-Session-Id": session_id
        },
        timeout=30.0
    )

    # Step 3: List tools
    tools_response = await handler.authenticated_request(
        server_name=server_name,
        server_url=server_url,
        url=server_url,
        method="POST",
        json={"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}},
        headers={
            "Accept": "application/json, text/event-stream",
            "Content-Type": "application/json",
            "Mcp-Session-Id": session_id
        },
        timeout=30.0
    )

    print(f"Tools: {tools_response.json()}")

asyncio.run(mcp_session_example())

SSE（服务器发送事件）支持：

许多MCP服务器以SSE格式而不是纯JSON格式返回响应。图书馆同时与以下两种机构合作：

def parse_sse_response(response_text: str) -> dict:
    """
    Parse Server-Sent Events (SSE) response format.

    SSE format example:
        event: message
        data: {"jsonrpc":"2.0","result":{...}}

    Returns the JSON data from the SSE message.
    """
    import json

    lines = response_text.strip().split('\n')
    data_lines = []

    for line in lines:
        if line.startswith('data: '):
            data_lines.append(line[6:])  # Remove 'data: ' prefix

    if data_lines:
        json_str = ''.join(data_lines)
        return json.loads(json_str)

    raise ValueError("No data found in SSE response")

# Use with authenticated_request:
response = await handler.authenticated_request(...)

content_type = response.headers.get('content-type', '')
if 'text/event-stream' in content_type:
    data = parse_sse_response(response.text)  # SSE format
else:
    data = response.json()  # Regular JSON

使用JSON的POST请求：

# Create a new resource
response = await handler.authenticated_request(
    server_name="notion-mcp",
    server_url="https://mcp.notion.com/mcp",
    url="https://mcp.notion.com/mcp",
    method="POST",
    json={"jsonrpc": "2.0", "id": 1, "method": "resources/create", "params": {...}}
)

自定义标题：

# Add custom headers to the authenticated request
response = await handler.authenticated_request(
    server_name="notion-mcp",
    server_url="https://mcp.notion.com/mcp",
    url="https://mcp.notion.com/mcp",
    method="POST",
    json={...},
    headers={
        "Accept": "application/json, text/event-stream",
        "Content-Type": "application/json",
        "Mcp-Session-Id": session_id,
        "X-Custom-Header": "value"
    }
)
# Both Authorization and custom headers are included

禁用自动重试：

# If you want to handle 401 responses yourself
try:
    response = await handler.authenticated_request(
        server_name="notion-mcp",
        server_url="https://mcp.notion.com/mcp",
        url="https://mcp.notion.com/mcp",
        method="POST",
        json={...},
        retry_on_401=False  # Don't auto-refresh on 401
    )
except httpx.HTTPStatusError as e:
    if e.response.status_code == 401:
        print("Unauthorized - handle manually")

它是如何工作的：

1. ✅ 确保您拥有有效的令牌（必要时获取/刷新）
2. ✅ 使用以下命令发出HTTP请求 Authorization: Bearer 头球
3. ✅ 支持JSON和SSE（服务器发送事件）响应格式
4. ✅ 如果服务器返回 401 Unauthorized，自动刷新令牌
5. ✅ 使用新令牌重试一次请求
6. ✅ 返回最终响应或提出 httpx.HTTPStatusError 如果仍然未经授权
7. ✅ 支持慢速操作的自定义超时（例如MCP初始化）

______________________________________________________________________

🔒 安全模型和威胁考虑因素

PKCE流安全

什么是PKCE？ PKCE（代码交换证明密钥）可防止授权代码拦截攻击。以下是该库的实现方式：

1. 代码验证器生成

- 为每个流生成随机128个字符串 - 仅存储在内存中（从不写入磁盘） - 代币兑换后销毁

1. 代码挑战

- 发送到授权端点的验证器的SHA-256哈希 - 服务器在令牌交换期间验证验证器匹配 - 防止使用被盗的身份验证码

# Behind the scenes (automatic):
code_verifier = secrets.token_urlsafe(96)  # 128 chars base64url
code_challenge = base64url(sha256(code_verifier))

# Authorization request includes:
# code_challenge=&code_challenge_method=S256

# Token exchange includes:
# code_verifier= (server validates hash matches)

令牌存储安全

静态加密：

# Encrypted File Storage (fallback):
- Algorithm: AES-256-GCM (authenticated encryption)
- Key Derivation: PBKDF2-HMAC-SHA256 (600,000 iterations)
- Salt: 32 bytes random per file
- IV: 16 bytes random per encryption
- Tag: 16 bytes authentication tag

# File structure:
# [32-byte salt][16-byte IV][encrypted data][16-byte tag]

访问控制：

• Unix：使用模式创建的文件 0600 （仅限所有者读/写）
• 视窗：受Windows帐户凭据保护
• 钥匙串：使用系统密钥链访问控制（需要用户身份验证）

令牌生命周期：

1. Access Token Generated → Stored encrypted
2. Access Token Used → Retrieved, decrypted in memory
3. Access Token Expires → Automatic refresh
4. Refresh Token Used → New tokens stored, old deleted
5. User Logout → All tokens deleted from storage

重定向URI策略

默认配置：

# Loopback address (RFC 8252 - OAuth for Native Apps)
redirect_uri = "http://127.0.0.1:/callback"

# Why this is secure:
# ✅ Random port prevents port hijacking
# ✅ 127.0.0.1 (not localhost) prevents DNS rebinding
# ✅ CSRF state parameter validates redirect
# ✅ PKCE verifier prevents code interception

CSRF保护：

# State parameter (RFC 6749):
state = secrets.token_urlsafe(32)  # 256 bits of entropy

# Sent in authorization request, validated on callback
# Prevents cross-site request forgery

自定义重定向URI（高级）：

# For production apps, use custom URI scheme:
client = MCPOAuthClient(
    server_url="https://mcp.example.com/mcp",
    redirect_uri="myapp://oauth/callback"  # Registered scheme
)

安全检查列表

部署此库时：

• \[ \] 使用平台本机存储 （钥匙链/凭证管理器）投入生产
• \[ \] 启用加密 用于文件存储（始终提供密码）
• \[ \] 验证服务器证书 （不要禁用SSL验证）
• \[ \] 使用PKCE （自动启用，不要禁用）
• \[ \] 旋转秘密 （在服务器上配置令牌刷新间隔）
• \[ \] 监控令牌使用情况 （实施日志记录/审计跟踪）
• \[ \] 限制范围 （请求最低必要权限）
• \[ \] 实施注销 （完成后撤销令牌）

未存储的内容

为了安全起见，这些是 从不 写入磁盘：

• ❌ PKCE代码验证器 （仅在流期间存储）
• ❌ CSRF状态参数 （仅在流期间存储）
• ❌ 用户密码 （从未由该图书馆处理）
• ❌ 明文令牌 （始终在文件存储中加密）

______________________________________________________________________

⚠️ 错误处理和恢复

错误分类法

库对不同的故障模式使用特定的异常：

from chuk_mcp_client_oauth.exceptions import (
    OAuthError,              # Base exception
    DiscoveryError,          # Discovery endpoint failed
    RegistrationError,       # Client registration failed
    AuthorizationError,      # User denied consent
    TokenExchangeError,      # Token exchange failed
    TokenRefreshError,       # Token refresh failed
    TokenStorageError,       # Storage backend failed
)

常见错误和解决方案

发现失败

错误： DiscoveryError: Failed to fetch discovery document

原因：

• 服务器不支持OAuth发现
• 网络连接问题
• 服务器URL无效

恢复：

try:
    await handler.ensure_authenticated_mcp(
        server_name="my-server",
        server_url="https://mcp.example.com/mcp"
    )
except DiscoveryError as e:
    print(f"❌ Discovery failed: {e}")

    # Fallback: Manual configuration
    from chuk_mcp_client_oauth import MCPOAuthClient
    client = MCPOAuthClient(
        server_url="https://mcp.example.com/mcp",
        authorization_url="https://mcp.example.com/oauth/authorize",  # manual
        token_url="https://mcp.example.com/oauth/token",  # manual
        redirect_uri="http://127.0.0.1:8080/callback"
    )

授权失败

错误： AuthorizationError: User denied consent

原因：

• 用户在浏览器中点击“拒绝”
• 用户关闭浏览器窗口
• 等待回调超时

恢复：

try:
    tokens = await client.authorize(scopes=["read", "write"])
except AuthorizationError as e:
    if "denied" in str(e).lower():
        print("❌ User denied access")
        print("ℹ️  Please approve the application to continue")
        # Retry with user guidance
    elif "timeout" in str(e).lower():
        print("❌ Authorization timeout")
        print("ℹ️  Please complete the flow within 5 minutes")
        # Retry with longer timeout

令牌刷新失败

错误： TokenRefreshError: Refresh token expired

原因：

• 刷新令牌已过期（服务器配置TTL）
• 服务器已撤销刷新令牌
• 刷新过程中出现网络错误

恢复：

try:
    new_tokens = await client.refresh_token(old_tokens.refresh_token)
except TokenRefreshError as e:
    print(f"❌ Refresh failed: {e}")

    # Clear old tokens and re-authenticate
    handler.clear_tokens("my-server")
    tokens = await handler.ensure_authenticated_mcp(
        server_name="my-server",
        server_url=server_url
    )

存储故障

错误： TokenStorageError: Failed to store token

原因：

• 存储目录上的权限被拒绝
• 钥匙扣锁定（macOS）
• 磁盘已满
• 加密密码错误

恢复：

from chuk_mcp_client_oauth import TokenManager, TokenStoreBackend
from pathlib import Path

try:
    manager = TokenManager(backend=TokenStoreBackend.AUTO)
    manager.save_tokens("server", tokens)
except TokenStorageError as e:
    print(f"❌ Storage failed: {e}")

    # Fallback to encrypted file with explicit password
    import tempfile
    fallback_manager = TokenManager(
        backend=TokenStoreBackend.ENCRYPTED_FILE,
        token_dir=Path(tempfile.mkdtemp()),
        password="explicit-password-123"
    )
    fallback_manager.save_tokens("server", tokens)

重试策略

自动重试（内置）：

# Token refresh automatically retries with exponential backoff
# 3 attempts: 1s, 2s, 4s delays
tokens = await handler.ensure_authenticated_mcp(...)
# ↑ Handles token refresh internally with retries

手动重试（您的代码）：

import asyncio
from tenacity import retry, stop_after_attempt, wait_exponential
from chuk_mcp_client_oauth import OAuthHandler

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=1, max=10)
)
async def connect_with_retry(server_name: str, server_url: str):
    """Connect with automatic retries on network errors."""
    handler = OAuthHandler()
    return await handler.ensure_authenticated_mcp(
        server_name=server_name,
        server_url=server_url
    )

async def main():
    """Main function to run the retry example."""
    try:
        tokens = await connect_with_retry("my-server", "https://mcp.example.com")
        print(f"✅ Connected successfully!")
    except Exception as e:
        print(f"❌ Failed after 3 retries: {e}")

# Usage
asyncio.run(main())

调试

启用调试日志记录：

import logging

# Enable library debug logs
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("chuk_mcp_client_oauth")
logger.setLevel(logging.DEBUG)

# Now you'll see:
# DEBUG:chuk_mcp_client_oauth:Discovering OAuth server at https://...
# DEBUG:chuk_mcp_client_oauth:Found authorization_endpoint: https://...
# DEBUG:chuk_mcp_client_oauth:Registering client with name: ...
# DEBUG:chuk_mcp_client_oauth:Starting local callback server on port 8080
# ...

______________________________________________________________________

🧪 测试

# Run all tests
uv run pytest

# Run with coverage
uv run pytest --cov=chuk_mcp_client_oauth --cov-report=html

# Run specific test file
uv run pytest tests/auth/test_oauth_handler.py -v

# Run with markers
uv run pytest -m "not slow"

测试覆盖范围： 99%（467次测试通过）

测试覆盖矩阵

______________________________________________________________________

🏗️ 发展

# Clone repository
git clone https://github.com/chrishayuk/chuk-mcp-client-oauth.git
cd chuk-mcp-client-oauth

# Install dependencies with uv
uv sync --all-extras

# Run quality checks
make check  # runs format, lint, typecheck, security, tests

# Individual checks
make format      # Format code with ruff
make lint        # Lint code
make typecheck   # Type checking with mypy
make security    # Security scan with bandit
make test        # Run tests
make test-cov    # Run tests with coverage

______________________________________________________________________

🤝 贡献

欢迎投稿！拜托：

1. 分叉存储库
2. 创建要素分支(git checkout -b feature/amazing-feature)
3. 进行更改
4. 运行测试(make check)
5. 承诺(git commit -m 'Add amazing feature')
6. 推(git push origin feature/amazing-feature)
7. 打开拉取请求

______________________________________________________________________

📄 许可证

MIT许可证-请参阅 许可证 文件以获取详细信息。

______________________________________________________________________

🆘 故障排除

“没有名为'keyring'的模块”

uv add keyring  # or pip install keyring

“OAuth流失败”

• 检查服务器URL是否正确且可访问
• 验证服务器是否支持MCP OAuth（具有 .well-known/oauth-authorization-server)
• 确保作用域对服务器有效

“令牌已过期”

# Tokens auto-refresh, but you can manually refresh:
if tokens.is_expired():
    new_tokens = await client.refresh_token(tokens.refresh_token)

令牌存储“权限被拒绝”

# Check directory permissions
ls -la ~/.chuk_oauth/
# Should be drwx------ (700)

# Fix if needed
chmod 700 ~/.chuk_oauth/
chmod 600 ~/.chuk_oauth/tokens/*.enc

______________________________________________________________________

🔗 链接

• 文档: 完整文档
• MCP规范: 模型上下文协议
• OAuth 2.0: RFC 6749
• PKCE: RFC 7636
• 问题:

______________________________________________________________________

由...制作❤️ 卓爱队