Strava OAuth FastAPI与FastMCP后端
一个现代的FastAPI后端,与FastMCP集成,用于OAuth2帐户链接(例如Strava),使用本地SQLite数据库,Authlib用于OAuth,以及用于后台令牌刷新的进程内异步任务。此服务器实现了MCP(模型上下文协议),以便与AI助手和其他MCP客户端无缝集成。
特性
- FastAPI与FastMCP集成,用于模块化项目结构
- MCP资源注册表和配置
- 通过SQLAlchemy 2.0实现SQLite
- 带有Authlib的OAuth2流(AsyncOAuth2Client)
- 使用异步循环刷新后台令牌(进程内)
- 注入依赖关系的数据库会话
- 启动时自动创建表
- 互动文档
/docs和/redoc - MCP兼容资源端点
项目结构
app/
api/
routes/
account_linking.py
health.py
oauth.py
strava.py
core/
app_config.py
db.py
mcp_config.py
mcp_resources.py
models/
user.py
schemas/
strava.py
user.py
services/
authlib_client.py
strava_client.py
user_service.py
main.py设置
- 创建并激活虚拟环境
- macOS/Linux(bash/zsh)
python3 -m venv .venv
source .venv/bin/activate- Windows(PowerShell)
python -m venv .venv
.venv\Scripts\Activate.ps1- 安装依赖项
pip install -r requirements.txt- 配置环境
- 复制 .env.example 到 .env 并填写数值 - 确保 DATABASE_URL 是 sqlite:///./database.db 对于本地文件数据库(默认) - 所需Strava变量: STRAVA_CLIENT_ID, STRAVA_CLIENT_SECRET, STRAVA_REDIRECT_URI
- 运行应用程序
uvicorn app.main:app --reloadSQLite数据库文件 database.db 将在第一次运行时自动创建。
- 测试MCP集成
您可以使用MCP检查器测试MCP服务器:
- 安装MCP检查器: npm install -g @modelcontextprotocol/inspector - 运行检查器: mcp-inspector - 使用URL连接到您的服务器: http://localhost:8000/mcp - 您应该看到可用的MCP工具: read_strava_activities, read_strava_activity, read_strava_athlete,以及 test_endpoint
- 后台令牌刷新
- 应用程序在中启动asyncio任务 app/main.py 每5分钟检查一次Strava用户,并刷新15分钟内到期的令牌。 - 这适用于本地开发和单进程部署。对于多工人生产设置,考虑分布式调度器。
用法
- 健康检查:
GET /health - OAuth流程:
- GET /oauth/authorize 返回Strava授权URL - 在重定向之后, GET /oauth/callback?code=... 使用Authlib交换代码,并保存令牌和运动员信息
MCP集成
FastAPI端点通过FastApiMCP集成自动转换为MCP工具:
- MCP端点:
http://localhost:8000/mcp-主MCP服务器端点 - 测试:使用MCP检查器测试集成(见上面的设置说明)
可用的MCP工具
以下工具是从FastAPI端点自动生成的:
- read_strava_活动 -收集Strava的活动
- 参数: limit (可选,默认值:10) - 返回:包含元数据的活动列表
- read_strava_活动 -获取Strava特定活动的详细信息
- 参数: activity_id (必填) - 返回:详细活动信息
- read_strava_运动员 -获取Strava运动员简介
- 参数:无 - 返回:运动员个人资料信息
- test_endpoint -MCP验证的简单测试端点
- 参数:无 - 返回:确认MCP集成的测试消息
测试MCP工具
一旦通过MCP检查器连接:
- 首先,测试
test_endpoint验证连接 - 使用OAuth流连接Strava帐户(请参阅上面的OAuth部分)
- 测试Strava端点:
read_strava_athlete,read_strava_activities等等。
备注:Strava端点需要经过身份验证的用户。在测试Strava相关工具之前,请先完成OAuth流程。
SQLite快速命令
默认数据库是位于以下位置的本地SQLite文件 database.db 在项目根中。
打开数据库:
sqlite3 database.db里面 sqlite3 提示:
- 列出表格
.tables- 显示的架构
users桌子
.schema users- 预览来自的行
users
SELECT id, email, provider, expires_at, created_at, updated_at FROM users LIMIT 10;- 放下
users表(小心:破坏性)
DROP TABLE IF EXISTS users;- 退出sqlite3
.quit非交互式CLI示例(单行):
# List tables
sqlite3 database.db '.tables'
# Show users schema
sqlite3 database.db '.schema users'
# Preview a few rows
sqlite3 -header -column database.db "SELECT id, email, provider, expires_at, created_at, updated_at FROM users LIMIT 10;"
# Drop users table (destructive!)
sqlite3 database.db "DROP TABLE IF EXISTS users;"模拟即将到期的令牌
要在不等待的情况下触发背景刷新,您可以设置 expires_at 对于拥有 refresh_token.循环在 app/main.py 款待 expires_at IS NULL AND refresh_token IS NOT NULL 因为需要刷新。
将下面的电子邮件替换为数据库中现有的Strava链接用户:
# Force token refresh by making expires_at NULL (requires refresh_token to be present)
sqlite3 database.db "UPDATE users SET expires_at = NULL WHERE email = 'your_user@example.com';"
# Alternatively, set expires_at to 30 seconds from now (also triggers near-expiration)
sqlite3 database.db "UPDATE users SET expires_at = datetime('now', '+30 seconds') WHERE email = 'your_user@example.com';"
# Verify
sqlite3 -header -column database.db "SELECT email, expires_at, refresh_token FROM users WHERE email = 'your_user@example.com';"备注
- 上的令牌持久性字段
User包括:access_token,refresh_token,expires_at以及时间戳。 - 如果您更改了数据库模型,请使用迁移(例如Alembic)或在开发过程中删除/重新创建SQLite数据库。
- 遵循FastAPI最佳实践来扩展路由、服务和模型。
FastAPI与FastMCP集成
此服务器使用FastAPI和FastMCP集成来支持MCP协议。关键组件包括:
- FastAPI应用程序:主要应用程序使用
FastAPI作为基础框架。 - FastMCP集成:The
FastApiMCP类用于向FastAPI应用程序添加MCP功能。 - API终点:定义常规FastAPI端点并自动转换为MCP工具。
- MCP配置:服务器功能和元数据在中定义
app/core/mcp_config.py. - 资源模型:Strava活动和运动员档案的Pydantic模型在
app/schemas/strava.py.
要作为MCP客户端与此服务器交互,请执行以下操作:
- 连接到MCP服务器
http://localhost:8000/mcp. - MCP协议自动处理工具发现。
- 通过可用的MCP工具访问Strava数据。
- 身份验证与以前一样通过OAuth2处理。
MCP集成故障排除
如果您遇到MCP集成问题,以下是一些常见问题和解决方案:
错误:未返回工具
当MCP服务器未正确注册任何工具时,会发生此错误。要解决此问题,请执行以下操作:
- 最重要的:定义所有FastAPI端点后创建FastApiMCP实例。FastApiMCP在创建时需要分析现有路由。
- 添加
operation_id将参数发送到FastAPI端点,以便更好地识别工具。 - 使用向端点添加描述性元数据
summary和description参数。 - 使用响应模型(Pydantic模型)来提供更好的模式信息。
- 检查端点是否返回有效响应。
测试MCP端点
您可以通过以下方式测试您的MCP服务器是否正常工作:
- 使用MCP检查器(推荐):连接到
http://localhost:8000/mcp - 检查服务器日志中的工具注册消息
- 验证
/mcp端点响应MCP协议请求
如果MCP检查器未显示任何工具,请检查FastAPI端点的配置和FastApiMCP初始化。
