TODO MCP服务
一个独立的基于SQLite的人工智能代理任务管理服务,具有变更历史跟踪、项目支持和MCP(模型上下文协议)API。
特性
- 任务管理:创建、更新、跟踪任务类型(具体、抽象、史诗)
- 项目支持:使用源URL和本地路径按项目组织任务
- 变更历史:具有代理身份跟踪的完整审计跟踪
- 代理业绩:每个代理的统计数据和成功率跟踪
- 备份和恢复:采用gzip压缩的自动夜间备份
- 速率限制:滑动窗口速率限制,具有全局、每个端点和每个代理限制
- MCP API:用于LLM代理框架的最小5功能API
- REST API:通过FastAPI进行完整的CRUD操作
快速开始
使用Docker Compose
# Clone and navigate
git clone
cd todorama-mcp-service
# Set data directory (optional)
export TODO_DATA_DIR=/path/to/data
export TODO_SERVICE_PORT=8004
# Start the service
docker compose up -d
# Check status
docker compose ps
docker compose logs -f备注:此服务遵循 MCP服务集装箱化标准Dockerfile使用UV进行依赖管理,并遵循安全性和可重复性的标准模式。
外部网络集成
To Do Rama创建了一个Docker网络(todo-network)其可由其他服务用于跨容器通信。这使得不同Docker Compose项目中的服务能够进行通信。
网络名称: todo-network
其他服务可以连接到此网络 通过将其声明为外部网络 docker-compose.yml:
networks:
todo-network:
external: true
name: todo-network然后在他们的服务中引用它:
services:
my-service:
networks:
- todo-network优点:
- ✅ 服务可以使用容器名称进行通信(例如。,
http://todo-mcp-service:8004) - ✅ 即使To Do Rama停止,网络也会持续存在(如果其他服务引用它)
- ✅ 支持独立部署和扩展服务
- ✅ 不需要一个单一的docker组成文件
备注:To Do Rama启动时,网络会自动创建。其他服务必须在To Do Rama之后启动,以确保网络存在。
直接使用Docker
# Build the image
docker build -t todo-mcp-service:latest -f src/Dockerfile .
# Run the service
docker run -d \
--name todo-mcp-service \
-p 8004:8004 \
-v /path/to/data:/app/data \
-v /path/to/backups:/app/backups \
-e TODO_DB_PATH=/app/data/todos.db \
-e TODO_BACKUPS_DIR=/app/backups \
todo-mcp-service:latest备注:Dockerfile使用 uv export 模式(标准的替代方案 uv sync --frozen 图案)。这两种模式都是有效的,并记录在 集装箱化标准.
地方发展
对于没有Docker的本地开发:
先决条件:
- Python 3.11+
- 紫外线 (用于依赖关系管理)- 安装UV
备注:UV是所有MCP服务(TODO服务、Bucket-O-Facts、Doc-O-Matic)的标准依赖管理器,用于一致性和性能。
设置:
# Clone the repository
git clone
cd todorama-mcp-service
# Install dependencies using UV
uv sync
# Activate virtual environment (optional)
source .venv/bin/activate
# Or use uv run for commands (no activation needed)
uv run python -m src.main
# Run tests
uv run pytest
# Run with coverage
uv run pytest --cov=src --cov-report=html常见UV命令:
uv sync-从安装依赖项pyproject.toml和uv.lockuv run-在虚拟环境中运行命令- `uv add
` -添加新的依赖项
- `uv remove
` -删除依赖项
uv lock-更新锁文件
有关UV的更多信息,请参见 UV文件.
CI/CD管道
该项目包括一个用GitHub Actions实现的全面的CI/CD管道。
特性
- 自动化测试:使用pytest运行单元测试,并强制执行80%的代码覆盖率
- 代码质量:检查格式(黑色)、导入排序(isort)、linting(flake8、pylint)和类型检查(mypy)
- 安全扫描:依赖性漏洞检查(安全)、代码安全分析(Bandit)和容器扫描(Trivy)
- Docker构建:自动构建Docker镜像并推送到GitHub容器注册表
- 自动化部署:在上进行临时部署
develop分支,生产部署main分支 - 回滚功能:部署失败时自动回滚
工作流文件
.github/workflows/ci-cd.yml-主CI/CD管道.github/workflows/deploy.yml-手动部署工作流程
部署环境
- 暂存:
docker-compose.staging.yml-预生产测试 - 生产:
docker-compose.production.yml-现场制作服务
有关详细的部署说明,请参阅 部署 下面的部分。
预提交检查
在提交代码之前,运行全面的检查脚本:
./run_checks.sh这运行:
- 依赖性验证
- 数据库架构验证
- 代码质量检查
- 单元测试
- 备份功能测试
- 服务启动验证
项目支持
任务被组织成项目。每个项目都有:
- 名字:唯一项目标识符
- 本地路径:项目代码所在的绝对路径
- 来源网址:源位置(GitHub URL、文件://路径等)
- 描述:可选项目描述
创建项目
curl -X POST http://localhost:8004/projects \
-H "Content-Type: application/json" \
-d '{
"name": "june-agent",
"local_path": "/home/rlee/dev/june",
"origin_url": "https://github.com/your-org/june",
"description": "June Agent project"
}'处理项目任务
# Create task in a project
curl -X POST http://localhost:8004/tasks \
-H "Content-Type: application/json" \
-d '{
"title": "Implement feature X",
"task_type": "concrete",
"task_instruction": "Implement the feature",
"verification_instruction": "Run tests and verify",
"agent_id": "agent-123",
"project_id": 1
}'
# Query tasks by project
curl "http://localhost:8004/tasks?project_id=1"MCP功能
- list_available_tasks -获取代理类型的任务(使用可选的项目筛选器)
- reserve_task -为代理锁定任务
- 完成任务 -完成可选跟进
- 创建任务 -创建新任务(可选地在项目中)
- 获取代理性能 -获取代理统计信息
CLI工具
TODO MCP服务包括一个命令行界面(CLI),用于管理来自终端的任务。
安装
CLI可用作 src/cli.py 并且可以直接使用:
# Direct execution
python3 src/cli.py --help
# Or make it executable and run directly
chmod +x src/cli.py
./src/cli.py --help配置
使用环境变量或命令行选项配置CLI:
# Environment variables
export TODO_SERVICE_URL=http://localhost:8004
export TODO_API_KEY=your-api-key-here
# Or use command-line options
python3 src/cli.py --url http://localhost:8004 --api-key your-key listCLI命令
列出任务
# List all tasks
python3 src/cli.py list
# Filter by status
python3 src/cli.py list --status in_progress
# Filter by type and project
python3 src/cli.py list --type concrete --project-id 1
# Output as JSON
python3 src/cli.py list --format json创建任务
python3 src/cli.py create \
--title "Implement feature X" \
--type concrete \
--instruction "Implement the feature" \
--verification "Run tests and verify" \
--agent-id "my-agent" \
--project-id 1 \
--priority high \
--notes "Additional notes"显示任务详细信息
python3 src/cli.py show --task-id 123
# JSON output
python3 src/cli.py show --task-id 123 --format json完成任务
python3 src/cli.py complete \
--task-id 123 \
--agent-id "my-agent" \
--notes "Completed successfully" \
--actual-hours 4.5保留/解锁任务
# Reserve a task
python3 src/cli.py reserve --task-id 123 --agent-id "my-agent"
# Unlock a task
python3 src/cli.py unlock --task-id 123 --agent-id "my-agent"筛选选项
这 list 命令支持各种过滤器:
--status:按任务状态筛选(可用、正在进行、已完成、已阻止、已取消)--type:按任务类型筛选(具体、抽象、史诗)--project-id:按项目ID筛选--agent:按分配的代理筛选--priority:按优先级筛选(低、中、高、严重)--limit:最大结果数(默认值:100)
示例
# List all available concrete tasks in project 1
python3 src/cli.py list --status available --type concrete --project-id 1
# Create a high-priority task
python3 src/cli.py create \
--title "Fix critical bug" \
--type concrete \
--instruction "Fix the bug in module X" \
--verification "Verify bug is fixed and tests pass" \
--agent-id "dev-agent" \
--project-id 1 \
--priority critical
# Reserve and complete a task
python3 src/cli.py reserve --task-id 456 --agent-id "dev-agent"
python3 src/cli.py complete --task-id 456 --agent-id "dev-agent" --notes "Done!"
# View task details
python3 src/cli.py show --task-id 456开发命令
该服务提供了几个开发实用程序命令,可通过以下方式访问 python -m todorama :
分析命令
分析cursor-agent.log文件以了解代理行为:
# Analyze log file (default: cursor-agent.log)
python -m todorama analyze cursor-agent.log
# Show detailed analysis for a specific task
python -m todorama analyze cursor-agent.log --task 123
# List all tasks found in log
python -m todorama analyze cursor-agent.log --list-tasks特征:
- 任务活动跟踪(预订、完成、解锁、更新)
- 工具调用细分和频率
- 循环检测(识别快速重复更新)
- 每个任务的事件时间表
- 成功/失败率分析
审计指挥部
审核MCP工具参数类型,以验证它们是否与实现相匹配:
# Audit MCP function parameter types
python -m todorama audit
# Specify custom paths
python -m todorama audit --mcp-api todorama/mcp_api.py --main todorama/main.py特征:
- 将MCP_FUNCTIONS定义与MCPTodoAPI方法签名进行比较
- 验证参数类型、可选标志和默认值
- 检查main.py中的路由处理程序
- 识别类型不匹配和缺少处理程序
验证命令
验证所有MCP_FUNCTIONS是否都有相应的处理程序:
# Verify MCP function routing
python -m todorama verify
# Specify custom paths
python -m todorama verify --functions todorama/mcp/functions.py --request-handlers todorama/mcp/request_handlers.py特征:
- 列出MCP_functions中定义的所有函数
- 列出request_handles.py中的所有处理程序
- 识别缺失的处理程序(没有路由的函数)
- 标识额外的处理程序(没有函数定义的路由)
公用事业
游标代理日志解析器
这 parse_cursor_agent.py 该实用程序从Cursor Agent JSON输出中提取人类可读的内容。
用途:
# Pipe cursor agent output through the parser
cursor-agent command | tee cursor-agent.log | python3 parse_cursor_agent.py
# Or just parse an existing log file
cat cursor-agent.log | python3 parse_cursor_agent.py
# Parse and save to a readable log
cursor-agent command | tee cursor-agent.log | python3 parse_cursor_agent.py > readable.log它提取了什么:
- 任务操作:任务预留(🔒), 完工情况(✅), 和创造(➕)
- 工具调用:显示代理正在调用哪些工具(文件读取、终端命令等)
- 任务摘要:从任务执行结果中提取摘要
- 消息:显示用户和助手消息
- 错误:突出显示错误消息
- 持续时间:显示长时间运行的操作的执行时间
输出格式:
→工具调用📝内容/消息❌错误⚠️警告/状态🔧函数📋纯文本日志行
示例:
# Watch agent work in real-time
cursor-agent fix-tests | tee cursor-agent.log | python3 parse_cursor_agent.py
# Parse existing log file
cat cursor-agent.log | python3 parse_cursor_agent.py
# Or use stdin redirection
python3 parse_cursor_agent.py " \
-d '{
"name": "My API Key",
"organization_id": 1
}'答复:
{
"key_id": 1,
"project_id": 1,
"name": "My API Key",
"key_prefix": "todo_abc",
"api_key": "todo_abc123...",
"enabled": true,
"created_at": "2025-11-01T12:00:00"
}重要提示: 完整的API密钥仅在创建时返回。立即安全地存储它,以后无法检索。
管理API密钥
# List API keys for a project
curl -X GET http://localhost:8004/api/projects/{project_id}/api-keys \
-H "X-API-Key: "
# Rotate an API key (generates new key, revokes old)
curl -X POST http://localhost:8004/api/api-keys/{key_id}/rotate \
-H "X-API-Key: "
# Revoke an API key
curl -X DELETE http://localhost:8004/api/api-keys/{key_id} \
-H "X-API-Key: "密钥管理
该服务包括一个密钥管理命令实用程序:
- 发布管理员密钥:
python3 -m todorama key-management issue --admin --save创建具有跨项目访问权限的管理员API密钥,并可选择将其保存到 api_keys.json.
- 发布项目密钥:
# Issue key for a specific project
python3 -m todorama key-management issue --project 1 --save
# Issue keys for all projects
python3 -m todorama key-management issue --project-all --save创建项目范围的API键。使用 --project-all 为所有项目创建密钥。
- 无效(撤销)密钥:
# Invalidate by key ID
python3 -m todorama key-management invalidate --key-id 123
# Invalidate by key prefix
python3 -m todorama key-management invalidate --key-prefix "admin_ab"
# Invalidate all keys for a project (requires --confirm)
python3 -m todorama key-management invalidate --project 1 --confirm
# Invalidate all keys (requires --confirm)
python3 -m todorama key-management invalidate --all --confirm- 列出密钥:
# Table format (default)
python3 -m todorama key-management list
# JSON format
python3 -m todorama key-management list --format json
# List keys for a specific project
python3 -m todorama key-management list --project 1
# List only admin keys
python3 -m todorama key-management list --admin-only- 保存密钥:
# Save current keys to api_keys.json
python3 -m todorama key-management save- 使用生成的密钥:
# Load keys from JSON file
python3 -c "
import json
with open('api_keys.json') as f:
keys = json.load(f)
print(f\"Admin key: {keys['admin_key']}\")
print(f\"Project 1 key: {keys['project_keys'][1]['api_key']}\")
"重要提示: 这 api_keys.json 文件会自动添加到 .gitignore 并且永远不应该致力于版本控制。
制作API密钥管理员(手册)
若要手动向现有API密钥授予管理员权限,请在 api_key_admin 表:
INSERT INTO api_key_admin (api_key_id) VALUES ();管理员密钥可以:
- 为任何项目创建任务(绕过项目范围的限制)
- 访问所有项目,无论是否分配了密钥
project_id - 执行系统级操作
常见授权错误
- 401未经授权:缺少API密钥或密钥无效
- 403禁止:API密钥未被授权用于请求的项目(用于非管理员密钥)
- 401未经授权:API密钥已被吊销(
enabled = 0)
最佳实践
- 谨慎使用管理密钥:仅向需要跨项目访问的密钥授予管理员权限
- 项目特定密钥:尽可能为每个项目创建单独的API密钥
- 关键点旋转:为了安全起见,定期轮换API密钥
- 安全存储:从不将API密钥提交到版本控制
- 环境变量:将API密钥存储在环境变量或安全机密管理系统中
示例:使用项目范围键创建任务
# This works: API key is for project 1, task is for project 1
curl -X POST http://localhost:8004/api/Task/create \
-H "X-API-Key: project1-key" \
-H "Content-Type: application/json" \
-d '{
"title": "Task in project 1",
"task_type": "concrete",
"task_instruction": "Do something",
"verification_instruction": "Verify it",
"project_id": 1
}'
# This fails: API key is for project 1, task is for project 2
curl -X POST http://localhost:8004/api/Task/create \
-H "X-API-Key: project1-key" \
-H "Content-Type: application/json" \
-d '{
"title": "Task in project 2",
"task_type": "concrete",
"task_instruction": "Do something",
"verification_instruction": "Verify it",
"project_id": 2
}'
# Response: 403 Forbidden - API key is not authorized for this project
# This works: Admin key can create tasks for any project
curl -X POST http://localhost:8004/api/Task/create \
-H "X-API-Key: admin-key" \
-H "Content-Type: application/json" \
-d '{
"title": "Task in project 2",
"task_type": "concrete",
"task_instruction": "Do something",
"verification_instruction": "Verify it",
"project_id": 2
}'有关API的综合文档,请参阅 API毫米 有关详细的端点规范、请求/响应格式和代码示例。
配置
TODO MCP服务使用 Pydantic设置 用于类型安全配置管理,支持 .env 文件和环境变量重写。这种标准化的方法确保了所有MCP服务的一致性。
标准化配置模式
该服务使用Pydantic的 BaseSettings 具有以下特点:
.env文件支持:创建一个.env项目根目录中的文件用于本地开发- 环境变量覆盖:所有设置都可以通过环境变量覆盖
- 类型验证:自动验证配置值,并显示有用的错误消息
- 默认值:提供合理的默认值以方便开发
- 不区分大小写:环境变量名称不区分大小写
配置可通过以下方式访问 get_settings() 返回缓存的单例实例的函数:
from todorama.config import get_settings
settings = get_settings()
db_path = settings.database_path
log_level = settings.log_level配置选项
标准化数据库配置
database_path/TODO_DB_PATH(字符串,必填)
- SQLite的数据库文件路径 - 默认值:根据环境自动解析 - 地方发展: data/todos.db (相对于项目根) - 集装箱: /app/data/todos.db - 环境变量: TODO_DB_PATH (向后兼容性)或 DATABASE_PATH - 路径解析:服务会自动检测它是否在容器中运行,并相应地调整默认路径。这 TODO_DB_PATH 环境变量具有最高优先级。
db_pool_size(整数,默认值:5)
- 连接池大小(用于未来的PostgreSQL支持) - 环境变量: DB_POOL_SIZE
db_max_overflow(整数,默认值:10)
- 最大溢出连接数 - 环境变量: DB_MAX_OVERFLOW
db_pool_timeout(整数,默认值:30)
- 连接超时(秒) - 环境变量: DB_POOL_TIMEOUT
sql_echo(布尔值,默认值:false)
- 启用SQL查询日志记录以进行调试 - 环境变量: SQL_ECHO (设置为 "true" 启用)
标准化日志配置
log_level(字符串,默认值:"INFO")
- 日志记录级别: DEBUG, INFO, WARNING, ERROR, CRITICAL - 环境变量: LOG_LEVEL
log_format(字符串,默认值:"json")
- 日志记录格式: "json" 或 "text" - 环境变量: LOG_FORMAT
标准化环境配置
environment(字符串,默认值:"development")
- 环境名称: "development", "staging", "production" - 环境变量: ENVIRONMENT
debug(布尔值,默认值:false)
- 启用调试模式 - 环境变量: DEBUG (设置为 "true" 启用)
服务特定配置
TODO_BACKUPS_DIR(字符串,默认值:/app/backups)
- 备份目录路径 - 环境变量: TODO_BACKUPS_DIR
TODO_SERVICE_PORT(整数,默认值:8004)
- HTTP服务端口 - 环境变量: TODO_SERVICE_PORT
TODO_BACKUP_INTERVAL_HOURS(整数,默认值:24)
- 备份间隔(小时) - 环境变量: TODO_BACKUP_INTERVAL_HOURS
配置示例
示例 .env 地方发展档案
创建一个 .env 项目根目录中的文件:
# Database Configuration
TODO_DB_PATH=data/todos.db
DB_POOL_SIZE=5
DB_MAX_OVERFLOW=10
DB_POOL_TIMEOUT=30
SQL_ECHO=false
# Logging Configuration
LOG_LEVEL=INFO
LOG_FORMAT=json
# Environment Configuration
ENVIRONMENT=development
DEBUG=false
# Service-Specific Configuration
TODO_BACKUPS_DIR=./backups
TODO_SERVICE_PORT=8004
TODO_BACKUP_INTERVAL_HOURS=24示例 .env 容器化部署文件
# Database Configuration
TODO_DB_PATH=/app/data/todos.db
DB_POOL_SIZE=10
DB_MAX_OVERFLOW=20
DB_POOL_TIMEOUT=30
SQL_ECHO=false
# Logging Configuration
LOG_LEVEL=INFO
LOG_FORMAT=json
# Environment Configuration
ENVIRONMENT=production
DEBUG=false
# Service-Specific Configuration
TODO_BACKUPS_DIR=/app/backups
TODO_SERVICE_PORT=8004
TODO_BACKUP_INTERVAL_HOURS=24环境变量覆盖示例
在不修改的情况下覆盖特定设置 .env 文件:
# Override database path
export TODO_DB_PATH=/custom/path/todos.db
# Enable SQL query logging for debugging
export SQL_ECHO=true
# Change log level
export LOG_LEVEL=DEBUG
# Enable debug mode
export DEBUG=true不同环境的示例
发展:
ENVIRONMENT=development
DEBUG=true
LOG_LEVEL=DEBUG
SQL_ECHO=true
TODO_DB_PATH=./data/todos.db分期:
ENVIRONMENT=staging
DEBUG=false
LOG_LEVEL=INFO
SQL_ECHO=false
TODO_DB_PATH=/app/data/todos.db生产:
ENVIRONMENT=production
DEBUG=false
LOG_LEVEL=WARNING
SQL_ECHO=false
TODO_DB_PATH=/app/data/todos.db数据库配置
SQLite配置(默认)
该服务默认使用SQLite进行本地开发,可以配置为容器化部署:
- 地方发展:数据库文件创建于
data/todos.db(相对于项目根) - 容器:数据库文件创建于
/app/data/todos.db(应从卷中安装) - 路径解析:服务会自动检测环境并使用适当的默认值
- 向后兼容:The
TODO_DB_PATH环境变量具有最高优先级
PostgreSQL配置(未来)
该服务旨在未来支持PostgreSQL。已配置连接池设置:
- 连接池:通过配置
db_pool_size,db_max_overflow,db_pool_timeout - SQL日志记录:通过启用
sql_echo=true用于调试
SQL查询日志记录
启用SQL查询日志记录以进行调试:
# In .env file
SQL_ECHO=true
# Or via environment variable
export SQL_ECHO=true这将把所有SQL查询记录到控制台,这对于调试数据库操作非常有用。
配置优先级
配置值按以下顺序(从高到低优先级)解析:
- 环境变量 (最高优先级)
.env文件 (如果存在)- 默认值 (最低优先级)
例如,如果您设置 TODO_DB_PATH 作为环境变量,它将覆盖 .env 文件。
类型验证和错误处理
Pydantic Settings会自动验证配置值:
- 类型检查:无效类型引发
ValidationError带有清晰的错误信息 - 必填字段:启动时检测到缺少必填字段
- 默认值:为可选字段提供了合理的默认值
错误消息示例:
ValidationError: 1 validation error for Settings
database_path
Field required [type=missing, input_value=None, input_type=NoneType]集装箱检测
该服务通过检查以下内容自动检测它是否在容器中运行:
- ······的出现
/.dockerenv文件 - 集装箱指示器
/proc/1/cgroup
在容器中运行时,默认数据库路径为 /app/data/todos.db。对于当地发展,它使用 data/todos.db 相对于项目根。
速率限制
该服务实现了滑动窗口速率限制,以防止滥用。执行三种类型的限制:
- 全球利率限额:适用于所有端点上的所有请求
- RATE_LIMIT_GLOBAL_MAX:最大请求数(默认值: 100) - RATE_LIMIT_GLOBAL_WINDOW:时间窗口(秒)(默认值: 60)
- 每个端点速率限制:每个端点的限制不同
- RATE_LIMIT_ENDPOINT_MAX:每个端点的最大请求数(默认值: 200) - RATE_LIMIT_ENDPOINT_WINDOW:时间窗口(秒)(默认值: 60) - RATE_LIMIT_ENDPOINT_OVERRIDES:以逗号分隔的端点特定覆盖列表 格式: ENDPOINT_PATH:max:window (例如。, /health:500:60,/mcp/sse:10:60)
- 每个代理的费率限制:每个代理ID的限制(从标头或查询参数中提取)
- RATE_LIMIT_AGENT_MAX:每个代理的最大请求数(默认值: 50) - RATE_LIMIT_AGENT_WINDOW:时间窗口(秒)(默认值: 60) - RATE_LIMIT_AGENT_OVERRIDES:以逗号分隔的代理特定替代列表 格式: AGENT_ID:max:window (例如。, agent-1:200:60,agent-2:100:60)
- 每用户速率限制:每个经过身份验证的用户的限制(从会话令牌中提取)-使用 令牌桶算法
- RATE_LIMIT_USER_MAX:存储桶容量/每个用户的最大请求数(默认值: 100) - RATE_LIMIT_USER_WINDOW:计算再填充率的时间窗口(秒)(默认值: 60) - 令牌桶 允许突发流量达到最大容量,并以稳定的速率(max_requests/window_seconds)重新填充令牌 - RATE_LIMIT_USER_OVERRIDES:逗号分隔的用户特定覆盖列表 格式: USER_ID:max:window (例如。, 123:200:60,456:150:60)
当超过速率限制时,服务返回HTTP 429(太多请求),其中包含:
Retry-After标头指示重试前的秒数X-RateLimit-Limit显示限制的标题X-RateLimit-Remaining显示剩余请求的标头X-RateLimit-Reset显示限额重置时间的标题
配置示例:
export RATE_LIMIT_GLOBAL_MAX=100
export RATE_LIMIT_GLOBAL_WINDOW=60
export RATE_LIMIT_ENDPOINT_OVERRIDES="/health:500:60,/mcp/sse:10:60"安全标头
该服务会自动向所有HTTP响应添加安全标头,以防止常见的web漏洞。所有标头都可以通过环境变量进行配置:
标准安全标头
- X-内容-类型选项:防止MIME类型嗅探
- SECURITY_HEADER_X_CONTENT_TYPE_OPTIONS (默认值: nosniff)
- X-Frame-Options:防止点击劫持攻击
- SECURITY_HEADER_X_FRAME_OPTIONS (默认值: DENY,选项: DENY, SAMEORIGIN)
- X-XSS-Protection:传统XSS保护(适用于较旧的浏览器)
- SECURITY_HEADER_X_XSS_PROTECTION (默认值: 1; mode=block)
- 推荐人政策:控制随请求发送的推荐人信息
- SECURITY_HEADER_REFERRER_POLICY (默认值: strict-origin-when-cross-origin)
- 内容安全策略:限制资源加载以防止XSS
- SECURITY_HEADER_CSP (默认:限制性策略,可自定义)
- 权限策略:限制浏览器功能和API
- SECURITY_HEADER_PERMISSIONS_POLICY (默认:限制权限)
- 跨源开放政策:隔离浏览上下文
- SECURITY_HEADER_CROSS_ORIGIN_OPENER_POLICY (默认值: same-origin)
- 跨来源资源政策:限制资源加载
- SECURITY_HEADER_CROSS_ORIGIN_RESOURCE_POLICY (默认值: same-origin)
- 跨源嵌入策略:需要CORP标头(可选,默认禁用)
- SECURITY_HEADER_COEP_ENABLED (默认值: false) - SECURITY_HEADER_CROSS_ORIGIN_EMBEDDER_POLICY (默认值: require-corp)
HSTS(HTTP严格传输安全)
HSTS仅在满足以下两个条件时设置:
SECURITY_HSTS_ENABLED=true- 请求通过HTTPS(或使用
X-Forwarded-Proto: https头球
配置:
SECURITY_HSTS_ENABLED:启用HSTS(默认值:false)SECURITY_HSTS_MAX_AGE:最大年龄(秒)(默认值:31536000=1年)SECURITY_HSTS_INCLUDE_SUBDOMAINS:包括子域(默认值:true)SECURITY_HSTS_PRELOAD:启用HSTS预加载(默认值:false)
配置示例:
# Enable HSTS for HTTPS deployments
export SECURITY_HSTS_ENABLED=true
export SECURITY_HSTS_MAX_AGE=31536000
export SECURITY_HSTS_INCLUDE_SUBDOMAINS=true
# Customize CSP policy
export SECURITY_HEADER_CSP="default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'self' 'unsafe-inline';"
# Allow same-origin framing
export SECURITY_HEADER_X_FRAME_OPTIONS=SAMEORIGIN备注:安全标头会自动应用于所有端点。基本操作不需要额外的配置。
存储库模式
TODO MCP服务使用 存储库模式 对于数据访问,在业务逻辑(服务)和数据访问(存储库)之间提供清晰的分离。此模式在所有MCP服务中都是标准化的,以实现一致性和可维护性。
标准化存储库模式
该服务使用存储库类来抽象核心实体(任务、项目、组织)的数据库操作。存储库包装 TodoDatabase 方法,并为服务与数据交互提供一个干净的接口。
存储库结构:
- 存储库位于
todorama/storage/repositories.py - 每种实体类型都有自己的存储库类(例如。,
TaskRepository,ProjectRepository,OrganizationRepository) - 仓库接收
TodoDatabase通过依赖注入实现实例
优点:
- 关注点分离:业务逻辑(服务)与数据访问(存储库)分离
- 可测试性:存储库可以很容易地模拟用于服务测试
- 可维护性:明确的结构和责任
- 一致性:所有MCP服务的标准接口
存储库接口
所有存储库都实现标准的CRUD操作:
核心方法:
create(**kwargs)-创建新实体,返回实体IDget_by_id(entity_id, organization_id=None)-按ID获取实体get_by_(field_value)-通过唯一字段获取实体(例如。,get_by_name())update(entity_id, **kwargs)-更新现有实体delete(entity_id)-删除实体list(**filters)-列出具有可选筛选器的实体search(query, **filters)-全文搜索
方法命名约定:
- 使用描述性方法名称:
get_by_id(),get_by_name(),list(),search() - 使用一致的参数名称:
entity_id,organization_id,limit,offset - 退货类型:
Optional[Dict[str, Any]]对于单个实体,List[Dict[str, Any]]对于列表
参数模式:
- ID参数:使用
entity_id对于主键 - 过滤器:使用具有描述性名称的关键字参数(例如。,
status,type,organization_id) - 分页:使用
limit(默认值:100)和offset(默认值:0) - 订购:使用
order_by参数(例如。,'created_at','priority') - 多租户技术:使用
organization_id租户隔离参数
服务存储库关系
服务通过依赖注入使用存储库:
from todorama.storage.repositories import TaskRepository, ProjectRepository
from todorama.database import TodoDatabase
# Initialize database
db = TodoDatabase()
# Initialize repositories
task_repository = TaskRepository(db)
project_repository = ProjectRepository(db)
# Services receive repositories via constructor
class TaskService:
def __init__(self, task_repository: TaskRepository):
self.task_repository = task_repository
def create_task(self, **kwargs):
# Business logic validation
if not kwargs.get('title'):
raise ValueError("Title is required")
# Delegate to repository
return self.task_repository.create(**kwargs)关注点分离:
- 仓库:处理数据库操作(查询、事务)
- 服务:处理业务逻辑(验证、编排、计算字段)
- 界限清楚服务从不直接访问数据库;存储库从不包含业务逻辑
示例
示例:TaskRepository使用情况
from todorama.storage.repositories import TaskRepository
from todorama.database import TodoDatabase
# Initialize
db = TodoDatabase()
task_repo = TaskRepository(db)
# Create a task
task_id = task_repo.create(
title="Implement feature X",
task_type="concrete",
task_instruction="Implement the feature",
verification_instruction="Run tests",
agent_id="agent-123",
project_id=1,
priority="high"
)
# Get task by ID
task = task_repo.get_by_id(task_id)
# List tasks with filters
tasks = task_repo.list(
task_status="available",
task_type="concrete",
project_id=1,
limit=50,
order_by="priority"
)
# Search tasks
results = task_repo.search("feature X", limit=20)示例:使用存储库模拟进行测试
from unittest.mock import Mock
from todorama.services.task_service import TaskService
def test_task_service():
# Create mock repository
mock_repo = Mock(spec=TaskRepository)
mock_repo.create.return_value = 123
mock_repo.get_by_id.return_value = {
'id': 123,
'title': 'Test Task',
'task_status': 'available'
}
# Initialize service with mock
service = TaskService(mock_repo)
# Test service methods
task_id = service.create_task(title="Test Task", ...)
assert task_id == 123
task = service.get_task(task_id)
assert task['title'] == 'Test Task'存储库代码
存储库实现位于:
- 文件:
todorama/storage/repositories.py - 类:
TaskRepository,ProjectRepository,OrganizationRepository
有关完整的存储库模板和最佳实践,请参阅 REPOSITORY_PATTERN.md.
数据库迁移
TODO MCP服务使用 蒸馏器 对于数据库模式迁移,为数据库更改提供版本控制和回滚功能。这种标准化方法用于所有MCP服务,以实现一致性和可维护性。
标准化迁移方法
该服务使用Alembic来管理数据库模式更改。所有模式修改都通过迁移脚本进行跟踪,从而启用:
- 版本控制:跟踪随时间推移的所有架构更改
- 回滚支持:如果需要,能够回滚迁移
- 团队协作:多个开发人员的标准工作流程
- 测试:在应用于生产之前,单独测试迁移
Alembic目录结构:
alembic/-Alembic迁移目录(位于项目根目录)alembic/versions/-迁移脚本文件alembic/env.py-Alembic环境配置alembic.ini-Alembic配置文件
迁移历史: 该服务已从ad-hoc迁移逻辑(嵌入在 database.py)Alembic。所有新的模式更改都应在Alembic迁移时创建。
迁移命令
应用迁移:
# Apply all pending migrations
alembic upgrade head
# Apply migrations up to a specific revision
alembic upgrade
# Apply one migration at a time
alembic upgrade +1回滚迁移:
# Rollback last migration
alembic downgrade -1
# Rollback to a specific revision
alembic downgrade
# Rollback all migrations (use with caution)
alembic downgrade base检查迁移状态:
# Show current migration version
alembic current
# Show migration history
alembic history
# Show detailed history with revisions
alembic history --verbose创建新迁移:
# Auto-generate migration from model changes
alembic revision --autogenerate -m "description of changes"
# Create empty migration script (manual)
alembic revision -m "description of changes"示例
本地开发(SQLite):
# Set database path
export TODO_DB_PATH=./data/todos.db
# Apply all migrations
alembic upgrade head
# Check current version
alembic current
# Create new migration after model changes
alembic revision --autogenerate -m "add new column to tasks"容器化部署(PostgreSQL):
# Set database connection
export DB_TYPE=postgresql
export DB_HOST=localhost
export DB_PORT=5432
export DB_NAME=todos
export DB_USER=postgres
export DB_PASSWORD=your_password
# Apply migrations in container
docker exec -it todo-mcp-service alembic upgrade head
# Or run migrations before starting service
alembic upgrade head
python -m todorama.server创建新迁移:
# 1. Make changes to models in todorama/models/ or database schema
# 2. Generate migration automatically
alembic revision --autogenerate -m "add priority column to tasks"
# 3. Review the generated migration in alembic/versions/
# 4. Edit if needed (add data migrations, custom logic)
# 5. Test the migration
alembic upgrade head
alembic downgrade -1 # Test rollback
alembic upgrade head # Re-apply测试迁移:
# Test migration on a copy of production database
cp production.db test.db
export TODO_DB_PATH=./test.db
alembic upgrade head
# Verify schema changes
sqlite3 test.db ".schema tasks"
# Test rollback
alembic downgrade -1
alembic upgrade head故障排除
常见迁移错误:
- “目标数据库不是最新的”
# Check current version
alembic current
# Apply pending migrations
alembic upgrade head- “找不到由'xyz'标识的修订版”
- 迁移历史记录不匹配-检查 alembic_version 桌子 - 可能需要手动设置版本: alembic stamp
- “检测到多个头”
- 存在多个迁移分支 - 合并分支: alembic merge -m "merge branches" heads
- 迁移与现有架构冲突
- 检查迁移脚本是否存在冲突 - 可能需要调整迁移或手动修复架构 - 使用 alembic revision --autogenerate 检测差异
正在检查迁移状态:
# Show current database version
alembic current
# Compare with latest migration
alembic heads
# Show pending migrations
alembic history | head -5回滚程序:
# Rollback last migration (safe)
alembic downgrade -1
# Rollback to specific version
alembic downgrade
# Verify rollback
alembic current
# Re-apply if needed
alembic upgrade head解决迁移冲突:
- 检查迁移历史记录:
alembic history - 识别冲突的迁移
- 在中查看迁移脚本
alembic/versions/ - 如果需要,合并分支:
alembic merge -m "merge" - 测试合并迁移:
alembic upgrade head
服务特定注意事项
SQLite和PostgreSQL支持:
- 迁移同时适用于SQLite(本地开发)和PostgreSQL(生产)
- PostgreSQL的一些特定功能(例如,某些索引类型)在迁移中可能需要条件逻辑
- 尽可能在两种数据库类型上测试迁移
多租户迁移:
- 与组织相关的迁移通过Alembic处理
- 看
alembic/versions/228b7c679817_add_organization_id_columns.py供参考
从Ad Hoc迁移到Alembic:
- 该服务以前在
database.py - 所有模式更改现在都通过Alembic迁移进行管理
- 历史迁移已转换为Alembic脚本
迁移最佳实践
- 始终查看自动生成的迁移
- alembic revision --autogenerate 是一个起点 - 在应用之前查看和编辑迁移脚本 - 如果架构更改需要数据转换,请添加数据迁移
- 生产前的测试迁移
- 在生产数据库副本上进行测试 - 测试升级和降级路径 - 迁移后验证数据完整性
- 使用描述性迁移消息
- 清晰、描述性的信息: "add priority column to tasks" - 避免模糊的信息: "update schema"
- 保持迁移规模小、重点突出
- 尽可能每次迁移进行一次逻辑更改 - 更易于查看、测试和回滚
- 记录复杂的迁移
- 在迁移脚本中为复杂逻辑添加注释 - 文档数据转换 - 注意所需的任何手动步骤
有关更多信息,请参阅 Alembic文件.
生产部署
资源限制
该服务包括资源限制 docker-compose.yml 为了防止资源枯竭:
# Set resource limits via environment variables
export TODO_SERVICE_CPU_LIMIT=2.0
export TODO_SERVICE_MEMORY_LIMIT=512M
export TODO_SERVICE_CPU_RESERVATION=0.5
export TODO_SERVICE_MEMORY_RESERVATION=256M
export POSTGRES_CPU_LIMIT=1.0
export POSTGRES_MEMORY_LIMIT=512M
export POSTGRES_CPU_RESERVATION=0.25
export POSTGRES_MEMORY_RESERVATION=128M默认值是保守的,但可以根据工作量进行调整。
优雅地关闭
该服务实现了优雅的关机处理:
- 信号处理:响应信号和信号情报信号
- FastAPI生命周期事件:正确关闭后台任务(备份调度程序)
- Uvicorn配置:优雅关机超时30秒
- 清洁资源清理:确保关闭备份调度程序和连接
部署到生产环境时:
- 确保编排器(Docker、Kubernetes)在强制终止之前发送SIGTERM
- 至少需要30秒才能正常关机
- 监控“应用程序正在关闭…”的日志,以确认完全关闭
用于生产的Webhook配置
该服务支持用于实时事件通知的webhooks。在生产环境中,正确配置webhooks:
Webhook设置
- 创建webhook端点 (在您的应用程序/服务中):
curl -X POST http://localhost:8004/webhooks \
-H "Content-Type: application/json" \
-d '{
"url": "https://your-service.com/webhooks/todo-events",
"events": ["task.created", "task.completed", "task.status_changed"],
"secret": "your-webhook-secret-here",
"enabled": true,
"retry_count": 3,
"timeout_seconds": 10
}'- Telegram机器人服务 (使用webhooks而不是轮询):
在中设置环境变量 docker-compose.yml:
environment:
- TELEGRAM_WEBHOOK_URL=https://your-domain.com/telegram/webhook
- TELEGRAM_WEBHOOK_SECRET=your-secret-key
- TELEGRAM_USE_POLLING=false # Set to true for development生产与发展:
- 生产:使用webhooks(
TELEGRAM_USE_POLLING=false)-更高效、实时、可扩展 - 发展:使用轮询(
TELEGRAM_USE_POLLING=true)-设置更简单,不需要公共URL
Webhook安全
- HMAC签名:使用
secret字段设置一个秘密。服务发送X-Webhook-Signature使用HMAC-SHA256作为标头。 - 仅限HTTPS:在生产环境中,始终对webhook URL使用HTTPS。
- 秘密轮换:定期轮换webhook机密。
- 验证:验证webhook接收器上的签名:
import hmac
import hashlib
def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
expected = hmac.new(
secret.encode(),
payload,
hashlib.sha256
).hexdigest()
return hmac.compare_digest(f"sha256={expected}", signature)Webhook检索
失败时重试Webhook(默认值:重试3次,超时10秒)。失败的Webhook会被记录,但不会阻止任务操作。
生产检查表
- \[\]中配置的资源限制
docker-compose.yml - \[\]为生产设置的环境变量(数据库、端口、日志记录)
- \[\]配置了HTTPS URL和机密的Webhooks
- \[\]根据预期负载调整速率限制
- \[\]已计划和测试的备份
- \[\]配置并监控健康检查
- \[\]适当设置日志级别(INFO用于生产,DEBUG用于故障排除)
- \[\]优雅关机测试(发送SIGTERM,验证干净关机)
监控
该服务在以下位置公开Prometheus指标 /metrics:
curl http://localhost:8004/metrics关键指标:
http_requests_total:按端点/状态列出的请求总数http_request_duration_seconds:请求延迟直方图http_errors_total:按类型列出的错误计数service_uptime_seconds:服务正常运行时间
健康终点: http://localhost:8004/health
部署
CI/CD管道
该项目包括一个用GitHub Actions实现的全面的CI/CD管道:
- 自动化测试:使用pytest运行单元测试,并强制执行80%的代码覆盖率
- 代码质量:检查格式(黑色)、导入排序(isort)、linting(flake8、pylint)和类型检查(mypy)
- 安全扫描:依赖性漏洞检查(安全)、代码安全分析(Bandit)和容器扫描(Trivy)
- Docker构建:自动构建Docker镜像并推送到GitHub容器注册表
- 自动化部署:在上进行临时部署
develop分支,生产部署main分支 - 回滚功能:部署失败时自动回滚
部署环境
- 暂存:
docker-compose.staging.yml-预生产测试(端口8005) - 生产:
docker-compose.production.yml-现场生产服务(端口8004)
手动部署
使用Docker Compose:
# Staging
export STAGING_PORT=8005
export STAGING_DATA_DIR=./data/staging
docker-compose -f docker-compose.staging.yml up -d
# Production
export PRODUCTION_PORT=8004
export PRODUCTION_DATA_DIR=./data/production
export POSTGRES_PASSWORD=your-secure-password
docker-compose -f docker-compose.production.yml up -d使用GitHub操作:
- 转到操作→ 部署工作流
- 点击“运行工作流”
- 选择环境(暂存/生产)
- 可选地指定图像标签(默认:最新)
- 点击“运行工作流”
回滚程序
自动回滚:
- 部署失败时,管道会自动回滚
- 检测部署健康检查失败
- 恢复以前的容器版本
手动回滚:
# Rollback using Docker Compose
docker-compose -f docker-compose.production.yml down
docker-compose -f docker-compose.production.yml up -d
# Or pull previous image tag
docker pull ghcr.io/your-repo/todo-mcp-service:previous-tag
# Update docker-compose.yml with previous tag, then redeploy演出
查询性能目标
| 操作 | 预期性能 | 注意事项 |
|---|---|---|
query_tasks() (简单滤波器) | \ |
SLACK_SIGNING_SECRET= SLACK_DEFAULT_CHANNEL=#general
### 用法
**Slash命令:**
- `/todo list` -列出可用任务(最多10个)
- `/todo reserve ` -保留任务
- `/todo complete [notes]` -完成任务
- `/todo help` -显示可用命令
**安全:**
- 所有Slack请求都使用HMAC-SHA256签名进行验证
- 时间戳验证可防止重放攻击
- 机器人令牌和签名秘密应保持安全
有关详细的设置说明,请参阅 [SLAK_SETUP.md](SLACK_SETUP.md).
## 音频转换器
该服务包括一个用于Telegram语音消息的音频转换器实用程序。
### 特性
- 将PCM/WAV音频文件转换为OGG/OPUS格式(Telegram的首选格式)
- 处理Telegram的约1分钟持续时间限制(自动截断较长的音频)
- 优化音频质量和文件大小
- 支持压缩较小的文件大小
### 需求
**系统依赖性:**
- `ffmpeg` 必须安装在系统上
**安装:**
Ubuntu/Debian
sudo apt-get install ffmpeg
macOS
brew install ffmpeg
### 用法
from audio_converter import TelegramAudioConverter
converter = TelegramAudioConverter()
Convert WAV to OGG/OPUS for Telegram
converter.convert_for_telegram( input_path="input.wav", output_path="output.ogg", compress=True # Optional compression )
### 电报约束
- **最长持续时间:** 约60秒(自动执行)
- **推荐比特率:** 64 kbps
- **最大文件大小:** 20 MB
- **格式:** OGG/OPUS
- **采样率:** 48000 Hz(自动设置)
- **频道:** 单声道(自动设置)
音频转换器位于 `todorama/services/audio_converter.py`.
## MCP功能指南
TODO服务公开了用于代理交互的MCP函数。看 [MCP_功能\_ GIDE.md](MCP_FUNCTION_GUIDE.md) 了解如何根据您的需求选择合适的MCP功能的全面指南。
### 核心任务工作流功能
- **`list_available_tasks()`** -为您的代理类型准备任务
- **`reserve_task()`** -工作前锁定任务(强制)
- **`complete_task()`** -标记任务完成(完成后必须)
- **`create_task()`** -使用自动关系链接创建新任务
- **`get_task_context()`** -获取完整的任务详细信息,包括项目、血统、更新
- **`add_task_update()`** -添加进度更新、发现、阻止程序或问题
- **`query_tasks()`** -按状态、类型、代理、优先级、标签灵活搜索任务
- **`search_tasks()`** -基于关键字的任务标题和说明搜索
### 功能选择决策树
- **我需要找个任务来做**:使用 `list_available_tasks()` 或 `query_tasks()`
- **我需要任务信息**:使用 `get_task_context()` 有关完整详细信息
- **我需要统计数据**:使用 `get_agent_performance()` 或 `get_project_statistics()`
- **我需要组织任务**:使用 `create_task()` 与关系,或 `add_task_tags()`
- **我需要沟通**:使用 `add_task_update()` 与适当 `update_type`
有关完整的功能参考和最佳实践,请参阅 [MCP_功能\_ GIDE.md](MCP_FUNCTION_GUIDE.md).
## 文档
- **[代理商.md](./AGENTS.md)** -代理商指南和开发实践
- **[API毫米](./API.md)** -全面的API文件
- **[MCP_功能\_ GIDE.md](./MCP_FUNCTION_GUIDE.md)** -MCP功能选择指南
- **[性能.md](./PERFORMANCE.md)** -数据库性能优化指南
- **[SLAK_SETUP.md](./SLACK_SETUP.md)** -Slack集成设置指南
## 许可证
MIT许可证