mcp-toolkit

Servidor MCP (Model Context Protocol) em Python que expõe, para agentes de IA, um conjunto de ferramentas tipadas sobre sistemas de negócio — CRM, base analítica de clientes e publicação de páginas HTML — com autenticação na borda, contratos testados e modo demo 100% offline.

Caso de uso real

Em uma empresa do setor de energia, diversos agentes de IA — um atendimento por voz para clientes e assistentes internos usados pelos times de operação — precisavam das mesmas capacidades: consultar e atualizar contatos no CRM, localizar clientes por telefone, CPF ou código de instalação numa base analítica, e publicar relatórios e apresentações em HTML sob um link estável que pudesse ser enviado a um cliente ou colado numa conversa.

Cada agente tinha nascido com integrações próprias: três formas diferentes de chamar o CRM, três formas de tratar erro, três lugares para corrigir quando um campo mudava. Regras importantes (quais campos um agente pode editar, como mascarar PII nos logs, como validar um slug) estavam duplicadas ou simplesmente ausentes em alguma das cópias.

O padrão aplicado foi centralizar essas capacidades num único servidor MCP, com contratos tipados e prefixos de nome por sistema (crm_*, db_*, pages_*), autenticação por token na borda e uma suíte de testes que trava o contrato exposto aos agentes. Qualquer cliente MCP — Claude, IDEs, orquestradores — passou a usar as mesmas ferramentas, e a equipe passou a evoluir uma única integração por sistema, com revisão de código e testes, em vez de três.

Este repositório é uma reimplementação genérica desse padrão, com dados sintéticos e integrações fake em memória. Nenhum nome, endpoint, schema ou dado real de empresa ou cliente está presente; as implementações "reais" (HttpCRMClient, PostgresCustomerRepository) apontam para APIs e schemas genéricos e servem como ponto de adaptação.

Related MCP server: namakan-mcp-crm

O que este projeto demonstra

• Servidor MCP com o SDK oficial (mcp 2.x) — tools, resource template e prompt registrados via MCPServer, transporte streamable-http, saída estruturada (structuredContent) derivada de modelos Pydantic.

• Composição ASGI — o app MCP é montado dentro de um Starlette próprio que também serve as páginas publicadas (GET /p/<slug>/) e um health check, com o lifespan do session manager do SDK integrado ao lifespan da aplicação.

• Autenticação na borda em middleware ASGI puro: Authorization: Bearer validado com hmac.compare_digest, suporte a múltiplos tokens (rotação) sem early-exit.

• Rate limiting por token (token bucket em memória) com Retry-After.

• Integrações plugáveis por Protocol (CRMClient, CustomerRepository): fakes em memória para demo/testes, cliente HTTP com retry e backoff exponencial, repositório Postgres com queries parametrizadas (psycopg 3). APP_MODE=demo|prod escolhe a implementação.

• Validação de domínio — allowlist de campos editáveis no CRM, detecção de tipo de identificador (telefone, CPF com dígito verificador, código de instalação), regex de slug, sanitização de HTML (scripts, iframes, handlers on*, URLs javascript:), limites de tamanho.

• Observabilidade básica — logging JSON estruturado com request_id propagado por contextvars, uma linha de acesso por requisição e mascaramento automático de CPF/telefone.

• Testes em três camadas (unit, integration, contract) 100% offline, incluindo o cliente MCP oficial falando com o app através do transporte streamable-http real em processo, e snapshot JSON do contrato (nomes, schemas de entrada/saída, anotações).

• Empacotamento de produção — uv, Dockerfile multi-stage sem uv na imagem final e usuário não-root, docker-compose com Postgres opcional por profile, CI com ruff + pytest + gate de cobertura.

Arquitetura

