Telemetria de tool calls em agentes com MCP

Um coding agent termina a revisão com sucesso, mas deixa perguntas bem menos confortáveis que o resumo final: quantas vezes leu os mesmos arquivos, quais comandos falharam e quanto tempo ficou esperando Bash em vez de inspecionar código?

O MCP não entrega essa conta pronta. Ele padroniza a chamada da ferramenta, não a contabilidade do seu harness. Para responder às perguntas, vamos observar tools/call, gravar um events.jsonl local e consultá-lo com DuckDB. O resultado separa volume de chamadas, latência e bytes por ferramenta, sem fingir que esses números são tokens ou dólares.

Versões e limites do exemplo

A base usada aqui é:

• MCP 2026-07-28;
• Python SDK mcp 2.1.1;
• Python 3.10 ou superior, com 3.11 como uma escolha válida para a execução;
• DuckDB 1.5.5 como auxiliar de consulta.

O SDK e o DuckDB exigem Python 3.10 ou superior. Não existe uma matriz conjunta entre eles: são dois requisitos independentes que se encontram nesse piso. A especificação de tools do MCP define tools/list, tools/call e os resultados; o Python SDK implementa o servidor. DuckDB não faz parte do protocolo.

Instale as dependências em um ambiente virtual:

python3.11 -m venv .venv
. .venv/bin/activate
python -m pip install "mcp[cli]==2.1.1" "duckdb==1.5.5"

No MCP 2026-07-28, o caminho moderno de descoberta é server/discover. Para um cliente desse SDK, fixe mode="2026-07-28" quando não quiser negociar pelo modo automático. O antigo initialize pertence à geração 2025-11-25; misturar os dois torna a telemetria confusa antes mesmo de ela começar a ajudar.

O que uma chamada de ferramenta realmente informa

Um cliente invoca uma ferramenta com o método JSON-RPC tools/call e envia name e arguments. Há dois canais de falha que precisam continuar separados:

• ToolError é erro de execução da ferramenta. A chamada JSON-RPC produz um resultado com isError: true e conteúdo que o modelo pode usar para corrigir a próxima tentativa.
• MCPError é erro de protocolo, como parâmetros inválidos recusados pelo host. Não há resultado de ferramenta nem isError para o modelo interpretar.
• uma exceção não capturada também vira um resultado com is_error=True, mas o cliente recebe apenas uma mensagem genérica; o traceback permanece no log do servidor.

Retornar uma string como "erro: livro ausente" não substitui ToolError: para o protocolo, isso continua sendo sucesso. É o tipo de sucesso que melhora o dashboard e piora todo o resto.

A documentação oficial de middleware usa a assinatura assíncrona (ctx, call_next). Essa API é provisória e pode mudar em uma versão 2.x menor. O middleware recebe todas as mensagens, inclusive descoberta e listagem, então o filtro por ctx.method == "tools/call" é obrigatório para produzir uma linha por ferramenta.

Middleware para gerar events.jsonl

O exemplo abaixo usa MCPServer e a lista oficial server.middleware. Ele mede o relógio monotônico com perf_counter, registra ToolError como resultado com erro e mantém MCPError no canal de protocolo. prompt_sha e tools_sha vêm dos bytes que o próprio harness enviou; não são campos do MCP.

from __future__ import annotations

import asyncio
import json
import os
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from mcp import MCPError
from mcp.server.mcpserver import MCPServer
from mcp.server.mcpserver.exceptions import ToolError

EVENTS = Path(os.getenv("EVENTS_FILE", "events.jsonl"))
WRITE_LOCK = asyncio.Lock()
CATALOG = {"Duna": "Frank Herbert"}


def json_bytes(value: Any) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        separators=(",", ":"),
        default=str,
    ).encode("utf-8")


def field(params: Any, name: str, default: Any) -> Any:
    if isinstance(params, dict):
        return params.get(name, default)
    return getattr(params, name, default)


async def append_event(event: dict[str, Any]) -> None:
    line = json.dumps(event, ensure_ascii=False, separators=(",", ":"))
    async with WRITE_LOCK:
        with EVENTS.open("a", encoding="utf-8") as output:
            output.write(line + "\n")


async def capture_tool_calls(ctx, call_next):
    if ctx.method != "tools/call":
        return await call_next(ctx)

    params = ctx.params  # raw e ainda não validado pelo handler
    tool = field(params, "name", "<unknown>")
    arguments = field(params, "arguments", {})
    started = time.perf_counter()

    base = {
        "schema": "events.v1",
        "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds")
        .replace("+00:00", "Z"),
        "tool": tool,
        "bytes_in": len(json_bytes(arguments)),
        "prompt_sha": os.getenv("PROMPT_SHA"),
        "tools_sha": os.getenv("TOOLS_SHA"),
        "mcp_method": "tools/call",
    }

    try:
        result = await call_next(ctx)
    except MCPError:
        await append_event({
            **base,
            "latency_ms": round((time.perf_counter() - started) * 1000, 3),
            "ok": False,
            "is_error": None,
            "error_channel": "protocol",
            "bytes_out": 0,
        })
        raise
    except Exception:
        await append_event({
            **base,
            "latency_ms": round((time.perf_counter() - started) * 1000, 3),
            "ok": False,
            "is_error": None,
            "error_channel": "crash",
            "bytes_out": 0,
        })
        raise

    is_error = bool(getattr(result, "is_error", False))
    content = getattr(result, "content", [])
    await append_event({
        **base,
        "latency_ms": round((time.perf_counter() - started) * 1000, 3),
        "ok": not is_error,
        "is_error": is_error,
        "error_channel": "tool_result" if is_error else None,
        "bytes_out": len(json_bytes(content)),
    })
    return result


