Zoho CRM MCP 서버 (FastAPI + FastMCP)

FastAPI + FastMCP로 구축된 프로덕션급 Model Context Protocol(MCP) 서버로, Claude 및 기타 AI 클라이언트에게 Zoho CRM REST API v8에 대한 완전하고 인증된 액세스를 제공합니다 — 레코드 조회부터 모듈 설계, 워크플로 자동화 작성까지.

레코드, COQL, 스키마 설계, 워크플로 규칙 및 해당 액션, 웹훅, 대량/일괄 작업, 태그, 메모, 이메일, 보안 설정, 대량 가져오기/내보내기를 다루는 167개의 MCP 도구 — 그리고 전용 도구가 없는 Zoho 기능에 대한 범용 zoho_api_request 탈출구까지 포함합니다.

🌟 주요 기능

• FastAPI 웹 프레임워크: Uvicorn으로 구동되는 고성능, 프로덕션 준비 완료 ASGI 앱.

• 이중 전송: Streamable HTTP MCP 서버(원격/클라우드 호스팅용) 및 STDIO MCP 서버(로컬 Claude Desktop용)로 실행.

• 전체 OAuth 2.0 수명 주기: 자동 코드 교환, 브라우저 리디렉션 핸들러(/auth/callback), 암호화된 토큰 저장, 서버 실행 중 토큰을 최신 상태로 유지하는 백그라운드 루프.

• 채팅에서 구성 가능한 자격 증명: .env 대신 채팅에서 Zoho Client ID/Secret 제공(set_zoho_credentials, 또는 get_auth_url/exchange_auth_code에 인라인으로 전달) — 서버 재시작 없이 Zoho 계정 전환에 유용.

• 완전한 자동화 작성: 워크플로 규칙을 종단 간 구축 — 필드 업데이트, 이메일 알림, 작업, 웹훅 액션을 생성한 다음 트리거와 기준으로 규칙에 연결.

• 스키마 설계: 사용자 정의 모듈(Zoho가 요구하는 프로필 포함), 필드, 글로벌 선택 목록, 레이아웃, 영업 파이프라인 생성.

• 범위 제한 세션 모드: 특정 레코드 ID로 작업을 제한하는 ID 기반 안전 필터(activate_scope).

• 인간 개입(HITL) 승인: 파괴적 작업은 실행 대신 보류 요청을 대기열에 넣습니다. ZOHO_REQUIRE_APPROVAL로 전환.

• 구조화된 활동 로깅: 모든 인증 이벤트, API 호출, 승인 결정이 JSON으로 기록되며 get_logs() / GET /logs로 검색 가능.

• 암호화된 토큰 저장: OAuth 토큰이 저장 시 암호화(Fernet/AES)되며, 평문으로 저장되지 않습니다.

• 복원력 있는 네트워크 클라이언트: 자동 401 새로고침 및 재시도, 429 백오프 제한, 지수 5xx 재시도, 아웃바운드 속도 제한기, Zoho의 레코드별 응답에서 부분 실패 감지를 갖춘 풀링된 httpx 클라이언트.

• 자동화된 테스트 스위트: HTTP 표면, 도구 등록, 요청 페이로드 형태, 클라이언트 가드를 다루는 35개의 pytest 테스트.

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 Console)

  • 클라이언트 유형: 서버 기반 애플리케이션

  • 리디렉션 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?)를 호출하도록 요청하거나, get_auth_url / exchange_auth_code에 client_id/client_secret를 직접 전달하세요. client_id를 전환하면 이전 계정에 저장된 토큰이 삭제되어, 다른 앱에 발급된 리프레시 토큰을 재사용할 때 발생하는 Zoho의 invalid_client 오류를 방지합니다.

3. 종속성 설치

pip install -r requirements.txt

🚀 실행 및 배포

옵션 A: 로컬 FastAPI 웹 서버

python server.py

또는 Uvicorn을 직접 사용:

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

실행 후:

• 웹 대시보드: 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에 로그인한 후 수락(Accept) 클릭.

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

🔐 토큰 보안

• 토큰은 ~/.zoho_crm_tokens.json에 저장 시 암호화(Fernet/AES)됩니다.

• 키는 첫 실행 시 ~/.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 도구 모두의 등록, 워크플로/모듈/메모/통화/웹훅/병합/잠금에 대해 전송되는 정확한 요청 페이로드, 클라이언트 측 검증 가드, 속도 제한 클램핑, Zoho 부분 실패 감지를 다룹니다.

테스트는 완전히 오프라인으로 실행됩니다 — Zoho 자격 증명이 필요 없습니다.

🛠️ MCP 도구 참조

† 기본적으로 승인 게이트가 적용됩니다. 즉시 실행하려면 ZOHO_REQUIRE_APPROVAL=false로 설정하세요.

* Zoho CRM 공개 REST API에는 이 작업을 위한 엔드포인트가 없습니다. 블루프린트 작성, Deluge 함수 소스, 보고서/대시보드 생성은 UI 전용이거나 별도의 Zoho Analytics 제품에 속합니다. 이러한 도구는 존재하지 않는 URL에 대해 실패하는 대신, 작동하는 대안을 명시한 명확한 NOT_SUPPORTED_BY_ZOHO_API 메시지를 반환합니다.

🧭 목록에 없는 모든 항목에 접근하기

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}.