# HubSpot CRM MCP Server
[](https://github.com/amin-ale/hubspot-mcp-server/actions/workflows/ci.yml)
[](LICENSE)
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](https://github.com/amin-ale/mcp-quickbooks) ·
[MCP Audit Gateway](https://github.com/amin-ale/mcp-audit-gateway) ·
[What production MCP actually requires](https://amin-ale.github.io/portfolio-site/what-production-mcp-actually-requires.html)
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`](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
```mermaid
flowchart TD
Agent["MCP client / agent"] -->|"stdio (JSON-RPC)"| Server["FastMCP server
server.py"]
Server --> Service["CrmService
scope checks · audit · orchestration"]
Service --> Cache["LocalCache
TTL + write invalidation"]
Service --> Idem["Idempotency store"]
Service --> Audit["Audit log
PII redaction"]
Service --> Client["HubSpotClient
retries · pagination · error mapping"]
Client --> Auth["Token provider
private-app · OAuth refresh"]
Client -->|HTTPS| HubSpot["HubSpot CRM API"]
Ingress["Your HTTP ingress
(optional, host-provided)"] -->|"signed v3 payload"| Processor["WebhookProcessor
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](#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:
```bash
uvx mcp-hubspot
```
Or install it:
```bash
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`):
```json
{
"mcpServers": {
"hubspot": {
"command": "uvx",
"args": ["mcp-hubspot"],
"env": { "HUBSPOT_PRIVATE_APP_TOKEN": "pat-na1-..." }
}
}
}
```
### Docker
```bash
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](https://docs.astral.sh/uv/).
```bash
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:
```python
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](#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.
```bash
uv run pytest -q
```
Regenerate the comparison document (CI also checks it stays in sync):
```bash
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](https://amin-ale.github.io/portfolio-site) ·
[amin.ale.business@gmail.com](mailto:amin.ale.business@gmail.com).