Zoho CRM MCP Server (FastAPI + FastMCP)

A production-grade Model Context Protocol (MCP) server built with FastAPI + FastMCP that gives Claude and other AI clients complete, authenticated access to the Zoho CRM REST API v8 — from reading records to designing modules and authoring workflow automation.

167 MCP tools covering records, COQL, schema design, workflow rules and their actions, webhooks, bulk/mass operations, tags, notes, email, security settings, and bulk import/export — plus a generic zoho_api_request escape hatch for anything Zoho exposes that has no dedicated tool.

🌟 Key Features

• FastAPI Web Framework: High-performance, production-ready ASGI app powered by Uvicorn.

• Dual Transport: Runs as a Streamable HTTP MCP server (for remote/cloud hosting) and as a STDIO MCP server (for local Claude Desktop).

• Full OAuth 2.0 Lifecycle: Automatic code exchange, browser redirect handler (/auth/callback), encrypted token storage, and a background loop that keeps the token fresh while the server runs.

• Chat-Configurable Credentials: Supply a Zoho Client ID/Secret from chat (set_zoho_credentials, or inline on get_auth_url/exchange_auth_code) instead of .env — useful for switching Zoho accounts without restarting.

• Complete Automation Authoring: Build workflow rules end to end — create field-update, email-notification, task, and webhook actions, then wire them into a rule with triggers and criteria.

• Schema Design: Create custom modules (with the profiles Zoho mandates), fields, global picklists, layouts, and sales pipelines.

• Scoped Session Mode: ID-based safety filter (activate_scope) restricting operations to specific record IDs.

• Human-In-The-Loop Approvals: Destructive actions queue a pending request instead of executing. Toggle with ZOHO_REQUIRE_APPROVAL.

• Structured Activity Logging: Every auth event, API call, and approval decision logged as JSON, retrievable via get_logs() / GET /logs.

• Encrypted Token Storage: OAuth tokens encrypted at rest (Fernet/AES), never plaintext.

• Resilient Network Client: Pooled httpx client with automatic 401 refresh-and-retry, clamped 429 backoff, exponential 5xx retry, an outbound rate limiter, and partial-failure detection on Zoho's per-record responses.

• Automated Test Suite: 35 pytest tests covering the HTTP surface, tool registration, request-payload shapes, and client guards.

Related MCP server: Zoho CRM MCP Server

📁 Repository Structure

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

⚙️ Setup & Installation

1. Prerequisites

• Python 3.10+

• A Zoho CRM API Console app (Zoho API Console)

  • Client Type: Server-based Applications

  • Redirect URI: http://localhost:8000/auth/callback (or your deployment callback URL)

2. Environment Setup

cp .env.example .env

Minimum configuration:

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

See .env.example for every supported variable, including the approval gate, OAuth scope override, rate limit, and timeout settings.

Working with more than one Zoho account? ZOHO_CLIENT_ID/ZOHO_CLIENT_SECRET are optional. Leave them blank and ask Claude to call set_zoho_credentials(client_id, client_secret, redirect_uri?, data_center?), or pass client_id/client_secret directly to get_auth_url / exchange_auth_code. Switching client_id clears tokens saved for the previous account, avoiding Zoho's invalid_client error from reusing a refresh token issued to a different app.

3. Install Dependencies

pip install -r requirements.txt

🚀 Running & Deploying

Option A: Local FastAPI Web Server

python server.py

Or with Uvicorn directly:

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

Once running:

• Web Dashboard: http://localhost:8000/

• Swagger Docs: http://localhost:8000/docs

• Health Check: http://localhost:8000/health

• MCP Endpoint: http://localhost:8000/mcp/

The MCP endpoint is served at the mount root, so the path is /mcp/ — not /mcp/mcp. Keep the trailing slash: /mcp answers with a 307 redirect, which not every MCP client follows correctly on a POST.

Option B: Local STDIO

python server.py --stdio

Option C: Cloud Deployment (Render, Railway, Docker, AWS, Heroku)

• Start Command: uvicorn server:app --host 0.0.0.0 --port $PORT

• Health Check Path: /health

• Environment Variables: set ZOHO_CLIENT_ID, ZOHO_CLIENT_SECRET, ZOHO_REDIRECT_URI, ZOHO_DATA_CENTER, and ZOHO_TOKEN_ENCRYPTION_KEY (so tokens survive restarts on ephemeral filesystems).

Live deployment

Staying authenticated on an ephemeral host. Render wipes the filesystem on every deploy and restart, taking ~/.zoho_crm_tokens.json with it — and the Fernet key at ~/.zoho_crm_mcp.key is regenerated on each boot, so even a surviving token file could not be decrypted. Set both of these in the host's environment and the server re-authenticates itself instead of asking for tokens again:

