McpServerPython

Servidor MCP local em Python para orquestrar uma jornada de empréstimo consignado usando tools expostas via HTTP.

O projeto funciona como um harness local: recebe chamadas padronizadas, valida o input, resolve contexto da jornada, executa uma tool registrada e devolve uma resposta com trace.

Status: POC / desenvolvimento local. Não está endurecido para produção.

Objetivo

Este repositório demonstra uma abordagem simples para expor capacidades de negócio como tools reutilizáveis em uma jornada conversacional.

O caso implementado é uma jornada de consignado com etapas como:

1. Buscar cliente por telefone.

2. Validar identidade por CPF.

3. Consultar elegibilidade.

4. Buscar ofertas de empréstimo.

5. Criar contrato.

6. Gerar link de formalização.

Related MCP server: Harpia Omnichannel MCP

Alterações da branch feature/melhorias

Esta branch mantém a natureza local/dev da POC, mas melhora alguns pontos de qualidade:

• server.py passou a centralizar a montagem de respostas de erro.

• server.py registra erro de tool em log estruturado via log_event("tool.error", ...).

• GET /healthz agora retorna também contagem de tools registradas e sessões em memória.

• A autenticação é tratada antes da execução da tool, retornando erro padronizado.

• IdempotencyStore passou a usar cópia defensiva em get e put.

• customers.py passou a respeitar valores booleanos explícitos vindos da API, inclusive false.

• eligibility.py passou a respeitar eligible=false explicitamente retornado pela API.

Arquitetura geral

Cliente / Agente / Orquestrador
        |
        v
+-----------------------------+
| FastAPI - MCP Local Server  |
| /healthz                    |
| /tools                      |
| /call                       |
+-------------+---------------+
              |
              v
+-----------------------------+
| Registry de Tools           |
| lookup_customer_by_phone    |
| resolve_identity_by_cpf     |
| check_eligibility           |
| get_loan_offers             |
| create_contract             |
| get_formalization_link      |
+-------------+---------------+
              |
              v
+-----------------------------+
| APIs de domínio             |
| Clientes                    |
| Elegibilidade               |
| Simulação                   |
| Formalização                |
+-----------------------------+

Stack

Principais tecnologias usadas:

• Python

• FastAPI

• Uvicorn

• Pydantic

• HTTPX

• LangChain / OpenAI, presentes nas dependências

• boto3, presente nas dependências

• SQLAlchemy, presente nas dependências

Estrutura do projeto

.
├── server.py
├── requirements.txt
└── mcp/
    ├── errors.py
    ├── http_clients.py
    ├── models.py
    ├── observability.py
    ├── registry.py
    ├── runtime.py
    ├── policies/
    │   ├── auth.py
    │   └── idempotency.py
    └── tools/
        ├── __init__.py
        ├── customers.py
        ├── eligibility.py
        ├── offers.py
        └── contracts.py

Responsabilidades principais

Endpoints

GET /healthz

Health check simples com informações úteis para debug local.

Resposta esperada:

{
  "ok": true,
  "tools": 7,
  "sessions": 0
}

Campos:

GET /tools

Lista todas as tools registradas no runtime, incluindo:

• nome;

• descrição;

• schema de input;

• schema de output.

POST /call

Executa uma tool registrada.

Formato base da requisição:

{
  "tool": "nome_da_tool",
  "input": {},
  "context": {
    "correlationId": "abc123456",
    "channel": "whatsapp",
    "subject": "cliente-ou-usuario"
  }
}

Formato base da resposta com sucesso:

{
  "ok": true,
  "output": {},
  "trace": {
    "correlationId": "abc123456",
    "tool": "nome_da_tool",
    "startedAt": 1710000000.0,
    "durationMs": 10
  }
}

Formato base da resposta com erro:

{
  "ok": false,
  "error": {
    "code": "BAD_REQUEST",
    "message": "Mensagem do erro",
    "retriable": false,
    "details": {}
  },
  "trace": {
    "correlationId": "abc123456",
    "tool": "nome_da_tool",
    "startedAt": 1710000000.0,
    "durationMs": 10
  }
}

Tools disponíveis

lookup_customer_by_phone

Busca cliente a partir do telefone.

Input:

{
  "phone": "+5511999999999"
}

Output:

{
  "found": true,
  "customerRef": "customer-123",
  "cpfRequired": true
}

Observação: a branch feature/melhorias corrige o parsing de cpfRequired para respeitar false explícito retornado pela API.

resolve_identity_by_cpf

Valida a identidade do cliente usando telefone e CPF.

Input:

{
  "phone": "+5511999999999",
  "cpf": "12345678901"
}

Output:

{
  "resolved": true,
  "customerRef": "customer-123",
  "cpfToken": "token-cpf"
}

Observação: a branch feature/melhorias corrige o parsing de resolved para respeitar false explícito retornado pela API.

get_identity_context

Retorna o contexto de identidade salvo na sessão da jornada.

Input:

{}

Output:

{
  "hasIdentity": true,
  "customerRef": "customer-123",
  "cpfToken": "token-cpf"
}

check_eligibility

Consulta elegibilidade do cliente para consignado.

Input:

{
  "customerRef": "customer-123",
  "channel": "whatsapp"
}

Output elegível:

{
  "eligible": true,
  "eligibilityId": "elig-123",
  "reasonCode": null
}

Output não elegível:

{
  "eligible": false,
  "eligibilityId": null,
  "reasonCode": "NOT_ELIGIBLE"
}

