MCP Keap

Keap MCP服务器

  

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

特性

核心联系人和标签管理

• 全面的联系人管理 -列出、搜索、过滤并获取详细的联系信息
• 高级标记操作 -完整的标签生命周期管理，包括创建、查询和批处理操作
• 批量标记操作 -高效地从多个联系人中应用或删除多个标签
• 自定义字段操作 -按自定义字段值搜索联系人并批量更新自定义字段
• 复杂逻辑筛选 -支持带嵌套条件的AND、OR、NOT运算符

性能与优化

• HTTP/2支持 -通过连接池增强连接性能
• 智能速率限制 -采用自适应退避策略的每日请求限制
• 查询优化 -智能服务器端与客户端过滤与性能分析
• 性能监控 -实时查询分析和优化建议
• 持久缓存 -基于SQLite的缓存可减少API调用并提高性能

高级功能

• 综合诊断 -API性能指标、系统监控和运行状况检查
• 增强的错误处理 -针对不同错误类型的指数回退的稳健重试逻辑
• 批量自定义字段更新 -跨多个联系人高效设置自定义字段值
• 高级过滤器操作员 -15+个运算符，包括BETWEEN、IN、SINCE、STARTS_WITH等。
• ID列表操作 -用于处理联系人和标签ID集的实用功能

建筑

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

• API客户端 (src/api/client.py)-通过HTTP/2、速率限制和诊断功能增强Keap API通信
• 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 -按自定义字段值查询联系人

标签操作

1. get_tags -使用可选过滤检索标签
2. get_tag_details -获取特定标签的详细信息
3. get_contacts_with_tag -获取具有特定标签的联系人
4. create_tag -创建新标签

标签管理（批量操作）

1. modify_tags -在联系人中添加或删除标签
2. apply_tags_to_contacts -使用批处理操作将多个标签应用于多个联系人
3. remove_tags_from_contacts -从多个联系人中删除多个标签

自定义字段管理

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

高级查询和性能操作

1. query_contacts_optimized -具有优化和性能分析功能的高级联系人查询
2. analyze_query_performance -分析查询性能和优化潜力

系统操作

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

公用事业运营

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

入门指南

先决条件

• Python 3.9或更高版本
• Keap API证书

安装

1. 克隆存储库：

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

1. 安装依赖项：

   pip install -r requirements.txt

1. 配置您的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上运行
• 覆盖范围跟踪：强制执行至少70%的单元测试覆盖率
• 代码质量：褶皱和格式检查
• 安全扫描：Bandit安全分析和安全依赖性检查
• 类型检查：MyPy静态类型分析（非阻塞）
• 预提交钩子：提交时自动进行代码质量检查
• 构建验证：导入和初始化测试

CI管道职位

1. Lint代码：Ruff linting和格式验证
2. 单元测试：具有跨Python版本覆盖率报告的完整测试套件
3. 安全扫描：土匪和安全安全分析
4. 类型检查：MyPy静态类型检查
5. 构建测试：包导入和服务器初始化验证
6. 覆盖范围报告：拉取请求的HTML覆盖率报告

测试类别

• 单元测试 (tests/unit/)-通过全面模拟进行单个组件测试
• 集成测试 (tests/integration/)-端到端功能验证
• 性能测试 (tests/performance/)-负载和优化验证
• API验证 -Keap API响应格式验证

覆盖范围要求

• 当前覆盖范围：55%的集成覆盖率，因组件而异
• API客户端：通过全面模拟测试核心功能
• MCP工具：具有模拟依赖关系的集成测试
• 缓存系统：全面的持久性和性能测试
• 公用事业：测试联系人处理和过滤功能
• 优化：涵盖性能分析和查询优化

运行特定测试

# 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通过连接池和保活连接来提高性能。

速率限制

• 每日请求限制（默认情况下为25000个请求/天）
• 智能退避策略
• 速率限制监测和诊断

缓存策略

• 基于SQLite的持久缓存
• 智能缓存失效
• 基于TTL的过期
• 缓存命中/未命中跟踪

错误处理

• 重试次数呈指数级回退
• 针对超时、网络和HTTP错误的不同策略
• 全面的错误跟踪和诊断

许可证

此项目根据MIT许可证获得许可-有关详细信息，请参阅许可证文件。