11.2 服务端授权服务器:实现客户端与令牌存储

11.2 服务端授权服务器:实现客户端与令牌存储

本节摘要:本节讲 MCP 服务端如何实现授权服务器。SDK 提供了完整的授权服务器骨架——授权码流、令牌签发、客户端注册、令牌吊销、元数据这些标准端点,SDK 都替你实现了。你只需实现「客户端注册存储」「令牌存储」「授权码存储」这几个存储接口,SDK 跑全套 OAuth 流程。再加一个 Bearer 令牌中间件,自动校验每个进来的请求。读完本节,你能为 MCP 服务端加上 OAuth 保护。

一、SDK 提供的授权服务器骨架

SDK 的授权服务器实现了 OAuth 2.1 的标准端点:

这些端点的协议逻辑 SDK 都实现了——授权码怎么生成、令牌怎么签发、PKCE 怎么校验,都不用你管。你要做的,只是实现几个存储接口,让 SDK 知道「客户端信息、令牌、授权码存哪」。

二、你要实现的存储接口

SDK 定义了几个存储接口(概念性),你实现它们:

from mcp.server.auth.provider import AuthorizationServerProvider

class MyProvider(AuthorizationServerProvider):
    # 1. 客户端注册存储:存/查客户端信息
    async def get_client(self, client_id: str) -> Client:
        return await db.get_client(client_id)
    async def register_client(self, client_info) -> Client:
        return await db.save_client(client_info)

    # 2. 授权码存储:存/校验授权码
    async def save_authorization_code(self, code, info):
        await db.save_auth_code(code, info)
    async def get_authorization_code(self, code):
        return await db.get_auth_code(code)

    # 3. 令牌存储:签发/校验/吊销令牌
    async def issue_token(self, client_id, scopes) -> Token:
        return await db.create_token(client_id, scopes)
    async def validate_token(self, token: str) -> TokenInfo:
        return await db.get_token(token)
    async def revoke_token(self, token):
        await db.delete_token(token)
​

每个存储接口对应 OAuth 流程的一个环节。你实现这些方法(通常用数据库做后端),SDK 在 OAuth 流程的对应节点调用它们。

三、存储的选择

存储接口怎么实现,取决于你的规模:

# 开发用:内存存储
class MemoryProvider(AuthorizationServerProvider):
    def __init__(self):
        self.clients = {}
        self.tokens = {}
        self.auth_codes = {}
    # ... 实现各方法,操作内存字典

# 生产用:Redis 存储
class RedisProvider(AuthorizationServerProvider):
    def __init__(self, redis):
        self.redis = redis
    # ... 实现各方法,操作 Redis(带过期)
​

💡 技巧:令牌用 Redis 存,自动过期。令牌有 expires_in,Redis 的 TTL 天然匹配——存令牌时设 TTL,过期自动清理,不用手动删。

四、Bearer 令牌中间件

实现授权服务器后,还要保护资源——让每个进来的请求都校验令牌。这就是 Bearer 令牌中间件:

from mcp.server import MCPServer
from mcp.server.auth import AuthSettings

mcp = MCPServer(
    "Protected",
    auth=AuthSettings(
        provider=my_provider,           # 你的存储实现
        # 其他认证配置
    )
)

# 注册的工具/资源现在受保护
@mcp.tool()
def get_secret_data() -> dict:
    """This tool now requires a valid token."""
    return {"secret": "only authenticated users see this"}
​

加了 auth 配置后,SDK 在每个请求前自动校验 Bearer 令牌:

请求到达(带 Authorization: Bearer <token>)
   │
   ▼ Bearer 中间件
提取 token
   │
   ▼ 调 my_provider.validate_token(token)
有效?
   是 → 放行,执行工具
   否 → 返回 401 Unauthorized
​

所有工具/资源自动受保护,你不用在每个工具里手动校验——中间件在前面统一处理。

五、完整的受保护服务端

把存储实现 + Bearer 中间件合起来,一个完整的受保护 MCP 服务端:

from mcp.server import MCPServer
from mcp.server.auth import AuthSettings

# 1. 实现存储
class MyProvider(AuthorizationServerProvider):
    # ... 实现 get_client、issue_token、validate_token 等
    ...

provider = MyProvider(...)

# 2. 构造受保护的服务端
mcp = MCPServer(
    "Protected",
    auth=AuthSettings(provider=provider),
)

# 3. 注册工具(自动受保护)
@mcp.tool()
def get_user_data(user_id: str) -> dict:
    """Get user data (requires auth)."""
    return fetch_user(user_id)

# 4. 跑成流式 HTTP 服务
if __name__ == "__main__":
    mcp.run("streamable-http", host="0.0.0.0", port=8000)
​

现在这个服务端:

• 有完整的 OAuth 端点(/authorize、/token 等)
• 所有工具受 Bearer 令牌保护
• 客户端必须持令牌才能调用

六、授权服务器骨架的价值

总结 SDK 授权服务器骨架的价值:

⚠️ 注意:实现存储时要小心安全。令牌要用强随机生成(别用自增 ID),授权码要一次性(用完即删),密码/token 不能明文存。OAuth 的安全细节很多,实现存储时务必参考 OAuth 2.1 安全最佳实践,或直接用成熟库的存储后端。

七、本节与全章的衔接

本节讲透了服务端的授权服务器实现。接下来:

• 第 11.3 节:客户端怎么配合——OAuthClientProvider 让 Client 自动获取令牌
• 第 11.4 节:机器到机器场景——客户端凭证扩展
• 第 11.5 节:对接外部 IdP——身份断言扩展

服务端、客户端、特殊场景,四节合起来构成 OAuth 全链路。

本节要点回顾

1. SDK 提供完整授权服务器骨架:授权码流、令牌签发、注册、吊销、元数据,协议逻辑 SDK 实现。
2. 你只需实现存储接口:客户端注册、授权码、令牌的存取。
3. 存储选择:开发用内存,生产用 Redis/数据库,企业用外部 IdP。
4. 令牌用 Redis 自动过期最省事,TTL 匹配 expires_in。
5. Bearer 令牌中间件自动校验每个请求,所有工具/资源受保护。
6. 价值:不用从零写 OAuth、减少出错、一致行为、关注点分离。
7. 实现存储要注意安全:强随机、一次性授权码、不明文存敏感信息。

服务端授权清楚了,下一节讲客户端怎么配合获取令牌。