flowchart LR
    subgraph Clients["Clientes MCP"]
        A[Agente de voz]
        B[Claude / IDE]
        C[scripts/demo_client.py]
    end

    subgraph App["Starlette (ASGI)"]
        direction TB
        M1[RequestIdMiddleware] --> M2[BearerAuthMiddleware] --> M3[RateLimitMiddleware]
        M3 --> R1["/mcp  (streamable-http)"]
        M3 --> R2["/p/{slug}/  (páginas públicas)"]
        M3 --> R3["/healthz"]
        R1 --> S[MCPServer]
    end

    subgraph Tools["tools/"]
        T1["crm_* "] & T2["db_*"] & T3["pages_*"]
        T4["resource://customers/{id}"]
        T5["prompt summarize_call"]
    end

    subgraph Services["services/"]
        SV1[CRMService<br/>allowlist, normalização]
        SV2[CustomerService<br/>detecção de identificador]
        SV3[PagesService<br/>slug, sanitize, FS atômico]
    end

    subgraph Clients2["clients/ (Protocols)"]
        C1[FakeCRMClient] -.-> P1((CRMClient))
        C2[HttpCRMClient] -.-> P1
        C3[FakeCustomerRepository] -.-> P2((CustomerRepository))
        C4[PostgresCustomerRepository] -.-> P2
    end

    A & B & C -->|Bearer token| M1
    S --> T1 & T2 & T3 & T4 & T5
    T1 --> SV1 --> P1
    T2 & T4 --> SV2 --> P2
    T3 --> SV3 --> FS[(published/<slug>/index.html)]
    R2 --> FS

Componentes

Fluxo de uma requisição (tools/call db_lookup_customer)

1. O cliente MCP faz POST /mcp com Authorization: Bearer <token> e um JSON-RPC tools/call.

2. RequestIdMiddleware lê ou gera X-Request-ID, coloca-o num contextvar (todos os logs da requisição o carregam) e devolve-o no header de resposta.

3. BearerAuthMiddleware compara o token com todos os configurados via hmac.compare_digest; sem match, responde 401 e a requisição nunca chega ao SDK.

4. RateLimitMiddleware consome um token do bucket daquele credential; vazio, responde 429 com Retry-After.

5. O app streamable-http do SDK valida o envelope JSON-RPC, resolve a tool e valida os argumentos contra o schema Pydantic (query: str).

6. db_lookup_customer chama CustomerService.lookup, que detecta o tipo do identificador ("+55 11 90000-0003" → telefone, normalizado para 5511900000003) e consulta CustomerRepository.find_by_phone (fake em demo, Postgres em prod).

7. O CustomerLookupResult (Pydantic) é serializado pelo SDK como structuredContent e também como texto; erros de domínio viram isError: true com mensagem acionável em vez de exceção.

8. A linha de acesso é logada em JSON com request_id, método, caminho, status e duração.

Decisões técnicas

Como executar

Pré-requisitos

• Python 3.12+ e uv (o projeto fixa .python-version em 3.12; uv baixa o interpretador se necessário).

• Docker e Docker Compose (opcional).

Instalação e modo demo (sem credenciais)

git clone https://github.com/romeuow/mcp-toolkit.git
cd mcp-toolkit
make install           # uv sync
cp .env.example .env   # opcional; os defaults já funcionam em demo
make dev               # uvicorn em http://127.0.0.1:8000 com reload

O modo demo (APP_MODE=demo, padrão) usa FakeCRMClient e FakeCustomerRepository com dez clientes sintéticos (Ana Exemplo … Joao Exemplo, CPFs 000.000.000-NN, telefones +55 11 90000-00NN) e aceita o token demo-token. Nenhuma rede externa é usada.

Variáveis relevantes (ver .env.example para todas):

Rodar com Docker

make up                 # build + sobe o serviço mcp em modo demo
make up PROFILE=prod    # também sobe o Postgres (scripts/init_db.sql é aplicado no primeiro boot)
make down

Em prod, defina APP_MODE=prod, MCP_AUTH_TOKENS, CRM_BASE_URL, CRM_API_KEY e DATABASE_URL no .env (o compose já aponta DATABASE_URL para o serviço postgres).

