运动跟踪器SoR(Postgres)
实现Workout Tracker MCP工具的数据库模式、摄取有效负载验证和事务写入路径。
先决条件
- Python 3.11+
- PostgreSQL(本地或容器)
DATABASE_URL指向数据库的环境变量,例如。
export DATABASE_URL=postgresql+psycopg://postgres:postgres@localhost:5432/workout_tracker设置
- 安装依赖项:
pip install -r requirements.txt- 通过配置数据库URL
DATABASE_URL.
迁移
运行Alembic migrations以创建架构:
alembic upgrade head运行MCP服务器
启动MCP服务器(流式HTTP传输):
python3 server.py认证
服务器需要Google OIDC ID令牌(由Google签名并使用 email_verified=true).配置: 集 GOOGLE_CLIENT_ID 添加到您的OAuth客户端ID以强制令牌受众。 集 GOOGLE_CLIENT_SECRET 用于Google令牌交换的OAuth客户端密钥。 可选设置 API_KEY 允许静态密钥通过 Authorization: Bearer . 可选设置 API_KEYS_FILE 加载多个密钥(默认值: api_keys.txt,逗号或换行符分隔)。 集 RESOURCE_SERVER_URL 到您的公共MCP URL(例如。,https://.../mcp)因此OAuth发现可以找到受保护的资源元数据。
要获取用于本地测试的ID令牌,请使用基于浏览器的登录或 gcloud:
gcloud auth application-default login
gcloud auth print-identity-token --audiences ""服务器在以下位置公开OAuth保护的资源元数据 /.well-known/oauth-protected-resource{path} 以及OAuth服务器元数据 /.well-known/oauth-authorization-server 用于OAuth发现。它还代理Google OAuth /oauth/authorize, /oauth/token,以及 /oauth/register 以确保设置范围并支持动态客户端注册。
将令牌发送到 Authorization 所有MCP请求的标头,例如:
curl -H "Authorization: Bearer $(gcloud auth print-identity-token --audiences \"\")" \
http://localhost:8000/mcp
# or with API key
curl -H "Authorization: Bearer $API_KEY" http://localhost:8000/mcp
# or with key from file
curl -H "Authorization: Bearer " http://localhost:8000/mcpMCP工具有效载荷示例
有效载荷 add_workout_entry:
{
"user_id": "b8d932e9-26ef-4f2d-8b7f-cc1e0a3e3b2c",
"idempotency_key": "workout-2024-09-01-1",
"workout": {
"started_at": "2024-09-01T10:00:00Z",
"ended_at": "2024-09-01T11:00:00Z",
"timezone": "America/Los_Angeles",
"title": "Upper Body",
"source": "manual",
"notes": "Felt strong"
},
"exercises": [
{
"display_name": "Bench Press",
"canonical_name": "bench press",
"notes": "Working sets",
"sets": [
{"reps": 8, "weight": {"value": 135, "unit": "lb"}, "rpe": 7.5},
{"reps": 6, "weight": {"value": 62.5, "unit": "kg"}, "rpe": 8.0}
]
}
]
}演示摄取
一个小的辅助脚本可以从Python shell运行:
from sqlalchemy.orm import Session
from src.db.session import engine
from src.service.ingest_workout import ingest_workout
import json
with open("examples/sample_workout.json") as f:
payload = json.load(f)
with Session(engine) as session:
result = ingest_workout(session, payload)
print(result)测试
测试期望通过以下方式提供实时PostgreSQL数据库 DATABASE_URL。如果未设置变量,它们将跳过。 使用Make目标确保本地运行与CI匹配,并通过Docker Compose启动本地Postgres:
# start the database used by tests
make db-up
# wait for Postgres to be ready (optional, test target already does this)
make db-wait
# run the test suite against the configured database
make test
# run all CI checks (tests + Docker image build), with automatic DB lifecycle management
make ciJSON模式
要导出摄取有效负载的JSON模式,请执行以下操作:
from src.domain.payloads import workout_payload_schema
import json
print(json.dumps(workout_payload_schema(), indent=2))云运行部署(CI/CD)
GitHub Actions工作流在推送时部署到Cloud Run main.
必需的GitHub机密
GCP_WORKLOAD_IDENTITY_PROVIDERGCP_SERVICE_ACCOUNT_EMAILDB_PASSWORD
GCP设置检查表
- 创建一个名为的工件注册表仓库
workout-tracker-mcp在us-central1. - 创建一个名为的Cloud SQL Postgres实例
workout-tracker-postgres在us-central1. - 创建数据库
workout_tracker和用户postgres(或更新工作流变量)。 - 为GitHub操作配置工作负载身份联合并授予服务帐户:
- roles/run.admin - roles/iam.serviceAccountUser - roles/artifactregistry.writer - roles/cloudsql.client
工作流文件是 /.github/workflows/deploy-cloudrun.yml.
