Zoho CRM MCP 服务器（FastAPI + FastMCP）

一个基于 FastAPI + FastMCP 构建的生产级模型上下文协议（MCP）服务器，为 Claude 和其他 AI 客户端提供对 Zoho CRM REST API v8 的完整、经过身份验证的访问——从读取记录到设计模块和编写工作流自动化。

167 个 MCP 工具，涵盖记录、COQL、架构设计、工作流规则及其操作、Webhook、批量/大规模操作、标签、备注、电子邮件、安全设置以及批量导入/导出——此外还有一个通用的 zoho_api_request 逃生通道，用于访问 Zoho 暴露但尚无专用工具的任何功能。

🌟 主要特性

• FastAPI Web 框架： 高性能、生产就绪的 ASGI 应用，由 Uvicorn 驱动。

• 双传输模式： 既可作为 Streamable HTTP MCP 服务器（用于远程/云托管）运行，也可作为 STDIO MCP 服务器（用于本地 Claude Desktop）运行。

• 完整的 OAuth 2.0 生命周期： 自动代码交换、浏览器重定向处理程序（/auth/callback）、加密令牌存储，以及一个在服务器运行期间保持令牌新鲜的后台循环。

• 可通过聊天配置凭据： 通过聊天提供 Zoho 客户端 ID/密钥（set_zoho_credentials，或直接在 get_auth_url/exchange_auth_code 中内联提供），而无需使用 .env——便于在不重启的情况下切换 Zoho 账户。

• 完整的自动化编写能力： 端到端构建工作流规则——创建字段更新、邮件通知、任务和 Webhook 操作，然后通过触发器和条件将其接入规则。

• 架构设计： 创建自定义模块（包含 Zoho 强制要求的配置文件）、字段、全局选择列表、布局和销售管道。

• 作用域会话模式： 基于 ID 的安全过滤器（activate_scope），将操作限制在特定记录 ID 范围内。

• 人工审批（HITL）： 破坏性操作会排队等待审批请求，而不是直接执行。可通过 ZOHO_REQUIRE_APPROVAL 切换。

• 结构化活动日志： 每个认证事件、API 调用和审批决策均以 JSON 格式记录，可通过 get_logs() / GET /logs 检索。

• 加密令牌存储： OAuth 令牌在静态存储时加密（Fernet/AES），绝不使用明文。

• 弹性网络客户端： 池化 httpx 客户端，具有自动 401 刷新重试、429 限流退避、指数 5xx 重试、出站速率限制器，以及针对 Zoho 逐记录响应的部分失败检测。

• 自动化测试套件： 35 个 pytest 测试，涵盖 HTTP 表面、工具注册、请求负载结构和客户端防护。

Related MCP server: Zoho CRM MCP Server

📁 仓库结构

zoho-crm-mcp/
├── server.py              # FastAPI app + all FastMCP tool definitions & REST endpoints
├── auth_manager.py        # OAuth 2.0 flow, scopes & token refresh
├── zoho_client.py         # Async HTTP client for Zoho CRM API v8 (151 methods)
├── models.py              # Pydantic state & validation models
├── token_store.py         # Encrypted (Fernet) token persistence
├── approval_manager.py    # HITL approval queue for high-risk actions
├── activity_log.py        # Structured JSON activity logger
├── test_server.py         # pytest suite
├── requirements.txt       # Dependencies
├── .env.example           # Environment configuration template
├── pyproject.toml         # Package metadata
└── README.md

⚙️ 设置与安装

1. 前置条件

• Python 3.10+

• 一个 Zoho CRM API 控制台应用（Zoho API 控制台）

  • 客户端类型： 基于服务器的应用

  • 重定向 URI： http://localhost:8000/auth/callback（或您的部署回调 URL）

2. 环境设置

cp .env.example .env

最低配置：

ZOHO_CLIENT_ID=1000.xxxxxxx
ZOHO_CLIENT_SECRET=xxxxxxx
ZOHO_REDIRECT_URI=http://localhost:8000/auth/callback
ZOHO_DATA_CENTER=com
PORT=8000

