HubSpot CRM Server

CI License: MIT

HubSpot CRM MCP server for Claude Desktop and any MCP client, free-tier compatible. 15 tools over contacts, deals, and pipelines, authenticated with either a HubSpot private-app token or a full OAuth app. Writes are idempotent, so a retried tool call replays its first result instead of creating a duplicate record, and every tool call is written to a PII-redacted audit trail, including the calls denied for a missing scope and the calls that errored.

Related: QuickBooks Online MCP Server · MCP Audit Gateway · What production MCP actually requires

HubSpot runs a hosted MCP server of its own, with wider object coverage than this one. It is built for self-hosting: you run the process, the source is short enough to read in a sitting, and the audit trail, the cache, and the credentials never leave your infrastructure.

The rest is what a REST wrapper usually leaves out. It prompts for the exact missing scope instead of leaking a raw 403, retries rate limits with backoff, serves record reads from a local cache (TTL plus write-invalidation, with a signature-verified webhook handler you can wire to your own HTTP ingress for out-of-band changes), walks cursor pagination, and returns per-item results when a batch partially fails.

See docs/COMPARISON.md for side-by-side transcripts of a naive API-wrapper MCP versus this one. Every transcript is generated by running both against the mocked test suite.

Architecture

flowchart TD
    Agent["MCP client / agent"] -->|"stdio (JSON-RPC)"| Server["FastMCP server<br/>server.py"]
    Server --> Service["CrmService<br/>scope checks · audit · orchestration"]

    Service --> Cache["LocalCache<br/>TTL + write invalidation"]
    Service --> Idem["Idempotency store"]
    Service --> Audit["Audit log<br/>PII redaction"]
    Service --> Client["HubSpotClient<br/>retries · pagination · error mapping"]

    Client --> Auth["Token provider<br/>private-app · OAuth refresh"]
    Client -->|HTTPS| HubSpot["HubSpot CRM API"]

    Ingress["Your HTTP ingress<br/>(optional, host-provided)"] -->|"signed v3 payload"| Processor["WebhookProcessor<br/>verify_signature"]
    Processor -->|"invalidate(object)"| Cache

The stdio server speaks JSON-RPC only; it does not listen for webhooks. WebhookProcessor and verify_signature are shipped as a tested component you mount on your own HTTP ingress (see Webhook cache invalidation).

Tools

Every tool carries MCP annotations (readOnlyHint, destructiveHint, idempotentHint, openWorldHint), a described input schema, and an output schema.

Tool	What it does	Scope
crm_search_contacts	Search contacts by free-text term, one page at a time	crm.objects.contacts.read
crm_list_contacts	List contacts in id order with cursor pagination	crm.objects.contacts.read
crm_get_contact	Fetch one contact by id, served from the local cache	crm.objects.contacts.read
crm_create_contact	Create a contact, idempotent on a supplied or derived key	crm.objects.contacts.write
crm_update_contact	Overwrite the properties passed and invalidate the cache entry	crm.objects.contacts.write
crm_delete_contact	Archive (soft-delete) a contact	crm.objects.contacts.write
crm_batch_create_contacts	Create up to 100 contacts and report per-row failures	crm.objects.contacts.write
crm_search_deals	Search deals by free-text term with cursor pagination	crm.objects.deals.read
crm_get_deal	Fetch one deal by id, served from the local cache	crm.objects.deals.read
crm_create_deal	Create a deal, idempotent on a supplied or derived key	crm.objects.deals.write
crm_update_deal	Update deal properties, including moving it to another stage	crm.objects.deals.write
crm_delete_deal	Archive (soft-delete) a deal	crm.objects.deals.write
crm_list_pipelines	List deal pipelines with stages in display order (cached)	crm.schemas.deals.read
crm_get_pipeline	Fetch one pipeline with its ordered stages	crm.schemas.deals.read
crm_export_audit_log	Export the session audit trail as JSON Lines	none

Quickstart

Run it without installing anything permanent:

uvx mcp-hubspot

Or install it:

pip install mcp-hubspot
mcp-hubspot

Either form speaks MCP over stdio on stdin and stdout. Authenticate with either a private-app token (HUBSPOT_PRIVATE_APP_TOKEN) or an OAuth app (HUBSPOT_CLIENT_ID + HUBSPOT_CLIENT_SECRET + HUBSPOT_REFRESH_TOKEN). The server auto-detects which is present.

Credentials are resolved lazily. The server starts and answers initialize and tools/list with nothing configured, which is what lets a directory or a sandbox introspect it; the first tool call is where a missing credential turns into an actionable error.

