Keap MCP 服务器

一个高性能的模型上下文协议 (MCP) 服务器，用于与 Keap CRM 数据进行交互，具备 HTTP/2 支持、全面诊断和批量操作等高级功能。

功能特性

核心联系人与标签管理

• 全面的联系人管理 - 列出、搜索、筛选并获取详细的联系人信息

• 高级标签操作 - 完整的标签生命周期管理，包括创建、查询和批量操作

• 批量标签操作 - 高效地为多个联系人添加或移除多个标签

• 自定义字段操作 - 按自定义字段值搜索联系人并批量更新自定义字段

• 复杂的逻辑筛选 - 支持带有嵌套条件的 AND、OR、NOT 运算符

性能与优化

• HTTP/2 支持 - 通过连接池增强连接性能

• 智能速率限制 - 带有自适应退避策略的每日请求限制

• 查询优化 - 结合性能分析的智能服务端与客户端筛选

• 性能监控 - 实时查询分析与优化建议

• 持久化缓存 - 基于 SQLite 的缓存，以减少 API 调用并提高性能

高级功能

• 全面诊断 - API 性能指标、系统监控和健康检查

• 增强的错误处理 - 针对不同错误类型的稳健重试逻辑及指数退避

• 批量自定义字段更新 - 高效地设置多个联系人的自定义字段值

• 高级筛选运算符 - 15+ 种运算符，包括 BETWEEN、IN、SINCE、STARTS_WITH 等

• ID 列表操作 - 用于处理联系人和标签 ID 集合的实用函数

Related MCP server: claude-cli-mcp

架构

Keap MCP 服务器采用了精简的高性能架构：

• API 客户端 (src/api/client.py) - 增强的 Keap API 通信，支持 HTTP/2、速率限制和诊断

• MCP 工具 (src/mcp/) - 经过优化的全面 MCP 协议实现

• 缓存管理器 (src/cache/) - 基于 SQLite 的持久化缓存，支持智能失效机制

• 优化引擎 (src/mcp/optimization/) - 查询优化与性能分析

• 模式 (src/schemas/) - 数据验证与模型

• 工具库 (src/utils/) - 用于联系人处理和筛选的共享工具

MCP 工具

服务器提供了 17 个全面的 MCP 工具：

联系人操作

1. list_contacts - 列出联系人，支持筛选和分页（现已优化）

2. search_contacts_by_email - 通过电子邮件地址查找联系人

3. search_contacts_by_name - 通过姓名查找联系人

4. get_contact_details - 获取特定联系人的详细信息

5. query_contacts_by_custom_field - 按自定义字段值查询联系人

标签操作

6. get_tags - 获取标签，支持可选筛选

7. get_tag_details - 获取特定标签的详细信息

8. get_contacts_with_tag - 获取拥有特定标签的联系人

9. create_tag - 创建新标签

标签管理（批量操作）

10. modify_tags - 为联系人添加或移除标签

11. apply_tags_to_contacts - 使用批量操作为多个联系人应用多个标签

12. remove_tags_from_contacts - 从多个联系人中移除多个标签

自定义字段管理

13. set_custom_field_values - 批量更新多个联系人的自定义字段值

高级查询与性能操作

14. query_contacts_optimized - 带有优化和性能分析的高级联系人查询

15. analyze_query_performance - 分析查询性能和优化潜力

系统操作

16. get_api_diagnostics - 全面的 API 诊断和性能指标

实用操作

17. intersect_id_lists - 查找多个 ID 列表的交集

入门指南

先决条件

• Python 3.9 或更高版本

• Keap API 凭据

安装

1. 克隆仓库：

   git clone https://github.com/yourusername/keapmcp.git
   cd keapmcp

2. 安装依赖：

   pip install -r requirements.txt

3. 配置您的 Keap API 凭据：

   • 应用程序使用 .env 文件进行配置

   • API 密钥已从 keapsync 复制，但您可以根据需要进行修改

   • 配置包括：

     KEAP_API_KEY=your_api_key_here
     KEAP_API_BASE_URL=https://api.infusionsoft.com/crm/rest/v1
     KEAP_MCP_HOST=127.0.0.1
     KEAP_MCP_PORT=5000
     KEAP_MCP_LOG_LEVEL=INFO
     KEAP_MCP_LOG_FILE=keap_mcp_server.log
     KEAP_MCP_CACHE_ENABLED=true
     KEAP_MCP_CACHE_TTL=3600

运行服务器

python run.py --host 127.0.0.1 --port 5000

命令行选项：

• --host - 绑定的主机（默认：127.0.0.1）

• --port - 监听的端口（默认：5000）

• --log-level - 日志级别（默认：INFO）

• --log-file - 日志文件路径（默认：keap_mcp_server.log）