Exemplos com curl

Health check (público):

curl -s http://localhost:8000/healthz
# {"status":"ok","mode":"demo","version":"0.1.0"}

Sem token, o endpoint MCP recusa:

curl -s -o /dev/null -w "%{http_code}\n" -X POST http://localhost:8000/mcp \
  -H 'Content-Type: application/json' -d '{}'
# 401

Listar tools:

curl -s -X POST http://localhost:8000/mcp \
  -H 'Authorization: Bearer demo-token' \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}' | jq '.result.tools[].name'
# "crm_get_contact"
# "crm_update_contact"
# "crm_register_call"
# "db_lookup_customer"
# "db_open_tickets"
# "pages_publish"
# "pages_list"

Buscar cliente por telefone (detecção automática do tipo):

curl -s -X POST http://localhost:8000/mcp \
  -H 'Authorization: Bearer demo-token' \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"db_lookup_customer","arguments":{"query":"+55 11 90000-0003"}}}' \
  | jq '.result.structuredContent'
# {
#   "query": "+55 11 90000-0003",
#   "detected_kind": "phone",
#   "normalized_query": "5511900000003",
#   "customer": { "id": "C003", "name": "Carla Exemplo", "installation_code": "INST-0000003", ... },
#   "found": true
# }

Publicar uma página (o <script> é removido e reportado) e abri-la:

curl -s -X POST http://localhost:8000/mcp \
  -H 'Authorization: Bearer demo-token' \
  -H 'Content-Type: application/json' \
  -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"pages_publish","arguments":{"slug":"relatorio-q3","html":"<h1>Relatorio Q3</h1><script>alert(1)</script>"}}}' \
  | jq '.result.structuredContent'
# {
#   "slug": "relatorio-q3",
#   "url": "http://localhost:8000/p/relatorio-q3/",
#   "bytes_written": 21,
#   "sanitized": true,
#   "removed": ["element <script>"]
# }

curl -s http://localhost:8000/p/relatorio-q3/
# <h1>Relatorio Q3</h1>

Tentar editar um campo fora da allowlist retorna erro de tool (não exceção):

# ... "name":"crm_update_contact","arguments":{"contact_id":"crm-1001","fields":{"lifecycle_stage":"x"}}
# .result.isError == true
# "Error executing tool crm_update_contact: validation_error: fields not editable: lifecycle_stage.
#  Allowed: city, consent_marketing, email, first_name, last_name, notes, phone"

Cliente MCP de exemplo

scripts/demo_client.py usa o cliente do SDK sobre streamable-http, lista as tools e chama db_lookup_customer:

make demo-client
# ou
MCP_URL=http://localhost:8000/mcp MCP_TOKEN=demo-token \
  uv run python scripts/demo_client.py "000.000.000-07"

Saída (resumida):

Connected to http://localhost:8000/mcp (server: mcp-toolkit)

Tools:
  - crm_get_contact: Find a CRM contact by phone, e-mail or document (CPF).
  - ...
Calling db_lookup_customer(query='000.000.000-07') ...
{ "detected_kind": "document", "customer": { "id": "C007", "name": "Gabriela Exemplo", ... }, "found": true }

Para usar com um cliente MCP genérico, configure a URL http://localhost:8000/mcp com o header Authorization: Bearer demo-token.

Testes

make test
# ou
uv run pytest -q --cov=src --cov-report=term-missing

Todos os testes rodam offline, sem chaves nem rede. Organização:

Estratégia de fakes: as integrações externas são substituídas por implementações completas em memória (FakeCRMClient, FakeCustomerRepository) que respeitam os mesmos Protocols das reais; as reais são testadas com transporte HTTP mockado e conexão SQL fake, sem subir serviços.

Cobertura atual medida (pytest-cov, branch coverage ligado):