Wiring into an MCP client

Add to your MCP host config (for example Claude Desktop's claude_desktop_config.json):

{
  "mcpServers": {
    "hubspot": {
      "command": "uvx",
      "args": ["mcp-hubspot"],
      "env": { "HUBSPOT_PRIVATE_APP_TOKEN": "pat-na1-..." }
    }
  }
}

Docker

docker build -t mcp-hubspot .
docker run --rm -i -e HUBSPOT_PRIVATE_APP_TOKEN=pat-na1-... mcp-hubspot

From source

Requires Python 3.12+ and uv.

uv venv
uv pip install -e ".[dev]"

cp .env.example .env   # then fill in your HubSpot credentials
uv run mcp-hubspot

OAuth authorization URL

For the OAuth flow, mcp_crm.auth.build_authorization_url(...) builds the consent URL with the scopes the tools need; exchange the returned code for a refresh token and set HUBSPOT_REFRESH_TOKEN.

Webhook cache invalidation

The stdio server does not receive webhooks. To invalidate the cache on changes made outside this process (edits in the HubSpot UI, other integrations), mount WebhookProcessor on your own HTTP endpoint and verify HubSpot's v3 signature with verify_signature (using the HUBSPOT_WEBHOOK_SECRET you configure). Point it at the same LocalCache your CrmService uses:

from mcp_crm.webhooks import WebhookProcessor, verify_signature

processor = WebhookProcessor(cache)

def handle_hubspot_webhook(request):
    ok = verify_signature(
        secret=webhook_signing_secret,
        method="POST",
        uri=request.url,
        body=request.raw_body,
        signature=request.headers["X-HubSpot-Signature-v3"],
        timestamp=request.headers["X-HubSpot-Request-Timestamp"],
    )
    if not ok:
        return 401
    processor.process(request.json())
    return 200

verify_signature rejects tampered bodies, wrong secrets, and stale timestamps; WebhookProcessor.process maps each subscription to the object it invalidates and reports what it touched.

Design decisions

• Cache scope. Only object-detail reads (crm_get_contact, crm_get_deal) and the pipeline list are cached; list/search results are query-dependent and left uncached to avoid serving stale result sets. Writes invalidate the relevant object immediately. For out-of-band changes, a signature-verified WebhookProcessor ships as a component you mount on your own HTTP ingress (see Webhook cache invalidation); the stdio server itself does not listen for webhooks.
• Idempotency is client-side. HubSpot's create endpoints are not natively idempotent, so a key (supplied or derived from the payload) is stored and replayed. This makes at-least-once tool retries safe without duplicating records. The store lives in the process, so it covers retries within a session rather than across restarts, and crm_batch_create_contacts deliberately does not use it; both facts are stated in the tool descriptions.
• Scope prompting happens twice. The service pre-checks granted scopes (via token introspection) for a fast, actionable error, and the HTTP client also maps a server-side MISSING_SCOPES 403 to the same typed error (belt and suspenders).
• Credentials load lazily. Nothing reads a token at import or at startup. A missing credential surfaces as a typed error on the first tool call, and that failure is audited like any other, so the server is still introspectable in a sandbox with an empty environment.
• Backoff and clocks are injectable. Retry sleep, RNG jitter, and time sources are constructor parameters, which is why the whole suite runs offline in well under a second.

Testing

Every external HubSpot call is served by an in-memory fake (tests/fake_hubspot.py) backed by JSON fixtures (tests/fixtures/), wired in through httpx.MockTransport. No network, no credentials, deterministic.

uv run pytest -q

Regenerate the comparison document (CI also checks it stays in sync):

uv run python scripts/generate_comparison.py

Registry metadata

server.json describes this package for the Model Context Protocol registry (io.github.amin-ale/hubspot-crm-mcp, PyPI mcp-hubspot, stdio transport). .mcp.json is the minimal client config for tools that auto-detect MCP servers from a repository root.

Scope and safety

This is a client for HubSpot data you own or are authorized to access. Point it only at HubSpot accounts you control or have written permission to operate. The audit log redacts emails and phone numbers before writing records; treat exported audit logs as sensitive regardless. Behaviour documented here is point-in-time against the included fixtures, not a guarantee about any live HubSpot account.

Hire me

I build MCP servers and API integrations that survive a senior-dev code review: auth, retries, idempotency, and audit trails included, not bolted on later. Portfolio and contact: https://amin-ale.github.io/portfolio-site · [email protected].