mcp = MCPServer("catalogo", middleware=[capture_tool_calls])


@mcp.tool()
def get_author(title: str) -> str:
    if title not in CATALOG:
        raise ToolError(f"Livro {title!r} não encontrado no catálogo.")
    return CATALOG[title]

O try não é decoração: sem ele, um MCPError sai de call_next antes de a linha ser escrita. Um ToolError, por outro lado, volta como resultado e passa pelo bloco final com is_error=true. Erros de schema podem ser convertidos pelo SDK em resultado corrigível antes de a função da ferramenta rodar.

Esse arquivo direto é suficiente para um processo. Em produção com vários processos ou hosts, envie os eventos para um único coletor ou use um sink que preserve cada registro de forma atômica; o asyncio.Lock protege apenas concorrência dentro do mesmo processo. Observe a saúde do coletor com contador de eventos descartados, tamanho da fila e alerta para ausência de eventos enquanto existem execuções. Log de auditoria que desaparece em silêncio tem uma serenidade admirável, mas pouca utilidade.

bytes_in e bytes_out são o tamanho da serialização feita pelo harness, não campos do protocolo nem uma reprodução garantida do frame no transporte. Se a equipe precisar de bytes exatos no fio, a medição deve ficar na camada de transporte.

Um schema pequeno e versionado

Uma linha mínima fica assim:

{"schema":"events.v1","ts":"2026-09-07T12:00:00.000Z","tool":"Read","latency_ms":18.4,"ok":true,"is_error":false,"bytes_in":128,"bytes_out":2048,"prompt_sha":"...","tools_sha":"...","mcp_method":"tools/call"}

O campo schema permite alterar colunas sem reinterpretar histórico. prompt_sha e tools_sha devem ser hashes dos bytes efetivamente enviados pelo harness, calculados em modo binário; git hash-object identifica um objeto Git e responde a outra pergunta. Para correlação em produção, acrescente run_id e attempt numa próxima versão do schema, em vez de tentar reconstruir uma execução por proximidade de horário.

Também vale decidir o que não gravar. Argumentos e conteúdo podem carregar código, segredos ou dados pessoais; por padrão, registre os tamanhos e identificadores, não o payload bruto. A observação segura é acompanhar taxa de redação, falhas do sink e cardinalidade de tool, mantendo retenção e acesso definidos fora do processo do agente.

Consultando o custo operacional com DuckDB

Aqui, “custo” significa contagem de chamadas, tempo acumulado e bytes calculados pelo harness. MCP não cobra tokens por ferramenta e não retorna custo em dólares por tools/call. Tokens e gasto do provedor continuam em usage.json, sob outro gate.

A documentação de JSON do DuckDB permite ler NDJSON diretamente com read_ndjson:

WITH calls AS (
    SELECT
        CASE
            WHEN tool IN ('Read', 'Grep') THEN 'inspect'
            WHEN tool IN ('Bash', 'Edit') THEN 'mutate'
            ELSE 'other'
        END AS bucket,
        tool,
        latency_ms,
        bytes_in,
        bytes_out,
        ok
    FROM read_ndjson('events.jsonl')
    WHERE schema = 'events.v1'
      AND mcp_method = 'tools/call'
)
SELECT
    bucket,
    tool,
    count(*) AS calls,
    round(100.0 * count(*) / sum(count(*)) OVER (), 1) AS call_share_pct,
    round(sum(latency_ms), 1) AS total_ms,
    round(quantile_cont(latency_ms, 0.95), 1) AS p95_ms,
    sum(bytes_in + bytes_out) AS payload_bytes,
    count(*) FILTER (WHERE NOT ok) AS errors
FROM calls
GROUP BY bucket, tool
ORDER BY calls DESC, tool;

A divisão entre Read/Grep e Bash/Edit é uma taxonomia do harness. Ela mostra se a execução gastou chamadas e espera inspecionando ou alterando o projeto, mas não autoriza converter bytes em tokens. A correção concreta para excesso de leituras repetidas pode ser cachear resultados imutáveis por run_id ou melhorar o packing de contexto; observe depois a participação de inspect, o p95 e a taxa de cache hit, sem registrar o conteúdo lido.

