TimeLooker MCP Server with x402 Payment Integration

这是一个提供自动化搜索监控功能并集成x402支付的Model Context Protocol (MCP)服务器。监控搜索查询，使用AI驱动的重复检测来检测新内容，并自动处理任务创建的支付。

架构概述

系统由多个协同工作的组件组成：

核心组件

• src/api/task_manager_api.py - 集成x402支付中间件的FastAPI服务
• src/api/task_manager_client.py - 支持x402支付的HTTP客户端
• src/mcp/searcher_mcp.py - 向Claude Desktop暴露工具的MCP服务器
• src/core/search_engine.py - 网络搜索执行和AI比较
• src/core/task_manager.py - 数据库操作和任务管理
• src/core/models.py - SQLAlchemy数据库模型

支付集成

• x402协议 - 自动支付任务创建（每个任务1.00美元）
• 支付地址 - 0x671cE47E4F38051ba3A990Ba306E2885C2Fe4102
• 网络 - base-sepolia
• 免费端点 - 除了任务创建的所有操作

设置说明

选项A：本地开发设置

1. 安装依赖

cd /path/to/MCP_Server
uv sync

2. 环境配置

复制示例环境文件并进行配置：

cp .env.example .env

编辑.env以添加您的配置：

# x402支付配置
PAY_TO_ADDRESS=0x671cE47E4F38051ba3A990Ba306E2885C2Fe4102
PRIVATE_KEY=your_ethereum_private_key_here

# API配置
TASK_MANAGER_API_URL=http://localhost:8000

# OpenAI配置
OPENAI_API_KEY=your_openai_api_key_here

# 数据库配置（可选）
DATABASE_URL=sqlite:///timelooker.db

选项B：AWS云部署

1. 前提条件

• 已配置具有适当权限的AWS CLI
• 已安装AWS CDK：npm install -g aws-cdk
• Python依赖项：uv sync

2. 部署基础设施

# 部署AWS基础设施（RDS、Lambda角色、SES等）
python scripts/deploy_infrastructure.py

这会创建：

• PostgreSQL RDS数据库（db.t3.micro）
• 用于电子邮件模板的S3桶
• AWS Secrets Manager中的API密钥
• Lambda执行的IAM角色
• SES电子邮件身份

3. 配置Secrets

部署后，更新AWS Secrets Manager中的密钥：

# 更新OpenAI API密钥
aws secretsmanager update-secret \
  --secret-id "timelooker/openai/api-key" \
  --secret-string '{"api_key":"your_openai_key_here"}'

# 更新X402私钥
aws secretsmanager update-secret \
  --secret-id "timelooker/x402/private-key" \
  --secret-string '{"private_key":"your_private_key_here"}'

4. 验证SES电子邮件

前往AWS控制台 > SES > 验证的身份并验证您的发件人电子邮件地址。

5. 环境配置

部署脚本创建了包含基础设施详情的.env.aws。用您的值更新它：

cp .env.aws .env
# 编辑.env以添加您的私钥和发件人电子邮件

6. 测试Secrets集成

验证系统能否从AWS Secrets Manager中检索密钥：

# 测试密钥检索
python scripts/test_secrets.py

这将显示系统是否能自动检索数据库凭据、API密钥和其他密钥。

3. 数据库管理

初始化数据库

# 基础初始化
python scripts/init_db.py

# 使用示例数据初始化
python scripts/init_db.py --sample

数据库操作

# 检查模式版本
python scripts/init_db.py --version

# 验证数据库完整性
python scripts/init_db.py --validate

# 运行数据库迁移
python scripts/init_db.py --migrate

# 重置数据库（删除并重新创建）
python scripts/init_db.py --reset

# 重置并创建示例数据
python scripts/init_db.py --reset --sample

4. 启动API服务器

python run_api_server.py
# 或
uv run run_api_server.py

5. 配置Claude Desktop

编辑您的Claude Desktop配置文件：

• macOS：~/Library/Application Support/Claude/claude_desktop_config.json
• Windows：%APPDATA%\Claude\claude_desktop_config.json

根据您的部署模式选择适当的配置：

对于本地开发：

cp claude_desktop_config_local.json ~/Library/Application\ Support/Claude/claude_desktop_config.json

对于AWS云部署：

cp claude_desktop_config_cloud.json ~/Library/Application\ Support/Claude/claude_desktop_config.json
# 然后编辑文件，用实际部署值替换YOUR_ACCOUNT和端点值

手动配置：

您也可以手动复制claude_config_example.json的内容并添加您的值。

