HubSpot CRM MCP Server

Servidor MCP do HubSpot CRM

Servidor MCP do HubSpot CRM para Claude Desktop e qualquer cliente MCP, compatível com o plano gratuito. 15 ferramentas para contatos, negócios e pipelines, autenticadas com um token de aplicativo privado do HubSpot ou um aplicativo OAuth completo. Gravações são idempotentes, então uma chamada de ferramenta repetida reproduz seu primeiro resultado em vez de criar um registro duplicado, e cada chamada de ferramenta é registrada em uma trilha de auditoria com PII removido, incluindo as chamadas negadas por escopo ausente e as chamadas que falharam.

Relacionados: Servidor MCP do QuickBooks Online · Gateway de Auditoria MCP · O que o MCP de produção realmente exige

O HubSpot executa um servidor MCP hospedado próprio, com cobertura de objetos mais ampla que este. Ele é feito para auto-hospedagem: você executa o processo, o código-fonte é curto o suficiente para ser lido em uma sentada, e a trilha de auditoria, o cache e as credenciais nunca saem da sua infraestrutura.

O resto é o que um wrapper REST geralmente deixa de fora. Ele solicita o escopo ausente exato em vez de vazar um 403 bruto, tenta novamente limites de taxa com backoff, atende leituras de registros de um cache local (TTL mais invalidação por gravação, com um manipulador de webhook verificado por assinatura que você pode conectar ao seu próprio ingress HTTP para alterações fora de banda), percorre paginação por cursor e retorna resultados por item quando um lote falha parcialmente.

Veja docs/COMPARISON.md para transcrições lado a lado de um wrapper de API ingênuo em MCP versus este. Cada transcrição é gerada executando ambos contra a suíte de testes simulada.

Arquitetura

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

O servidor stdio fala apenas JSON-RPC; ele não escuta webhooks. WebhookProcessor e verify_signature são fornecidos como um componente testado que você monta no seu próprio ingress HTTP (veja Invalidação de cache por webhook).

Ferramentas

Cada ferramenta carrega anotações MCP (readOnlyHint, destructiveHint, idempotentHint, openWorldHint), um esquema de entrada descrito e um esquema de saída.

Ferramenta	O que faz	Escopo
crm_search_contacts	Pesquisa contatos por termo de texto livre, uma página por vez	crm.objects.contacts.read
crm_list_contacts	Lista contatos em ordem de id com paginação por cursor	crm.objects.contacts.read
crm_get_contact	Busca um contato por id, servido do cache local	crm.objects.contacts.read
crm_create_contact	Cria um contato, idempotente em uma chave fornecida ou derivada	crm.objects.contacts.write
crm_update_contact	Sobrescreve as propriedades passadas e invalida a entrada do cache	crm.objects.contacts.write
crm_delete_contact	Arquiva (exclusão suave) um contato	crm.objects.contacts.write
crm_batch_create_contacts	Cria até 100 contatos e relata falhas por linha	crm.objects.contacts.write
crm_search_deals	Pesquisa negócios por termo de texto livre com paginação por cursor	crm.objects.deals.read
crm_get_deal	Busca um negócio por id, servido do cache local	crm.objects.deals.read
crm_create_deal	Cria um negócio, idempotente em uma chave fornecida ou derivada	crm.objects.deals.write
crm_update_deal	Atualiza propriedades do negócio, incluindo movê-lo para outro estágio	crm.objects.deals.write
crm_delete_deal	Arquiva (exclusão suave) um negócio	crm.objects.deals.write
crm_list_pipelines	Lista pipelines de negócios com estágios em ordem de exibição (em cache)	crm.schemas.deals.read
crm_get_pipeline	Busca um pipeline com seus estágios ordenados	crm.schemas.deals.read
crm_export_audit_log	Exporta a trilha de auditoria da sessão como JSON Lines	nenhum

Início rápido

Execute sem instalar nada permanente:

uvx mcp-hubspot

Ou instale:

pip install mcp-hubspot
mcp-hubspot

Qualquer forma fala MCP sobre stdio em stdin e stdout. Autentique com um token de aplicativo privado (HUBSPOT_PRIVATE_APP_TOKEN) ou um aplicativo OAuth (HUBSPOT_CLIENT_ID + HUBSPOT_CLIENT_SECRET + HUBSPOT_REFRESH_TOKEN). O servidor detecta automaticamente qual está presente.

As credenciais são resolvidas de forma preguiçosa. O servidor inicia e responde a initialize e tools/list com nada configurado, o que é o que permite que um diretório ou um sandbox o inspecione; a primeira chamada de ferramenta é onde uma credencial ausente se transforma em um erro acionável.