• ZOHO_REFRESH_TOKEN — picked up on cold boot, so a fresh access token is minted automatically. Refresh tokens do not expire unless revoked.

• ZOHO_TOKEN_ENCRYPTION_KEY — a stable Fernet key, so the cached access token survives too. Generate one with:

  python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"

Token refresh itself is automatic and needs no configuration: get_valid_access_token() renews anything expiring within 5 minutes, and a background loop re-checks every ZOHO_TOKEN_REFRESH_INTERVAL seconds (default 120).

🖥️ Claude Desktop Integration

Mode 1: HTTP / Remote MCP Connection

Local:

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

Deployed:

{
  "mcpServers": {
    "zoho-crm": {
      "url": "https://zoho-crm-mcp-07t7.onrender.com/mcp/"
    }
  }
}

The server has no MCP-level authentication, so leave any "OAuth Client ID" field in the connector settings blank. The Zoho OAuth inside the server authenticates it to Zoho; it does not gate the MCP endpoint. Pointing a client at the wrong path (/mcp/mcp) returns 404, which some clients report as a misleading sign-in or registration failure rather than as a bad URL.

Mode 2: Local STDIO Connection

{
  "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"
      }
    }
  }
}

🔑 First-Run OAuth Flow

1. Start the server: python server.py

2. Open http://localhost:8000/auth/url, or ask Claude to run get_auth_url().

3. Open the returned URL, sign in to Zoho CRM, click Accept.

4. Zoho redirects to /auth/callback?code=...; the server exchanges the code and saves encrypted tokens to ~/.zoho_crm_tokens.json.

🧩 Building Automation: The Workflow Recipe

Zoho models a workflow rule as a trigger plus conditions, where each condition points at pre-created action objects. Build them in that order:

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="...")

The same pattern applies with create_email_notification_action, create_automation_task, and create_webhook as the action source.

🎯 Scoped Session Mode (Safety Filter)

Restrict every operation to specific record IDs:

• Activate: activate_scope(module="Deals", record_ids=["4153...001", "4153...002"])

• REST: POST /scope/activate with {"module": "Leads", "record_ids": ["123", "456"]}

• Deactivate: deactivate_scope() or POST /scope/deactivate

While active, reads on that module are filtered to those IDs, and writes to any other ID are refused with OUT_OF_SCOPE.

✅ Human-In-The-Loop (HITL) Approvals

By default, destructive actions queue a pending request and return a request_id instead of running:

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.

• Review: list_pending_approvals() or GET /approvals

• Approve & execute: approve_action(request_id="...") or POST /approvals/{id}/approve

• Reject & discard: reject_action(request_id="...") or POST /approvals/{id}/reject

• Disable the gate entirely: set ZOHO_REQUIRE_APPROVAL=false so these tools execute immediately.

Every request, approval, and rejection is written to the activity log.

📜 Activity Logging

Auth events, outbound Zoho API calls, function executions, and approval decisions are recorded as {timestamp, action, status, details} entries — held in memory and appended to ~/.zoho_crm_mcp_activity.log.jsonl.

• Retrieve: get_logs(limit=50, action=None, status=None) or GET /logs

🔐 Token Security

• Tokens are encrypted at rest (Fernet/AES) in ~/.zoho_crm_tokens.json.

• The key is auto-generated into ~/.zoho_crm_mcp.key on first run (user-only permissions on POSIX), or set explicitly via ZOHO_TOKEN_ENCRYPTION_KEY for a stable key across container restarts.

• Tokens are tagged with the client_id that issued them and discarded on mismatch, which prevents Zoho's invalid_client error after switching accounts.

• Outbound calls are self-throttled (ZOHO_RATE_LIMIT_PER_SEC, default 10/sec) on top of 429/5xx backoff.

🧪 Testing

pytest -v

Covers the HTTP surface (/health, /, /auth/*, /scope/*, /approvals/*, /logs), registration of all 167 MCP tools, the exact request payloads sent for workflows/modules/notes/calls/webhooks/merges/locks, client-side validation guards, rate-limit clamping, and Zoho partial-failure detection.

Tests run entirely offline — no Zoho credentials required.

🛠️ MCP Tool Reference

† Approval-gated by default. Set ZOHO_REQUIRE_APPROVAL=false to execute immediately.

* Zoho CRM's public REST API has no endpoint for this operation — blueprint authoring, Deluge function source, and report/dashboard creation are UI-only or belong to the separate Zoho Analytics product. These tools return a clear NOT_SUPPORTED_BY_ZOHO_API message naming a working alternative, rather than failing against a URL that doesn't exist.

🧭 Reaching Anything Not Listed

Zoho's API is larger than any hand-written wrapper. zoho_api_request covers the rest with the same auth, throttling, and retry handling:

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 selects the URL base: crm → {domain}/crm/v8 (default), bulk → {domain}/crm/bulk/v8, root → {domain}.