1. 认证入门——OAuth 与受保护资源 #
本章主要介绍如何使用 MCP 的认证机制，包括在服务端实现 TokenVerifier 来验证访问令牌，以及在客户端侧使用 OAuth 认证连接受保护的 MCP 服务器。你将学会：
- 在服务端实现
TokenVerifier来验证访问令牌； - 在客户端侧使用 OAuth 认证连接受保护的 MCP 服务器。
2. 服务器：OAuth 资源服务器与受保护资源 #
新建 C:\mcp-project\oauth_server.py：
# 导入 AnyHttpUrl，用于验证 URL 格式
from pydantic import AnyHttpUrl
# 导入认证相关类型：AccessToken（访问令牌）、TokenVerifier（令牌验证器基类）
from mcp.server.auth.provider import AccessToken, TokenVerifier
# 导入认证设置 AuthSettings
from mcp.server.auth.settings import AuthSettings
# 导入 FastMCP 高层服务器主类
from mcp.server.fastmcp import FastMCP
# 实现一个简化的 TokenVerifier
class SimpleTokenVerifier(TokenVerifier):
"""令牌验证器"""
# 异步方法：验证访问令牌，返回 AccessToken 或 None
async def verify_token(self, token: str) -> AccessToken | None:
"""验证访问令牌并返回令牌信息。"""
# 实际应用中应在此处：
# 1. 验证 JWT 签名
# 2. 检查令牌过期时间
# 3. 验证令牌的受众（audience）
# 4. 检查令牌的权限范围（scope）
if token.startswith("_token_"):
# 提取用户ID（模拟解析令牌）
user_id = token.replace("_token_", "")
# 构造并返回 AccessToken 对象
return AccessToken(
token=token, # 原始令牌
sub=user_id, # 用户标识
aud="mcp-user", # 受众
scope="user", # 权限范围
exp=9999999999, # 过期时间
client_id="client_id", # 客户端ID
scopes=["user"], # 权限范围列表
)
# 如果令牌无效，返回 None
return None
# 创建 FastMCP 实例，配置为资源服务器
mcp = FastMCP(
# 服务器名称
name="OAuth Server",
# 配置令牌验证器
token_verifier=SimpleTokenVerifier(),
# 配置认证设置（用于 RFC 9728 Protected Resource Metadata）
auth=AuthSettings(
# 授权服务器 URL
issuer_url=AnyHttpUrl("https://auth.example.com"),
# 本服务器 URL
resource_server_url=AnyHttpUrl("http://127.0.0.1:8000"),
# 要求的权限范围
required_scopes=["user"],
),
)
# 注册受保护的资源（需要有效令牌才能访问）
@mcp.resource("user://profile")
def get_user_profile() -> str:
"""获取用户资料（受保护资源）。"""
# FastMCP 会自动进行认证检查，无需手动校验令牌
# 如果令牌无效或缺失，FastMCP 会自动拒绝访问
return """{
"name": "User",
"email": "user@example.com",
"role": "user"
}"""
# 注册受保护的工具（需要有效令牌才能调用）
@mcp.tool()
def get_user_data(user_id: str = "current") -> dict[str, str]:
"""获取用户数据（受保护工具）。"""
# 工具调用的认证检查同样是自动的
if user_id == "current":
# 返回当前用户的模拟数据
return {
"user_id": "user_001",
"status": "active",
"last_login": "2024-01-01T00:00:00Z",
}
else:
# 返回指定用户的错误信息
return {"user_id": user_id, "status": "unknown", "error": "User not found"}
# 主程序入口：以 Streamable HTTP 方式运行，便于演示 OAuth 流程
if __name__ == "__main__":
# 使用 Streamable HTTP 传输，默认监听 127.0.0.1:8000
mcp.run(transport="streamable-http")要点：
- 实现
SimpleTokenVerifier类来验证访问令牌； - 配置
AuthSettings声明本服务器为资源服务器； - 受保护的资源与工具会自动进行认证检查。
3. 客户端：使用 OAuth 认证连接受保护服务器 #
新建 C:\mcp-project\oauth_client.py：
# 导入异步IO库
import asyncio
# 从mcp包导入客户端会话
from mcp import ClientSession
# 从mcp客户端包导入streamable http客户端
from mcp.client.streamable_http import streamablehttp_client
# 定义主异步函数
async def main():
# OAuth 认证流程说明
print(" 启动 OAuth 客户端...")
# 设置服务器URL
server_url = "http://127.0.0.1:8000/mcp"
# 打印连接信息
print(f" 连接到: {server_url}")
# 提示用户输入模拟OAuth令牌
print(" 请输入令牌（格式：_token_用户名）：")
# 读取用户输入并去除首尾空白
token = input().strip()
# 检查令牌格式是否正确
if not token.startswith("_token_"):
# 格式错误提示
print(" 令牌格式错误，应该以 '_token_' 开头")
return
# 打印使用的令牌
print(f" 使用令牌: {token}")
try:
# 创建自定义认证头
custom_headers = {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json",
}
# 打印认证尝试信息
print(" 尝试使用 Bearer 令牌认证...")
# 使用自定义头部连接到服务器
async with streamablehttp_client(server_url, headers=custom_headers) as (
read,
write,
_,
):
# 连接成功提示
print(" HTTP 连接成功")
# 创建MCP客户端会话
async with ClientSession(read, write) as session:
# 会话创建成功提示
print(" MCP 会话创建成功")
# 尝试初始化会话
try:
# 初始化会话
await session.initialize()
# 初始化成功提示
print(" 会话初始化成功")
except Exception as e:
# 初始化失败提示
print(f" 会话初始化失败: {e}")
return
# 尝试列出受保护资源
try:
# 获取资源列表
resources = await session.list_resources()
# 打印资源URI
print(f" 受保护资源: {[r.uri for r in resources.resources]}")
except Exception as e:
# 获取资源失败提示
print(f" 获取资源列表失败: {e}")
# 尝试列出受保护工具
try:
# 获取工具列表
tools = await session.list_tools()
# 打印工具名称
print(f" 受保护工具: {[t.name for t in tools.tools]}")
except Exception as e:
# 获取工具失败提示
print(f" 获取工具列表失败: {e}")
# 尝试读取受保护资源
try:
# 导入AnyUrl用于资源URI
from pydantic import AnyUrl
# 读取指定资源
resource_result = await session.read_resource(
AnyUrl("user://profile")
)
# 用于存储文本内容
texts = []
# 遍历资源内容块
for block in resource_result.contents:
# 获取文本内容并加入列表
texts.append(getattr(block, "text", ""))
# 打印读取结果
print(f" 读取资源成功: {' | '.join(texts)}")
except Exception as e:
# 读取资源失败提示
print(f" 读取资源失败: {e}")
# 尝试调用受保护工具
try:
# 调用工具get_user_data
tool_result = await session.call_tool(
"get_user_data", {"user_id": "current"}
)
# 用于存储文本内容
texts = []
# 遍历工具返回内容块
for block in tool_result.content:
# 判断内容类型为"text"时提取文本
if block.type == "text":
texts.append(getattr(block, "text", ""))
# 打印调用结果
print(f" 调用工具成功: {' | '.join(texts)}")
except Exception as e:
# 调用工具失败提示
print(f" 调用工具失败: {e}")
except Exception as e:
# 连接失败提示
print(f" 连接失败: {e}")
# 打印错误类型
print(f"错误类型: {type(e).__name__}")
# 判断是否为认证失败
if "401" in str(e) or "Unauthorized" in str(e):
print(" 认证失败，请检查令牌是否正确")
else:
# 其他异常打印详细堆栈
import traceback
traceback.print_exc()
# 判断是否为主程序入口
if __name__ == "__main__":
try:
# 运行主异步函数
asyncio.run(main())
except KeyboardInterrupt:
# 捕获Ctrl+C退出
print("\n 客户端已停止")
except Exception as e:
# 启动失败提示
print(f" 客户端启动失败: {e}")
print(" 请确保 OAuth 服务器正在运行")
说明：
- 使用
OAuthClientProvider处理 OAuth 认证流程。 - 实现
streamablehttp_client函数来创建 HTTP 客户端。 - 使用
ClientSession管理 MCP 会话。 - 通过
headers参数传递认证信息。
4. 运行与验证 #
步骤 1：启动 OAuth 服务器 #
cd C:\mcp-project
python oauth_server.py步骤 2：运行 OAuth 客户端 #
cd C:\mcp-project
python oauth_client.py当客户端提示输入令牌时，输入：_token_user001
预期输出：
启动 OAuth 客户端...
连接到: http://127.0.0.1:8000/mcp
请输入令牌（格式：_token_用户名）：
_token_user001
使用令牌: _token_user001
尝试使用 Bearer 令牌认证...
HTTP 连接成功
MCP 会话创建成功
会话初始化成功
受保护资源: [AnyUrl('user://profile')]
受保护工具: ['get_user_data']
读取资源成功: {
"name": "User",
"email": "user@example.com",
"role": "user"
}
调用工具成功: {
"user_id": "user_001",
"status": "active",
"last_login": "2024-01-01T00:00:00Z"
}