重要：将/absolute/path/to/MCP_Server替换为您实际的绝对路径！

6. 重启Claude Desktop

在编辑配置后完全重启Claude Desktop。

MCP可用工具

MCP服务器为Claude Desktop提供了6个强大的工具：

创建搜索任务（需要支付）

为任何搜索查询创建自动化监控任务。

• 支付：每个任务创建1.00美元
• 参数：查询、频率（最小1分钟）、电子邮件、运行时间、发件人电子邮件
• 云模式：自动部署Lambda函数+EventBridge计划
• 本地模式：在数据库中创建任务以供手动执行
• 示例："创建一个搜索任务，每小时监测AI伦理工作职位发布，持续3天"

执行搜索（免费）

对现有任务执行搜索并获取新结果。

• 参数：任务ID
• 示例："执行任务1的搜索"

列出搜索任务（免费）

查看所有活动的监控任务。

• 示例："显示我所有的搜索任务"

获取任务状态（免费）

获取任务的详细状态和执行历史。

• 参数：任务ID，要显示的最近执行次数
• 示例："任务2的状态是什么？"

删除搜索任务（免费）

停用一个监控任务。

• 参数：任务ID
• 示例："删除任务3"

搜索预览（免费）

预览搜索结果而不创建任务。

• 参数：查询、最大结果数
• 示例："预览搜索结果‘Python远程工作’"

使用示例

一旦配置完成，您可以在Claude Desktop中使用自然语言：

创建任务（需要支付）

• "创建一个搜索任务，每2小时监测新的iPhone发布，持续一周"
• "设置‘远程Python工作’的监控，每5分钟检查一次，持续1小时"
• "监测AI安全研究论文，每天检查一次，持续2周"

管理任务（免费）

• "显示我所有的活动搜索任务"
• "我的第一个搜索任务的状态是什么？"
• "执行任务2的搜索"
• "删除iPhone监控任务"

预览（免费）

• "预览搜索结果‘2025年机器学习会议’"
• "显示我将得到的加密新闻监控结果"

关键特性

AI驱动的重复检测

• 使用OpenAI识别真正的新项目而非格式变化
• 显著减少误报
• 考虑内容相似性、公司和位置

x402支付集成

• 自动处理任务创建的支付
• 其他所有操作免费访问
• 基于以太坊的支付，在base-sepolia网络上

丰富的搜索结果

• 从网络上的多个来源查找项目
• 提取标题、描述、URL、地点等
• 结构化数据格式便于处理

灵活的监控

• 任何搜索查询（工作、产品、新闻、研究等）
• 可配置的频率（1分钟到几天）
• 可定制的运行期（1分钟到几周）

电子邮件通知

• 当发现新项目时自动发送通知
• 结构化的电子邮件格式，包含所有项目细节
• 可配置的发件人/收件人

完整的历史跟踪

• 每个任务的完整执行历史
• 性能指标和错误跟踪
• 通过Claude Desktop轻松监控状态

强大的数据库管理

• 自动化模式迁移和版本跟踪
• 数据库完整性验证和孤立检测
• 一致的会话管理和自动清理
• 全面的CLI工具用于数据库操作

部署选项

本地开发模式

• 设置：使用带有本地数据库（SQLite）的.env
• 任务执行：通过MCP工具或调度脚本手动执行
• API服务器：本地运行python run_api_server.py
• MCP服务器：本地运行python run_mcp_server.py
• 数据库：SQLite文件
• 支付：仍然通过x402处理

AWS云模式

• 设置：使用带有DEPLOY_TO_CLOUD=true的.env
• 任务执行：通过Lambda函数+EventBridge自动执行
• 基础设施：RDS PostgreSQL、Lambda函数、SES、S3
• 扩展：无服务器，自动处理多个任务
• 成本：典型使用情况下每月约15-30美元
• 优点：
  • 不需要手动任务执行
  • 运行期结束后自动清理
  • 通过SES的专业电子邮件模板
  • 可扩展且容错

混合模式

• API服务器：本地开发，使用生产基础设施的数据库
• 任务创建：创建Lambda函数以供执行
• 适合：在使用生产基础设施的同时进行开发

自动化选项

选项1：Cron作业

# 每5分钟检查一次，运行符合条件的任务
*/5 * * * * cd /path/to/MCP_Server && python scripts/run_scheduled_tasks.py

选项2：后台服务

# 持续运行，每60秒检查一次
python scripts/run_scheduled_tasks.py --daemon --interval  60

API端点

FastAPI服务器提供以下端点：