• --no-console-log - 禁用控制台日志

测试与覆盖率

Keap MCP 服务器包含一套全面的测试套件，以确保可靠性和正确运行，并集成了完整的 CI/CD 流水线。

快速测试

使用 Makefile 进行简单的测试执行：

# Run all unit tests
make test

# Run tests with coverage reporting  
make coverage

# Generate HTML coverage report
make coverage-html

# Run service-specific tests
make test-services
make test-models

# Full development workflow
make dev-test

CI/CD 集成

该项目包含自动化测试和覆盖率报告：

• 持续集成：在 Python 3.9、3.10 和 3.11 上运行测试

• 覆盖率跟踪：强制要求至少 60% 的覆盖率

• 提交前检查：代码格式化、Linting 和安全检查

• 覆盖率徽章：在主分支上自动更新

测试类别

• 单元测试 (tests/unit/) - 带有全面 Mock 的独立组件测试

• 集成测试 (tests/integration/) - 端到端功能验证

• 性能测试 (tests/performance/) - 负载和优化验证

• API 验证 - Keap API 响应格式验证

覆盖率要求

• 当前覆盖率：55% 的集成覆盖率，因组件而异

• API 客户端：通过全面 Mock 测试核心功能

• MCP 工具：带有 Mock 依赖的集成测试

• 缓存系统：全面的持久化和性能测试

• 工具库：测试了联系人处理和筛选功能

• 优化：涵盖了性能分析和查询优化

运行特定测试

# Run all tests
python -m pytest tests/ -v

# With coverage reporting
python -m pytest tests/ --cov=src --cov-fail-under=90

# Integration tests (requires running server)
python -m pytest tests/integration/ -v

使用 MCP 服务器

示例：列出联系人（现已优化）

{
  "function": "list_contacts",
  "params": {
    "filters": [
      { "field": "email", "operator": "contains", "value": "@company.com" }
    ],
    "limit": 50,
    "include": ["id", "given_name", "family_name", "email"]
  }
}

注意：list_contacts 现在在内部使用优化引擎以获得更好的性能。如需详细的性能指标，请直接使用 query_contacts_optimized。

示例：按电子邮件搜索

{
  "function": "search_contacts_by_email",
  "params": {
    "email": "john.doe@company.com",
    "include": ["id", "given_name", "family_name", "email", "tags"]
  }
}

示例：获取标签

{
  "function": "get_tags",
  "params": {
    "include_categories": true,
    "limit": 100
  }
}

示例：批量标签操作

{
  "function": "apply_tags_to_contacts",
  "params": {
    "tag_ids": ["123", "456"],
    "contact_ids": ["1001", "1002", "1003"]
  }
}

示例：自定义字段查询

{
  "function": "query_contacts_by_custom_field",
  "params": {
    "field_id": "7",
    "field_value": "Engineering",
    "operator": "contains",
    "include": ["id", "given_name", "family_name", "email"]
  }
}

示例：创建新标签

{
  "function": "create_tag",
  "params": {
    "name": "VIP Customer",
    "description": "High-value customer segment",
    "category_id": "2"
  }
}

示例：批量自定义字段更新

{
  "function": "set_custom_field_values",
  "params": {
    "field_id": "7",
    "contact_ids": ["1001", "1002", "1003"],
    "common_value": "VIP Customer"
  }
}

或者为每个联系人设置单独的值：

{
  "function": "set_custom_field_values",
  "params": {
    "field_id": "7",
    "contact_values": {
      "1001": "Gold Tier",
      "1002": "Silver Tier", 
      "1003": "Bronze Tier"
    }
  }
}

示例：API 诊断

{
  "function": "get_api_diagnostics",
  "params": {}
}

示例：ID 列表交集

{
  "function": "intersect_id_lists",
  "params": {
    "lists": [
      {"list_id": "active_contacts", "item_ids": ["1", "2", "3", "4"]},
      {"list_id": "newsletter_subscribers", "item_ids": ["2", "3", "5", "6"]}
    ],
    "id_field": "item_ids"
  }
}

性能特性

HTTP/2 支持

服务器使用 HTTP/2 以增强性能，并支持连接池和保持连接 (keepalive)。

速率限制

• 每日请求限制（默认每天 25,000 次请求）

• 智能退避策略

• 速率限制监控和诊断

缓存策略

• 基于 SQLite 的持久化缓存

• 智能缓存失效机制

• 基于 TTL 的过期策略

• 缓存命中/未命中跟踪

错误处理

• 指数退避重试

• 针对超时、网络和 HTTP 错误的各种策略

• 全面的错误跟踪和诊断

许可证

本项目采用 MIT 许可证授权 - 有关详细信息，请参阅 LICENSE 文件。