有关所有支持的变量（包括审批门控、OAuth 作用域覆盖、速率限制和超时设置），请参阅 .env.example。

需要同时使用多个 Zoho 账户？ ZOHO_CLIENT_ID/ZOHO_CLIENT_SECRET 是可选的。将其留空，让 Claude 调用 set_zoho_credentials(client_id, client_secret, redirect_uri?, data_center?)，或者将 client_id/client_secret 直接传递给 get_auth_url / exchange_auth_code。切换 client_id 会清除为先前账户保存的令牌，从而避免因重用颁发给其他应用的刷新令牌而触发 Zoho 的 invalid_client 错误。

3. 安装依赖

pip install -r requirements.txt

🚀 运行与部署

选项 A：本地 FastAPI Web 服务器

python server.py

或直接使用 Uvicorn：

uvicorn server:app --host 0.0.0.0 --port 8000

运行后：

• Web 仪表盘： http://localhost:8000/

• Swagger 文档： http://localhost:8000/docs

• 健康检查： http://localhost:8000/health

• MCP 端点： http://localhost:8000/mcp

选项 B：本地 STDIO

python server.py --stdio

选项 C：云部署（Render、Railway、Docker、AWS、Heroku）

• 启动命令： uvicorn server:app --host 0.0.0.0 --port $PORT

• 健康检查路径： /health

• 环境变量： 设置 ZOHO_CLIENT_ID、ZOHO_CLIENT_SECRET、ZOHO_REDIRECT_URI、ZOHO_DATA_CENTER 和 ZOHO_TOKEN_ENCRYPTION_KEY（以便令牌在临时文件系统上重启后仍然有效）。

🖥️ Claude Desktop 集成

模式 1：HTTP / 远程 MCP 连接

{
  "mcpServers": {
    "zoho-crm": {
      "url": "http://localhost:8000/mcp"
    }
  }
}

模式 2：本地 STDIO 连接

{
  "mcpServers": {
    "zoho-crm": {
      "command": "python",
      "args": ["C:/Users/Lenovo/Desktop/zoho MCP/server.py", "--stdio"],
      "env": {
        "ZOHO_CLIENT_ID": "1000.YOUR_CLIENT_ID",
        "ZOHO_CLIENT_SECRET": "YOUR_CLIENT_SECRET",
        "ZOHO_REDIRECT_URI": "http://localhost:8000/auth/callback",
        "ZOHO_DATA_CENTER": "com"
      }
    }
  }
}

🔑 首次运行 OAuth 流程

1. 启动服务器：python server.py

2. 打开 http://localhost:8000/auth/url，或让 Claude 运行 get_auth_url()。

3. 打开返回的 URL，登录 Zoho CRM，点击 接受。

4. Zoho 重定向到 /auth/callback?code=...；服务器交换代码并将加密令牌保存到 ~/.zoho_crm_tokens.json。

🧩 构建自动化：工作流配方

Zoho 将工作流规则建模为触发器加条件，其中每个条件指向预先创建的操作对象。请按以下顺序构建：

1. get_workflow_configurations(module="Leads")
   -> see which triggers, comparators, and action types this org supports

2. create_field_update_action(
       name="Mark as Hot", module="Leads",
       field_api_name="Rating", value="Hot")
   -> returns the action id

3. create_workflow(
       name="Hot Lead Router",
       module="Leads",
       execute_when={"type": "create_or_edit"},
       conditions=[{
           "sequence_number": 1,
           "criteria_details": {"criteria": {"group_operator": "and", "group": [
               {"comparator": "equal",
                "field": {"api_name": "Lead_Source"},
                "value": "Web Form"}]}},
           "instant_actions": {"actions": [
               {"id": "<action id from step 2>", "type": "field_updates"}]}}])

4. activate_workflow(workflow_id="...")

同样的模式也适用于以 create_email_notification_action、create_automation_task 和 create_webhook 作为操作来源的情况。

