如何在FastMCP 3.2.0中使用from_openapi() 调用受OAuth保护的API？

后端开发 2026-07-10

我有一个REST API，它暴露了自己的OpenAPI定义。
该API需要身份验证（Bearer令牌，OAuth2身份提供者）。

我使用FastMCP的 OpenAPI功能为这个API生成了一个Python MCP服务器（参见 https://gofastmcp.com/integrations/openapi）。

用户必须经过身份验证才能使用FastMCP服务器。
我使用一个FastMCP的 OIDC代理来实现这一目标（https://gofastmcp.com/servers/auth/oidc-proxy），因为OAuth IdP不处理DCR请求（动态客户端注册）。

我的问题相当简单，但我找不到相关信息：当用户调用一个自动生成的工具时，如何确保FastMCP在发送请求时附带从IdP收到的真实令牌作为Authorization头？实际上，MCP现在并不会发送任何Authorization头。

我的代码：

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(
    # Provider's configuration URL
    config_url="<redacted>",

    # Your registered app credentials
    client_id="<redacted>",
    client_secret="<redacted>",

    # Your FastMCP server's public URL
    base_url="http://localhost:8080/",

    token_verifier=token_verifier
)

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

# Initialize FastMCP with OpenAPI specification
mcp = FastMCP.from_openapi(
    openapi_spec=openapi_spec,
    name="My API Server",
    auth=auth,
)

# Add a protected tool to test authentication
@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:
        # The provider stores user data in token claims
        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")
    # mcp.run()

对 get_access_token() 在 get_user_info 工具中的调用返回了我想要的令牌！但自动生成的工具似乎并不使用此函数来传输Authorization头。

我不能使用这里描述的方法：https://gofastmcp.com/integrations/openapi#authentication。
事实上，这种方法使用一个静态令牌，必须写入代码或MCP服务器的配置文件。我需要用户的令牌！

解决方案

好吧，这么简单就解决了！

我必须添加一个httpx钩子，在所有外发请求中添加Authorization头：

async def add_bearer_auth(request: httpx.Request) -> httpx.Request:
    '''
    Populates an Authorization header (Bearer), with the token from the IdP, in the request.
    '''
    token = get_access_token()
    if token is not None:
        request.headers["Authorization"] = f"Bearer {token.token}"
    return request

# HTTP client for the targeted API
client = httpx.AsyncClient(
    base_url="https://<redacted>",
    event_hooks={"request": [add_bearer_auth]}
)

# Initialize FastMCP with OpenAPI specification
mcp = FastMCP.from_openapi(
...
    client=client,
...
)

API