MCP Avançado: sampling, transports, StreamableHTTP e produção

# Padrão de error handling para tools MCP em produção:

from mcp.server import Server
from mcp import types
import logging, traceback

logger = logging.getLogger(__name__)
app = Server("servidor-producao")

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
    try:
        if name == "buscar_cliente":
            return await _buscar_cliente(arguments)
        raise ValueError(f"Tool desconhecida: {name}")

    except ClienteNaoEncontradoError as e:
        # Erro de negócio — Claude pode tentar alternativa
        logger.info(f"Cliente não encontrado: {arguments}")
        return [types.TextContent(
            type="text",
            text=json.dumps({
                "erro": "cliente_nao_encontrado",
                "mensagem": str(e),
                "sugestao": "Tente buscar por CPF ou email em vez de nome"
            }),
            isError=True  # ← Claude recebe e adapta sua resposta
        )]

    except PermissaoNegadaError as e:
        # Erro de autorização — deve ser informado ao usuário
        logger.warning(f"Acesso negado: {name} com {arguments}")
        return [types.TextContent(
            type="text",
            text=json.dumps({
                "erro": "permissao_negada",
                "recurso": str(e),
                "acao": "Solicite ao administrador acesso ao recurso"
            }),
            isError=True
        )]

    except Exception as e:
        # Erro inesperado — log completo interno, mensagem genérica ao Claude
        logger.error(
            f"Erro inesperado em tool '{name}'",
            extra={"arguments": arguments, "error": str(e)},
            exc_info=True
        )
        return [types.TextContent(
            type="text",
            text=json.dumps({
                "erro": "erro_interno",
                "mensagem": "Erro interno ao processar a solicitação. Tente novamente."
            }),
            isError=True
        )]

# Resources estáticos vs dinâmicos:

# ─── Resources estáticos (lista fixa) ────────────────────
@app.list_resources()
async def list_resources() -> list[types.Resource]:
    return [
        types.Resource(
            uri="config://database/schema",
            name="Schema do banco de dados",
            description="Estrutura completa das tabelas e relacionamentos",
            mimeType="application/json"
        ),
        types.Resource(
            uri="docs://api/reference",
            name="Referência da API interna",
            description="Documentação dos endpoints internos",
            mimeType="text/markdown"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri == "config://database/schema":
        return json.dumps(get_db_schema())  # função sua
    if uri == "docs://api/reference":
        return pathlib.Path("docs/api.md").read_text()
    raise ValueError(f"Resource desconhecido: {uri}")

# ─── Resource templates (URI dinâmico) ────────────────────
# Para expor coleções de dados addressable por ID:

@app.list_resource_templates()
async def list_resource_templates() -> list[types.ResourceTemplate]:
    return [
        types.ResourceTemplate(
            uriTemplate="cliente://{cliente_id}/historico",
            name="Histórico de compras do cliente",
            description="Lê o histórico completo de pedidos de um cliente específico. "
                        "Use {cliente_id} como o ID do cliente (ex: cliente://C001/historico)",
            mimeType="application/json"
        )
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    # Processa tanto URIs fixos quanto de templates
    if uri.startswith("cliente://") and uri.endswith("/historico"):
        cliente_id = uri.replace("cliente://", "").replace("/historico", "")
        historico = get_historico_cliente(cliente_id)  # função sua
        return json.dumps(historico)
    # ... outros casos

# Sampling permite que o servidor peça ao host para chamar o LLM
# Caso de uso: usar Claude para processar/sumarizar antes de retornar

from mcp.server import Server
from mcp.types import SamplingMessage, TextContent

app = Server("servidor-com-sampling")

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "analisar_log":
        # 1. Busca os logs brutos
        logs_brutos = get_logs(arguments["periodo"])  # pode ser 10MB de logs

        # 2. Usa sampling para pedir ao host que Claude sumarize
        # O host tem as credenciais de API — o servidor não precisa ter
        summary_response = await app.request_context.session.create_message(
            messages=[
                SamplingMessage(
                    role="user",
                    content=TextContent(
                        type="text",
                        text=f"""Analise estes logs e retorne:
1. Top 5 erros por frequência com contagem
2. Padrão de horário com mais erros (hora do dia)
3. Serviços mais afetados

LOGS:
{logs_brutos[:50000]}  # limita para caber no contexto"""
                    )
                )
            ],
            max_tokens=1000,
            system="Você é um analisador de logs. Responda em JSON estruturado."
        )

        # 3. Retorna o sumário — não os 10MB de logs brutos
        return [TextContent(
            type="text",
            text=summary_response.content.text
        )]

# Padrão de autenticação para servidores MCP que acessam dados sensíveis:

import os, hmac, hashlib
from functools import wraps

# ─── 1. Credenciais via env vars ──────────────────────────
# Nunca hardcode. O mcp.json passa via env:
# "env": {"DATABASE_URL": "${DATABASE_URL}", "API_KEY": "${MY_API_KEY}"}

DATABASE_URL = os.environ["DATABASE_URL"]
INTERNAL_API_KEY = os.environ["INTERNAL_API_KEY"]

# ─── 2. Validação de argumentos (não confie no input) ─────
def validar_cliente_id(cliente_id: str) -> bool:
    """IDs de cliente sempre são alfanuméricos, 6-20 chars."""
    return bool(re.match(r'^[A-Z0-9]{6,20}$', cliente_id))

async def _buscar_cliente(arguments: dict) -> list:
    cliente_id = arguments.get("cliente_id", "")

    if not validar_cliente_id(cliente_id):
        return [TextContent(
            type="text",
            text=json.dumps({"erro": "ID de cliente inválido", "recebido": cliente_id}),
            isError=True
        )]

    # Agora é seguro usar o ID
    cliente = await db.buscar(cliente_id)
    return [TextContent(type="text", text=json.dumps(cliente))]

# ─── 3. Audit logging ─────────────────────────────────────
import structlog

audit_log = structlog.get_logger("mcp.audit")

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list:
    # Log toda chamada de tool com dados sensíveis
    audit_log.info(
        "tool_called",
        tool=name,
        # Redact dados sensíveis antes de logar
        args_redacted={k: "***" if k in ("senha", "token", "cpf") else v
                       for k, v in arguments.items()}
    )
    # ... resto da lógica

# ─── 4. Rate limiting por sessão ──────────────────────────
from collections import defaultdict
import time

_call_counts: dict[str, list[float]] = defaultdict(list)
RATE_LIMIT = 50  # chamadas por minuto por sessão

def check_rate_limit(session_id: str) -> bool:
    now = time.time()
    # Remove chamadas > 60s atrás
    _call_counts[session_id] = [t for t in _call_counts[session_id] if now - t < 60]
    if len(_call_counts[session_id]) >= RATE_LIMIT:
        return False
    _call_counts[session_id].append(now)
    return True

# Opções de deploy para servidores MCP:

# ─── 1. Local via stdio (mais comum para dev) ──────────────
# .claude/mcp.json:
{
  "mcpServers": {
    "meu-servidor": {
      "command": "python",
      "args": ["mcp_server/main.py"]
    }
  }
}

# ─── 2. Remote via SSE (para times, SaaS) ─────────────────
# O servidor roda como serviço HTTP com SSE endpoint
# mcp.json para servidor remoto:
{
  "mcpServers": {
    "meu-servidor-remoto": {
      "url": "https://meu-servidor.empresa.com/mcp",
      "headers": {
        "Authorization": "Bearer ${MCP_TOKEN}"
      }
    }
  }
}

# Servidor SSE com FastAPI:
from fastapi import FastAPI
from mcp.server.sse import SseServerTransport

app_web = FastAPI()
mcp_app = Server("meu-servidor")
sse = SseServerTransport("/messages/")

@app_web.get("/sse")
async def sse_endpoint(request: Request):
    async with sse.connect_sse(request.scope, request.receive, request._send) as streams:
        await mcp_app.run(streams[0], streams[1], mcp_app.create_initialization_options())

# ─── 3. npx para servidores Node.js (distribuição fácil) ───
# O usuário configura assim sem instalar nada:
{
  "mcpServers": {
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"]
    }
  }
}
# npx baixa e roda automaticamente — zero instalação manual