146 passed in 1.9s
TOTAL  1086 stmts  8 miss  248 branches  12 partial  99%

Segurança

Autenticação na borda. POST /mcp exige Authorization: Bearer <token>. A comparação usa hmac.compare_digest e percorre todos os tokens configurados sem retorno antecipado, para não vazar por tempo qual token bateu. Múltiplos tokens permitem rotação sem janela de indisponibilidade. Em prod, a ausência de MCP_AUTH_TOKENS impede o boot (validação em Settings).

Validação de entrada. Argumentos de tool são validados pelo SDK contra schemas Pydantic (Literal para enumerações, tipos estritos). Acima disso, a camada de serviços aplica regras de domínio: allowlist de campos editáveis no CRM (qualquer chave fora dela rejeita a chamada inteira), limites de tamanho (summary ≤ 4000 chars, HTML ≤ 2 MiB), normalização de telefone/e-mail/CPF.

Páginas publicadas. O slug é validado por regex estrita (minúsculas, dígitos, hífens, 3–64) e o caminho final é resolvido e confinado ao diretório published/ (bloqueia path traversal mesmo que a regex falhasse). O HTML passa por um sanitizador baseado em html.parser que remove script, iframe, object, embed, applet, noscript, base, atributos on*, URLs com javascript:/vbscript:/data:text/html (inclusive com espaços e quebras de linha intercalados), <meta http-equiv="refresh">, comentários e instruções de processamento. A escrita é atômica (tmp + replace). Slugs reservados (mcp, healthz, p, ...) são recusados.

Segredos. Lidos apenas de variáveis de ambiente via pydantic-settings; .env está no .gitignore e .env.example só tem placeholders. Tokens nunca são logados: o rate limiter e os logs usam um hash SHA-256 truncado do token.

PII em logs. O formatter JSON mascara CPF (***.***.***-07) e telefones (***-0003) em mensagens, campos extras (recursivamente) e tracebacks. Os serviços logam apenas metadados (kind, found, by), nunca o identificador consultado.

Superfícies consideradas. Agente em loop esgotando integrações a jusante (rate limit por token); XSS via páginas publicadas (sanitização + Content-Type: text/html só para arquivos dentro do diretório confinado); SQL injection (queries parametrizadas, nunca interpolação); escrita indevida no CRM (allowlist); DNS rebinding do transporte (delegado ao reverse proxy — ver abaixo); imagem Docker rodando como usuário não-root e sem ferramentas de build.

O que NÃO está coberto (honestamente).

• Não há autorização por escopo: qualquer token válido acessa todas as tools. Perfis por token (ex.: agente de voz só lê) são roadmap.

• Tokens são estáticos; não há OAuth 2.1, expiração nem revogação individual além de remover o token da configuração e reiniciar.

• O sanitizador é uma lista de bloqueio, não uma lista de permissão: CSS inline e <style> são mantidos integralmente (incluindo url() em CSS), e <link rel=stylesheet>/<form> passam. É adequado para HTML gerado por agentes sob controle da própria empresa, não para conteúdo de terceiros hostis. Páginas em /p/ são públicas por design (link estável), sem CSP própria.

• A proteção contra DNS rebinding do SDK está desligada no app (TransportSecuritySettings), pois a validação de Host/Origin é esperada no reverse proxy ou ingress em produção.

• Rate limiting e cache de buckets são por processo; múltiplas réplicas multiplicam o limite.

• Não há auditoria persistente das mutações (crm_update_contact, crm_register_call, pages_publish) além dos logs estruturados.

• HttpCRMClient e PostgresCustomerRepository são testados com fakes/mocks; não há teste de integração contra um Postgres ou CRM reais neste repositório.

Estrutura do projeto

