Template padrão para criação de novos MCP (Model Context Protocol) servers no barramento.
cp -r src/mcp-servers/MCP_TEMPLATE src/mcp-servers/meu-novo-mcp
cd src/mcp-servers/meu-novo-mcp# Renomear arquivos
mv main_template.py main.py
mv requirements_template.txt requirements.txt
# Editar configurações
vim config/settings.py
vim docker-compose.yml# Editar arquivo principal
vim main.py
# Adicionar testes
vim tests/test_main.py
# Configurar observabilidade
vim config/observability.pymeu-novo-mcp/
├── main.py # Servidor principal FastAPI
├── requirements.txt # Dependências Python
├── Dockerfile # Container do serviço
├── README.md # Documentação específica
├── config/ # Configurações
│ ├── settings.py # Configurações principais
│ ├── observability.py # Métricas e logging
│ └── governance.py # Governança GitOps
├── tests/ # Testes automatizados
│ ├── test_main.py # Testes unitários
│ ├── test_integration.py # Testes de integração
│ └── test_performance.py # Testes de performance
├── logs/ # Logs do serviço
└── docs/ # Documentação específica
└── api.md # Documentação da API
# config/settings.py
SERVICE_PORT = 800X # Substituir X por número único# main.py
app = FastAPI(
title="Meu Novo MCP",
description="Descrição do serviço",
version="1.0.0"
)# docker-compose.yml
services:
meu-novo-mcp:
build: .
ports:
- "800X:800X"
environment:
- SERVICE_PORT=800X# config/observability.py
from prometheus_client import Counter, Histogram, Gauge
# Métricas obrigatórias
request_counter = Counter(
'meu_mcp_requests_total',
'Total requests',
['method', 'endpoint', 'status']
)
request_duration = Histogram(
'meu_mcp_request_duration_seconds',
'Request duration',
['method', 'endpoint']
)
active_connections = Gauge(
'meu_mcp_active_connections',
'Active connections'
)# Exemplo de uso
observability.log_structured("info", "operation_completed",
operation="process_data",
duration=1.23,
records_processed=100)# Criar snapshot
snapshot_id = governance.create_mcp_snapshot(config)
# Rollback se necessário
governance.rollback_to_snapshot(snapshot_id)- Verificação de tamanho de arquivos
- Validação de tipos de arquivo
- Verificação de secrets
- Testes obrigatórios
- Unitários: 80%
- Integração: 60%
- Performance: Crítico
# Todos os testes
pytest tests/ -v --cov=.
# Apenas unitários
pytest tests/test_main.py -v
# Com relatório de cobertura
pytest tests/ --cov=. --cov-report=html- Build: Construção da imagem Docker
- Test: Execução de todos os testes
- Security: Scan de vulnerabilidades
- Deploy: Deploy automático (dev/staging)
- Approval: Aprovação manual (produção)
- dev: Deploy automático
- staging: Deploy com aprovação
- prod: Deploy com múltiplas aprovações
- Descrição do serviço
- Endpoints disponíveis
- Exemplos de uso
- Troubleshooting
- OpenAPI/Swagger automático
- Exemplos de request/response
- Códigos de erro
- Testes passando (100%)
- Cobertura mínima atingida
- Documentação atualizada
- Configurações validadas
- Observabilidade funcionando
- Governança configurada
- Security scan limpo
- Health check funcionando
- Métricas sendo coletadas
- Logs estruturados
- Performance dentro do esperado
- Erro de porta: Verificar se porta está livre
- Dependências: Verificar requirements.txt
- Configuração: Validar variáveis de ambiente
- Testes: Executar pytest com -v para detalhes
# Ver logs do container
docker-compose logs meu-novo-mcp
# Logs em tempo real
docker-compose logs -f meu-novo-mcp
# Logs estruturados
tail -f logs/meu-novo-mcp.log | jq .@app.post("/process")
async def process_data(request: ProcessRequest):
"""Processa dados conforme especificação"""
# Log da operação
observability.log_structured("info", "process_started",
request_id=request.id)
try:
# Processar dados
result = await process_service.handle(request)
# Métricas de sucesso
request_counter.labels(
method="POST",
endpoint="/process",
status="success"
).inc()
return result
except Exception as e:
# Métricas de erro
request_counter.labels(
method="POST",
endpoint="/process",
status="error"
).inc()
# Log do erro
observability.log_structured("error", "process_failed",
error=str(e))
raise HTTPException(status_code=500, detail=str(e))🎯 Este template garante que todos os MCPs sigam os mesmos padrões de qualidade, observabilidade e governança do barramento!