🎯 作用域会话模式（安全过滤器）

将所有操作限制在特定记录 ID 范围内：

• 激活： activate_scope(module="Deals", record_ids=["4153...001", "4153...002"])

• REST： POST /scope/activate，请求体为 {"module": "Leads", "record_ids": ["123", "456"]}

• 停用： deactivate_scope() 或 POST /scope/deactivate

激活后，对该模块的读取将被过滤为仅限这些 ID，对任何其他 ID 的写入将被拒绝并返回 OUT_OF_SCOPE。

✅ 人工审批（HITL）

默认情况下，破坏性操作会排队等待审批请求并返回 request_id，而不是直接执行：

delete_record、bulk_update_records、bulk_delete_records、mass_update_records、mass_delete_records、change_owner、mass_change_owner、merge_records、delete_workflow、delete_workflows、execute_blueprint、update_layout、activate_layout、delete_layout、delete_field、delete_user、delete_tag、bulk_write_create_job。

• 查看： list_pending_approvals() 或 GET /approvals

• 批准并执行： approve_action(request_id="...") 或 POST /approvals/{id}/approve

• 拒绝并丢弃： reject_action(request_id="...") 或 POST /approvals/{id}/reject

• 完全禁用门控： 设置 ZOHO_REQUIRE_APPROVAL=false，使这些工具立即执行。

每个请求、批准和拒绝都会写入活动日志。

📜 活动日志

认证事件、出站 Zoho API 调用、函数执行和审批决策均记录为 {timestamp, action, status, details} 条目——保存在内存中并追加到 ~/.zoho_crm_mcp_activity.log.jsonl。

• 检索： get_logs(limit=50, action=None, status=None) 或 GET /logs

🔐 令牌安全

• 令牌在静态存储时加密（Fernet/AES），保存在 ~/.zoho_crm_tokens.json 中。

• 密钥在首次运行时自动生成到 ~/.zoho_crm_mcp.key（在 POSIX 系统上仅限当前用户权限），或通过 ZOHO_TOKEN_ENCRYPTION_KEY 显式设置，以便在容器重启时保持密钥稳定。

• 令牌带有颁发它们的 client_id 标签，不匹配时会被丢弃，这可以防止切换账户后出现 Zoho 的 invalid_client 错误。

• 出站调用在 429/5xx 退避的基础上还会进行自我限流（ZOHO_RATE_LIMIT_PER_SEC，默认 10 次/秒）。

🧪 测试

pytest -v

涵盖 HTTP 表面（/health、/、/auth/*、/scope/*、/approvals/*、/logs）、全部 167 个 MCP 工具的注册、为工作流/模块/备注/通话/Webhook/合并/锁定发送的确切请求负载、客户端验证防护、速率限制钳制以及 Zoho 部分失败检测。

测试完全离线运行——无需 Zoho 凭据。

🛠️ MCP 工具参考

† 默认需要审批。设置 ZOHO_REQUIRE_APPROVAL=false 可立即执行。

* Zoho CRM 的公共 REST API 没有针对此操作的端点——蓝图编写、Deluge 函数源码以及报表/仪表盘创建仅限 UI 操作，或属于独立的 Zoho Analytics 产品。这些工具会返回一条明确的 NOT_SUPPORTED_BY_ZOHO_API 消息，并指明可用的替代方案，而不是针对不存在的 URL 报错。

🧭 访问未列出的任何内容

Zoho 的 API 比任何手写封装都要庞大。zoho_api_request 以相同的认证、限流和重试处理覆盖其余部分：

zoho_api_request(
    method="GET",
    endpoint="settings/territories")

zoho_api_request(
    method="POST",
    endpoint="settings/automation/scoring_rules",
    body={"scoring_rules": [{...}]})

zoho_api_request(
    method="GET",
    endpoint="read/1234567890",
    api_root="bulk")

api_root 选择 URL 基础地址：crm → {domain}/crm/v8（默认），bulk → {domain}/crm/bulk/v8，root → {domain}。