mcp-toolkit/
├── src/mcp_toolkit/
│   ├── __main__.py            # python -m mcp_toolkit -> uvicorn
│   ├── server.py              # build_mcp_server() + create_app(): Starlette, rotas, lifespan
│   ├── middleware.py          # RequestId, BearerAuth (compare_digest), RateLimit (token bucket)
│   ├── settings.py            # pydantic-settings; APP_MODE demo|prod; validação de prod
│   ├── container.py           # wiring: escolhe fakes ou reais e monta os serviços
│   ├── logging.py             # JSON formatter, request_id contextvar, mask_pii()
│   ├── schemas/               # Pydantic: Contact, Customer, Ticket, PublishResult, ...
│   ├── clients/
│   │   ├── protocols.py       # CRMClient, CustomerRepository (+ NotFoundError, UpstreamError)
│   │   ├── fake_crm.py        # FakeCRMClient (10 contatos sintéticos, mutações em memória)
│   │   ├── fake_customers.py  # FakeCustomerRepository (10 clientes, 5 tickets)
│   │   ├── http_crm.py        # HttpCRMClient: httpx, retry com backoff, 404 -> None/NotFound
│   │   └── postgres_customers.py  # psycopg 3, queries parametrizadas, factory injetável
│   ├── services/
│   │   ├── identifiers.py     # detect_identifier(), is_valid_cpf(), normalizações
│   │   ├── sanitize.py        # sanitize_html() baseado em html.parser
│   │   ├── crm_service.py     # allowlist EDITABLE_CONTACT_FIELDS, coerção, limites
│   │   ├── customer_service.py
│   │   ├── pages_service.py   # validate_slug(), publicação atômica, listagem, leitura segura
│   │   └── errors.py          # DomainError, InvalidInputError
│   └── tools/
│       ├── _base.py           # register_tool(): cleandoc + mapeamento de erros -> ToolError
│       ├── crm.py db.py pages.py   # crm_*, db_*, pages_*
│       ├── resources.py       # resource://customers/{id}
│       └── prompts.py         # summarize_call
├── scripts/
│   ├── demo_client.py         # cliente MCP do SDK: lista tools e chama db_lookup_customer
│   └── init_db.sql            # schema + seed sintético para o Postgres (modo prod)
├── tests/
│   ├── conftest.py            # settings temporárias, container, servidor, app, cliente httpx
│   ├── unit/  integration/  contract/
│   └── snapshots/tools.json   # contrato MCP versionado
├── .github/workflows/ci.yml   # ruff + pytest (3.12 e 3.13) com --cov-fail-under=80
├── Dockerfile                 # multi-stage: builder com uv, runtime slim non-root sem uv
├── docker-compose.yaml        # serviço mcp + postgres (profile prod)
├── Makefile                   # install, dev, test, lint, format, up, down, demo-client
├── pyproject.toml             # deps, grupo dev, ruff, pytest (anyio auto), coverage
└── .env.example

Roadmap / limitações conhecidas

• Escopos por token (leitura vs. escrita) e, em seguida, OAuth 2.1 via TokenVerifier do SDK para clientes interativos.

• Allowlist de HTML (tags/atributos permitidos) e CSP nas páginas servidas, substituindo a lista de bloqueio atual.

• Pool de conexões (psycopg_pool) e timeouts explícitos no repositório Postgres.

• Rate limiting distribuído (Redis) quando houver mais de uma réplica.

• Auditoria das mutações numa tabela própria, com request_id e identificador do token.

• Métricas (Prometheus/OpenTelemetry) além dos logs; o SDK 2.x já expõe hooks OTel.

• Paginação em pages_list e db_open_tickets para volumes maiores.

• Testes de integração opcionais contra Postgres real via testcontainers, fora do caminho padrão do CI.

• A detecção de identificador para 11 dígitos sem formatação usa o dígito verificador do CPF como desempate; um telefone cujos dígitos formem um CPF válido seria classificado como documento. Clientes que conhecem o tipo devem usar crm_get_contact(by=...).

Licença

MIT — veja LICENSE. Copyright (c) 2026 Romeu Oliveira.