如何让FastMCP 3.2.0 from_openapi()携带用户OAuth令牌调用API

要让自动生成的OpenAPI工具把用户的真实令牌作为Authorization头发送给后端API，你需要通过自定义HTTP客户端拦截器动态注入令牌，具体实现步骤如下：

1. 定义请求拦截钩子 #

创建异步钩子函数，在每个请求发送前从FastMCP上下文获取当前用户的access token，并添加到请求头：

async def inject_user_token(request: httpx.Request) -> None:
    from fastmcp.server.dependencies import get_access_token
    token = get_access_token()
    if token is not None:
        request.headers["Authorization"] = f"Bearer {token.raw}"

2. 配置带钩子的HTTP客户端 #

初始化httpx.AsyncClient实例，将钩子绑定到request事件：

async_client = httpx.AsyncClient(event_hooks={"request": [inject_user_token]})

3. 传入自定义客户端初始化FastMCP #

在FastMCP.from_openapi中通过http_client参数传入配置好的客户端：

mcp = FastMCP.from_openapi(
    openapi_spec=openapi_spec,
    name="My API Server",
    auth=auth,
    http_client=async_client,
)

完整修改后的代码 #

import httpx
from fastmcp import FastMCP
from fastmcp.server.auth import OIDCProxy
from fastmcp.server.auth.providers.jwt import JWTVerifier

# Configure token validation for your identity provider
token_verifier = JWTVerifier(
    jwks_uri="<redacted>",
    issuer="<redacted>",
)

auth = OIDCProxy(
    config_url="<redacted>",
    client_id="<redacted>",
    client_secret="<redacted>",
    base_url="http://localhost:8080/",
    token_verifier=token_verifier
)

# Load your OpenAPI spec 
openapi_spec = httpx.get("<redacted>/swagger.json").json()

# 注入用户令牌的请求钩子
async def inject_user_token(request: httpx.Request) -> None:
    from fastmcp.server.dependencies import get_access_token
    token = get_access_token()
    if token is not None:
        request.headers["Authorization"] = f"Bearer {token.raw}"

# 配置带钩子的异步客户端
async_client = httpx.AsyncClient(event_hooks={"request": [inject_user_token]})

# 初始化FastMCP并传入自定义客户端
mcp = FastMCP.from_openapi(
    openapi_spec=openapi_spec,
    name="My API Server",
    auth=auth,
    http_client=async_client,
)

# 自定义测试工具
@mcp.tool
async def get_user_info() -> dict | None:
    """Returns information about the authenticated LBP user."""
    from fastmcp.server.dependencies import get_access_token
    
    token = get_access_token()
    if token is not None:
        return {
            "name": token.claims.get("name"),
            "email": token.claims.get("email")
        }
    else:
        return None

if __name__ == "__main__":
    mcp.run(transport="streamable-http", port=8080, host="localhost")

原理说明 #

• 拦截钩子会在自动生成工具调用后端API前触发，从请求上下文获取当前用户的原始令牌，动态添加Authorization头。
• 完全替代静态令牌方案，确保每个请求携带的是当前用户的真实有效令牌。
• 自定义工具中get_access_token()的原有逻辑不受影响，可正常获取用户身份信息。

内容的提问来源于stack exchange，提问作者Hervé Brun