• POST /tasks/ - 创建任务（需要支付）
• GET /tasks/ - 列出活动任务
• GET /tasks/{task_id} - 获取任务详情
• DELETE /tasks/{task_id} - 停用任务
• POST /executions/ - 创建执行记录
• PUT /executions/{execution_id} - 更新执行
• GET /tasks/{task_id}/should-run - 检查任务是否应运行
• POST /tasks/{task_id}/results - 保存搜索结果
• GET /tasks/{task_id}/results - 获取先前的结果
• POST /tasks/{task_id}/notify - 发送电子邮件通知

故障排除

MCP服务器未显示

1. 检查Claude Desktop日志：~/Library/Logs/Claude/mcp*.log
2. 验证配置中的绝对路径
3. 确保已安装所有依赖项
4. 完全重启Claude Desktop

支付问题

1. 验证.env中设置了PRIVATE_KEY
2. 确保您在base-sepolia网络上有资金
3. 检查私钥是否有效（没有0x前缀）

搜索错误

1. 验证OPENAI_API_KEY或ANTHROPIC_API_KEY已设置且有效
2. 检查互联网连接
3. 查看API服务器日志以获取详细的错误消息

数据库问题

1. 检查timelooker.db是否存在且可写
2. 运行python scripts/init_db.py --validate以检查数据库完整性
3. 运行python scripts/init_db.py --reset以重新初始化如果损坏
4. 使用python scripts/init_db.py --version检查模式版本
5. 验证环境变量中的SQLAlchemy连接字符串

任务频率指南

• 实时测试：1-5分钟（短时段）
• 突发新闻：5-30分钟
• 职位发布：1-6小时
• 产品发布：6-24小时
• 研究论文：1-7天

注意：非常频繁的检查非常适合测试，但请注意长期任务的OpenAI API费用。

相关文件

项目组织

MCP_Server/
├── src/
│   ├── api/          # HTTP API层
│   ├── core/         # 核心业务逻辑
│   └── mcp/          # MCP服务器接口
├── tests/            # 所有测试文件
├── scripts/          # 实用脚本
├── run_mcp_server.py # MCP服务器入口点
└── run_api_server.py # API服务器入口点

• 测试：tests/test_search_quality.py，tests/quick_quality_test.py，tests/monitor_query_test.py
• Lambda：scripts/lambda_function.py，tests/test_lambda.py
• 自动化：scripts/run_scheduled_tasks.py
• 入口点：run_mcp_server.py，run_api_server.py

测试

该项目包括一个全面的测试套件，涵盖搜索质量、支付集成和API功能。

快速测试运行

# 运行所有测试
python tests/run_all_tests.py

# 交互式测试选择
python tests/run_quality_tests.py

测试类别

1. 搜索质量测试

验证搜索结果质量和重复检测：

python tests/quick_quality_test.py
python tests/test_search_quality.py

2. X402支付集成测试

使用模拟的x402客户端测试支付流程：

python tests/test_x402_integration.py

测试包括：

• 使用支付配置初始化TaskManager
• 创建任务的支付流程（模拟）
• 免费端点无需支付即可工作
• 处理缺失的私钥
• 支付失败场景

3. API集成测试

测试FastAPI端点和数据库集成：

# 首先启动API服务器
python run_api_server.py

# 在另一个终端运行测试
python tests/test_api_integration.py

测试包括：

• 健康端点功能
• 免费端点（GET请求）
• 需要支付的端点（POST /tasks/）
• 数据库模型操作

4. Lambda函数测试

测试AWS Lambda兼容性：

python tests/test_lambda.py

测试配置

对于API集成测试，确保：

• API服务器在localhost:8000上运行
• .env中有有效的环境变量
• 数据库已使用python scripts/init_db.py初始化

测试环境变量

# 支付测试所需
PRIVATE_KEY=your_test_private_key_here
PAY_TO_ADDRESS=0x671cE47E4F38051ba3A990Ba306E2885C2Fe4102
X402_NETWORK=base-sepolia

# 搜索测试所需
OPENAI_API_KEY=your_openai_api_key_here
ANTHROPIC_API_KEY=your_anthropic_api_key_here

# 测试可选
DATABASE_URL=sqlite:///test.db
LOG_LEVEL=INFO

测试结果

测试套件验证：

• ✅ 搜索结果的质量和一致性
• ✅ AI驱动的重复检测准确性
• ✅ 支付集成流程
• ✅ API端点功能
• ✅ 数据库操作
• ✅ 错误处理和边缘情况

该系统提供了强大的搜索监控能力，并通过x402协议无缝集成了支付！