Observação: para o canal whatsapp, o fluxo exige identidade previamente resolvida. A branch feature/melhorias corrige o parsing para respeitar eligible=false explicitamente retornado pela API.

get_loan_offers

Busca ofertas de empréstimo com base no valor solicitado.

Input:

{
  "requestedAmount": 5000.0
}

Output:

{
  "offers": [
    {
      "id": "offer-123",
      "installment": 250.0,
      "termMonths": 24,
      "rateMonthly": 0.018,
      "totalCost": 6000.0
    }
  ]
}

create_contract

Cria um contrato pendente a partir de uma oferta.

Input:

{
  "offerId": "offer-123",
  "idempotencyKey": "journey-abc123-contract"
}

Output:

{
  "contractId": "contract-123"
}

A tool usa idempotência em memória para evitar duplicidade no mesmo runtime. Na branch feature/melhorias, o store faz cópia defensiva na gravação e leitura.

get_formalization_link

Gera ou busca o link de formalização do contrato.

Input:

{
  "contractId": "contract-123"
}

Output:

{
  "formalizationUrl": "https://exemplo.com/formalizacao/contract-123",
  "expiresAt": "2026-01-01T12:00:00Z"
}

Fluxo sugerido de chamada

1. lookup_customer_by_phone
2. resolve_identity_by_cpf
3. check_eligibility
4. get_loan_offers
5. create_contract
6. get_formalization_link

Todas as chamadas devem reutilizar o mesmo correlationId para preservar o contexto da jornada.

Variáveis de ambiente

O projeto usa variáveis por prefixo para configurar os serviços externos.

Clientes

CLIENTES_BASE_URL=https://localhost:7287
CLIENTES_VERIFY_SSL=false
CLIENTES_TIMEOUT=10
CLIENTES_AUTH_PATH=/api/auth/token
CLIENTES_CLIENT_ID=mcp
CLIENTES_CLIENT_SECRET=mcp-secret

Elegibilidade

ELEG_BASE_URL=https://localhost:7184
ELEG_VERIFY_SSL=false
ELEG_TIMEOUT=10
ELEG_AUTH_PATH=/api/auth/token
ELEG_CLIENT_ID=mcp
ELEG_CLIENT_SECRET=mcp-secret

Simulação

SIM_BASE_URL=https://localhost:7173
SIM_VERIFY_SSL=false
SIM_TIMEOUT=10
SIM_AUTH_PATH=/api/auth/token
SIM_CLIENT_ID=mcp
SIM_CLIENT_SECRET=mcp-secret

Formalização

FORM_BASE_URL=https://localhost:7131
FORM_VERIFY_SSL=false
FORM_TIMEOUT=10
FORM_AUTH_PATH=/api/auth/token
FORM_CLIENT_ID=mcp
FORM_CLIENT_SECRET=mcp-secret

Como executar localmente

1. Criar ambiente virtual

python -m venv .venv

2. Ativar ambiente virtual

Windows:

.venv\Scripts\activate

Linux/macOS:

source .venv/bin/activate

3. Instalar dependências

pip install -r requirements.txt

4. Subir o servidor

uvicorn server:app --reload

A API ficará disponível em:

http://127.0.0.1:8000

Documentação automática do FastAPI:

http://127.0.0.1:8000/docs

Exemplos com cURL

Health check

curl http://127.0.0.1:8000/healthz

Listar tools

curl http://127.0.0.1:8000/tools

Chamar lookup_customer_by_phone

curl -X POST http://127.0.0.1:8000/call \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer local-dev-token" \
  -d '{
    "tool": "lookup_customer_by_phone",
    "input": {
      "phone": "+5511999999999"
    },
    "context": {
      "correlationId": "journey-abc123",
      "channel": "whatsapp",
      "subject": "user-123"
    }
  }'

Tratamento de erros

As tools usam erros padronizados com os seguintes campos:

{
  "code": "UPSTREAM_UNAVAILABLE",
  "message": "Descrição do erro",
  "retriable": true,
  "details": {}
}

Códigos comuns:

Na branch feature/melhorias, falhas de autenticação e erros de execução das tools também passam pelo envelope padronizado de erro.

Limitações atuais

Este projeto ainda tem características de desenvolvimento local:

• Autenticação simplificada: aceita Bearer token sem validação real.

• Permite execução anônima quando não há header Authorization.

• Sessão em memória, indexada por correlationId.

• Idempotência em memória.

• Sem persistência externa.

• Sem testes automatizados documentados.

• Sem Dockerfile ou pipeline de CI/CD documentado.

• Defaults locais para as APIs de domínio.

Melhorias recomendadas

Antes de levar para produção, considerar:

• Validar JWT real, por exemplo Azure AD, Cognito ou outro IdP corporativo.

• Persistir sessão em Redis, DynamoDB ou store equivalente.

• Persistir idempotência fora do processo.

• Adicionar testes unitários e de contrato para cada tool.

• Adicionar logs estruturados com correlação ponta a ponta.

• Adicionar métricas de latência, erro e uso por tool.

• Implementar rate limit e políticas por canal.

• Remover defaults inseguros de ambiente.

• Criar Dockerfile e pipeline de build/deploy.

• Documentar contratos das APIs externas.

Natureza do projeto

Este servidor não é uma implementação completa de plataforma MCP corporativa. Ele é uma POC objetiva para demonstrar como encapsular capacidades de negócio como tools executáveis por um agente ou orquestrador conversacional.

O foco está em clareza arquitetural, separação de responsabilidades e simulação de uma jornada bancária real.