Para Bash ou Edit com erro frequente, não aumente retries no escuro. Restrinja comandos, valide argumentos antes do efeito e acompanhe error_rate por ferramenta e por versão de tools_sha. Assim é possível recuar uma definição de tool problemática sem culpar o modelo por um schema que mudou debaixo dele.

Gate de erro no CI, sem ler tokens

O limite de 20% abaixo é política local de exemplo, não constante do MCP. O job falha apenas sobre eventos de tools/call e não abre usage.json:

- name: Recusar taxa alta de erro nas tools
  shell: bash
  run: |
    python - <<'PY'
    import sys
    import duckdb

    errors, total = duckdb.sql("""
      SELECT count(*) FILTER (WHERE NOT ok), count(*)
      FROM read_ndjson('events.jsonl')
      WHERE schema = 'events.v1'
        AND mcp_method = 'tools/call'
    """).fetchone()

    error_rate = errors / total if total else 0.0
    print(f"tool_error_rate={error_rate:.3f} ({errors}/{total})")
    sys.exit(1 if error_rate > 0.20 else 0)
    PY

O harness precisa publicar events.jsonl antes desse passo; GitHub Actions não enxerga as chamadas por conta própria. Em produção, exporte também total, errors e a ausência inesperada do arquivo como métricas. Um arquivo vazio não deve virar aprovação automática quando a execução afirma ter usado ferramentas: valide o número esperado de runs ou um marcador de conclusão.

Mantenha os gates ortogonais:

Uma execução barata com 40% de ToolError deve falhar no primeiro gate. Uma execução correta e cara pode passar nele e falhar no segundo. Juntar os dois numa média produz um número elegante que não decide nada.

retry_max pertence ao harness

retry_max não existe na especificação nem no SDK. É uma política local aplicada pelo harness quando recebe is_error=true. Erros de execução ficam visíveis ao modelo para que ele corrija argumentos; erros de protocolo ficam com o host e não devem ser entregues como se fossem uma nova tentativa da ferramenta.

Uma fronteira simples pode ser escrita assim:

async def call_with_local_retry(call_tool, request, retry_max: int = 1):
    # retry_max é política local; MCP não possui esse campo.
    for attempt in range(retry_max + 1):
        result = await call_tool(request)
        if not result.is_error:
            return result
        if attempt == retry_max:
            return result
        request = await revise_arguments_from_tool_error(request, result.content)

Use retry automático somente em ferramentas de leitura ou operações comprovadamente idempotentes. Para Bash, Edit e qualquer side effect, exija request_id, valide a intenção e recuse com MCPError antes do efeito quando a política não permitir a operação. Depois de retry_max, grave o último evento e siga para o fluxo de execute-verify-stop, sem uma quarta tentativa “só para confirmar”. O guia de execute-verify-stop detalha essa fronteira de efeitos.

A forma segura de observar retries é incluir run_id, attempt e o resultado de verificação, sem copiar argumentos sensíveis. Acompanhe tentativas por chamada, ferramentas que esgotam o limite e efeitos recusados antes da execução. Se os retries crescem sem reduzir is_error, corrija schema, descrição ou validação da ferramenta; apenas subir retry_max prolonga o diagnóstico.

Armadilhas e verificação

Antes de ligar o gate no repositório inteiro, rode uma tarefa controlada que produza pelo menos um sucesso e um ToolError conhecido. Confira estes pontos em conjunto:

1. chamadas de server/discover e tools/list não aparecem como ferramentas;
2. get_author("Duna") gera ok=true e is_error=false;
3. um título ausente gera resultado com is_error=true, não erro JSON-RPC;
4. parâmetros recusados por política geram error_channel="protocol" e nenhum efeito;
5. a consulta do DuckDB separa inspect de mutate e soma apenas medidas locais;
6. o CI retorna código diferente de zero somente quando errors / total > 0.20;
7. usage.json continua sendo a única entrada do gate de tokens e custo do provedor.

OpenTelemetry pode complementar o arquivo com um span por mensagem, mas o middleware padrão não envia traces a lugar algum sem opentelemetry-sdk e um exporter configurado. Se adotar OTel, alerte sobre falha de exportação e sobre ausência de spans; não use a simples construção de MCPServer como prova de que Jaeger ou outro backend recebeu dados.

A recomendação é firme: adote events.jsonl e o gate de erro quando o harness controla a fronteira de tools/call, consegue correlacionar cada run e preserva os eventos de forma confiável. Se ainda não há run_id, sink serializado ou definição estável de sucesso, comece apenas observando; bloquear PR com telemetria incompleta transforma o guardrail no incidente.

Próximo passo

Implemente primeiro o middleware em modo observação, force uma chamada válida e um ToolError, e rode a consulta do DuckDB. Quando o arquivo estiver completo e correlacionado por execução, habilite o limite no CI. O cost_cap continua onde deve ficar: lendo usage.json, sem atribuir ao MCP uma fatura que ele nunca emitiu.