Conectando a um cliente MCP

Adicione à configuração do seu host MCP (por exemplo, o claude_desktop_config.json do Claude Desktop):

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

A partir do código-fonte

Requer Python 3.12+ e uv.

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

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

URL de autorização OAuth

Para o fluxo OAuth, mcp_crm.auth.build_authorization_url(...) constrói a URL de consentimento com os escopos que as ferramentas precisam; troque o código retornado por um token de atualização e defina HUBSPOT_REFRESH_TOKEN.

Invalidação de cache por webhook

O servidor stdio não recebe webhooks. Para invalidar o cache em alterações feitas fora deste processo (edições na interface do HubSpot, outras integrações), monte WebhookProcessor no seu próprio endpoint HTTP e verifique a assinatura v3 do HubSpot com verify_signature (usando o HUBSPOT_WEBHOOK_SECRET que você configura). Aponte-o para o mesmo LocalCache que seu CrmService usa:

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 rejeita corpos adulterados, segredos errados e timestamps obsoletos; WebhookProcessor.process mapeia cada assinatura para o objeto que ela invalida e relata o que tocou.

Decisões de design

• Escopo do cache. Apenas leituras de detalhes de objetos (crm_get_contact, crm_get_deal) e a lista de pipelines são armazenadas em cache; resultados de lista/pesquisa dependem da consulta e ficam sem cache para evitar servir conjuntos de resultados obsoletos. Gravações invalidam o objeto relevante imediatamente. Para alterações fora de banda, um WebhookProcessor verificado por assinatura é fornecido como um componente que você monta no seu próprio ingress HTTP (veja Invalidação de cache por webhook); o servidor stdio em si não escuta webhooks.
• Idempotência é do lado do cliente. Os endpoints de criação do HubSpot não são nativamente idempotentes, então uma chave (fornecida ou derivada do payload) é armazenada e reproduzida. Isso torna tentativas de ferramenta pelo menos uma vez seguras sem duplicar registros. O armazenamento vive no processo, então cobre tentativas dentro de uma sessão em vez de reinicializações, e crm_batch_create_contacts deliberadamente não o usa; ambos os fatos são declarados nas descrições das ferramentas.
• Solicitação de escopo acontece duas vezes. O serviço pré-verifica escopos concedidos (via introspecção de token) para um erro rápido e acionável, e o cliente HTTP também mapeia um MISSING_SCOPES 403 do lado do servidor para o mesmo erro tipado (cinto e suspensórios).
• Credenciais carregam de forma preguiçosa. Nada lê um token na importação ou na inicialização. Uma credencial ausente aparece como um erro tipado na primeira chamada de ferramenta, e essa falha é auditada como qualquer outra, então o servidor ainda é inspecionável em um sandbox com um ambiente vazio.
• Backoff e relógios são injetáveis. Sono de tentativa, jitter de RNG e fontes de tempo são parâmetros de construtor, que é o motivo pelo qual toda a suíte roda offline em bem menos de um segundo.

Testes

Cada chamada externa ao HubSpot é atendida por um fake em memória (tests/fake_hubspot.py) apoiado por fixtures JSON (tests/fixtures/), conectado via httpx.MockTransport. Sem rede, sem credenciais, determinístico.

uv run pytest -q

Regenere o documento de comparação (o CI também verifica se ele permanece sincronizado):

uv run python scripts/generate_comparison.py

Metadados do registro

server.json descreve este pacote para o registro do Model Context Protocol (io.github.amin-ale/hubspot-crm-mcp, PyPI mcp-hubspot, transporte stdio). .mcp.json é a configuração mínima de cliente para ferramentas que detectam automaticamente servidores MCP a partir da raiz de um repositório.

Escopo e segurança

Este é um cliente para dados do HubSpot que você possui ou está autorizado a acessar. Aponte-o apenas para contas do HubSpot que você controla ou tem permissão escrita para operar. O log de auditoria remove emails e números de telefone antes de gravar registros; trate logs de auditoria exportados como sensíveis independentemente. O comportamento documentado aqui é pontual em relação às fixtures incluídas, não uma garantia sobre qualquer conta HubSpot ao vivo.

Contrate-me

Eu construo servidores MCP e integrações de API que sobrevivem a uma revisão de código de desenvolvedor sênior: autenticação, tentativas, idempotência e trilhas de auditoria incluídas, não adicionadas depois. Portfólio e contato: https://amin-ale.github.io/portfolio-site · amin.ale.business@gmail.com.