Servidor MCP com execute-verify-stop e tool gating: o que o modelo não pode decidir sozinho

Um coding agent pode passar pelo gate do repositório, respeitar o teto de custo e ainda chamar uma tool destrutiva no meio da execução. O PR mostra o estado final; não mostra necessariamente cada efeito produzido entre duas chamadas ao modelo.

O resultado que queremos é verificável, não uma promessa no prompt:

• tools/list expõe write_note e read_ledger, mas não rm_rf;
• uma chamada direta a rm_rf recebe erro JSON-RPC antes de tocar o disco;
• write_note grava uma única nota por request_id;
• read_ledger observa o estado depois da mutação;
• um teste mata o servidor antes do commit e repete a chamada sem duplicar o efeito;
• cada decisão entra em um JSONL append-only com uma cadeia SHA-256 editorial.

Isso complementa os gates antes do primeiro PR e o circuit breaker de custo. O ponto novo é a fronteira da chamada de tool: ali o servidor consegue dizer “não” sem negociar com o modelo.

Contexto mínimo: MCP 2026-07-28 e Python SDK 2.1.1

O exemplo usa Python 3.11, funciona a partir do Python 3.10 e fixa mcp==2.1.1, versão publicada em 25 de agosto de 2026. No SDK v2, a classe de alto nível é MCPServer; FastMCP pertence à linha v1. A combinação está documentada nas notas do SDK v2 e na página do pacote 2.1.1.

uv init gated-mcp
cd gated-mcp
uv add "mcp==2.1.1"

No transporte stdio, o host inicia o servidor como subprocesso. stdin e stdout carregam mensagens JSON-RPC, uma por linha, sem Content-Length; logs ficam em stderr. Portanto, nada de print() no servidor. Um print("subiu") parece inocente até virar parte do protocolo. O protocolo, como de costume, não aprecia improviso.

A anatomia atual é curta:

1. o cliente pode consultar server/discover;
2. chama tools/list para obter as tools disponíveis;
3. envia tools/call com nome e argumentos;
4. recebe um resultado ou um erro JSON-RPC.

Na revisão MCP 2026-07-28, não existe handshake de inicialização. Cada request leva em _meta a versão do protocolo e as capacidades do cliente, e todo servidor deve implementar 1. Se escrever JSON à mão, os resultados também precisam de resultType, normalmente "complete".

{"jsonrpc":"2.0","id":"d1","method":"server/discover","params":{"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}}}
{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}}}
{"jsonrpc":"2.0","id":2,"method":"tools/call","params":{"name":"write_note","arguments":{"request_id":"req-0001","text":"gate aplicado"},"_meta":{"io.modelcontextprotocol/protocolVersion":"2026-07-28","io.modelcontextprotocol/clientCapabilities":{}}}}

Ainda existem coding agents presos à revisão 2025-11-25, cujo fluxo começa com initialize. MCPServer 2.1.1 atende as duas eras. Num cliente stdio próprio, a compatibilidade correta é sondar server/discover e cair para initialize somente diante de erro não moderno ou timeout — nunca diante de -32022, que já identifica um servidor moderno com versão incompatível. É fallback, não baseline do tutorial.

A especificação de tools define tools/list, tools/call, schemas e resultados. Ela não entrega arquivo de ACL, cadeia de auditoria, idempotência nem execute-verify-stop. OAuth também é assunto do transporte HTTP; um processo stdio não deve fingir que autenticou o processo pai.

Política versionada antes do side-effect

Crie policy.json:

{
  "version": 1,
  "allow": ["write_note", "read_ledger"],
  "deny": ["rm_rf"],
  "arg_rules": {
    "write_note": {
      "request_id": {"min_len": 8},
      "text": {"max_len": 2000}
    }
  }
}

A política é carregada uma vez, na criação do processo. Uma mudança nela exige restart controlado; se o time precisar de reload, pode acrescentar SIGHUP depois, com teste para troca atômica. Não vale reler o JSON no meio de cada chamada e chamar a corrida resultante de “configuração dinâmica”.

O servidor abaixo faz defesa em profundidade:

• filtra tools/list, para o modelo não planejar rm_rf;
• intercepta tools/call, para uma chamada manual também ser recusada;
• repete o gate no topo de cada handler;
• só registra allow depois do efeito durável.

# server.py
from __future__ import annotations

import fcntl
import hashlib
import json
import logging
import os
import time
from pathlib import Path
from threading import Lock
from typing import Any

from mcp import MCPError
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import INVALID_PARAMS, ToolAnnotations

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

ROOT = Path(os.environ.get("LEDGER_DIR", ".ledger")).resolve()
NOTES = ROOT / "notes"
AUDIT = ROOT / "audit.jsonl"
POLICY_PATH = Path(os.environ.get("POLICY_PATH", "policy.json"))
AUDIT_LOCK = Lock()


def canonical(value: Any) -> bytes:
    return json.dumps(
        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
    ).encode("utf-8")


class Policy:
    def __init__(self, raw: dict[str, Any]):
        if raw.get("version") != 1:
            raise ValueError("unsupported policy version")
        self.allow = set(raw.get("allow", []))
        self.deny = set(raw.get("deny", []))
        self.arg_rules = raw.get("arg_rules", {})

    @classmethod
    def load(cls, path: Path) -> "Policy":
        return cls(json.loads(path.read_text(encoding="utf-8")))

    def permits(self, name: str, arguments: dict[str, Any]) -> bool:
        if name in self.deny or name not in self.allow:
            return False

        for field, rule in self.arg_rules.get(name, {}).items():
            value = arguments.get(field)
            if not isinstance(value, str):
                return False
            if "min_len" in rule and len(value) < rule["min_len"]:
                return False
            if "max_len" in rule and len(value) > rule["max_len"]:
                return False
        return True


def append_audit(
    request_id: str,
    name: str,
    decision: str,
    arguments: dict[str, Any],
    outcome: str,
) -> None:
    ROOT.mkdir(parents=True, exist_ok=True)
    args_digest = hashlib.sha256(canonical(arguments)).hexdigest()

    with AUDIT_LOCK, AUDIT.open("a+", encoding="utf-8") as stream:
        fcntl.flock(stream.fileno(), fcntl.LOCK_EX)
        stream.seek(0)
        lines = [line for line in stream if line.strip()]
        previous = json.loads(lines[-1])["sha256"] if lines else "0" * 64
        record = {
            "ts": time.time_ns(),
            "request_id": request_id,
            "method": "tools/call",
            "name": name,
            "decision": decision,
            "outcome": outcome,
            "args_digest": args_digest,
            "prev_sha256": previous,
        }
        record["sha256"] = hashlib.sha256(
            previous.encode("ascii") + canonical(record)
        ).hexdigest()
        stream.seek(0, os.SEEK_END)
        stream.write(canonical(record).decode("utf-8") + "\n")
        stream.flush()
        os.fsync(stream.fileno())
        fcntl.flock(stream.fileno(), fcntl.LOCK_UN)


class GatedServer(MCPServer):
    def __init__(self) -> None:
        super().__init__("gated-fs")
        self.policy = Policy.load(POLICY_PATH)

    async def list_tools(self):
        tools = await super().list_tools()
        return [tool for tool in tools if tool.name in self.policy.allow]

    def require_allowed(self, name: str, arguments: dict[str, Any]) -> None:
        if self.policy.permits(name, arguments):
            return
        request_id = str(arguments.get("request_id", "missing"))
        append_audit(request_id, name, "deny", arguments, "not_executed")
        raise MCPError(code=INVALID_PARAMS, message=f"tool denied: {name}")

    async def call_tool(self, name, arguments, context=None):
        self.require_allowed(name, arguments)
        return await super().call_tool(name, arguments, context)


mcp = GatedServer()


@mcp.tool(
    annotations=ToolAnnotations(
        idempotent_hint=True,
        destructive_hint=False,
        read_only_hint=False,
    )
)
def write_note(request_id: str, text: str) -> dict[str, str]:
    """Write one note per request_id."""
    arguments = {"request_id": request_id, "text": text}
    mcp.require_allowed("write_note", arguments)
    NOTES.mkdir(parents=True, exist_ok=True)

    key = hashlib.sha256(request_id.encode("utf-8")).hexdigest()
    destination = NOTES / f"{key}.json"
    payload = {"request_id": request_id, "text": text}

    if destination.exists():
        stored = json.loads(destination.read_text(encoding="utf-8"))
        if stored != payload:
            raise ToolError("request_id já usado com outro conteúdo")
        append_audit(request_id, "write_note", "allow", arguments, "replay")
        return {"request_id": request_id, "status": "replay"}

    temporary = NOTES / f".{key}.{os.getpid()}.tmp"
    with temporary.open("xb") as stream:
        stream.write(canonical(payload))
        stream.flush()
        os.fsync(stream.fileno())

    marker = os.environ.get("CRASH_MARKER")
    if marker:
        Path(marker).touch()
        time.sleep(30)

    os.replace(temporary, destination)
    directory_fd = os.open(NOTES, os.O_RDONLY)
    try:
        os.fsync(directory_fd)
    finally:
        os.close(directory_fd)

    append_audit(request_id, "write_note", "allow", arguments, "committed")
    return {"request_id": request_id, "status": "committed"}


@mcp.tool(
    annotations=ToolAnnotations(
        idempotent_hint=True,
        destructive_hint=False,
        read_only_hint=True,
    )
)
def read_ledger() -> dict[str, list[dict[str, str]]]:
    """Observe committed notes."""
    arguments: dict[str, Any] = {}
    mcp.require_allowed("read_ledger", arguments)
    notes = []
    if NOTES.exists():
        for path in sorted(NOTES.glob("*.json")):
            notes.append(json.loads(path.read_text(encoding="utf-8")))
    append_audit("observe", "read_ledger", "allow", arguments, "observed")
    return {"notes": notes}


@mcp.tool()
def rm_rf(path: str) -> str:
    """Tool registrada para provar que list e call têm gates independentes."""
    mcp.require_allowed("rm_rf", {"path": path})
    raise AssertionError("unreachable")


if __name__ == "__main__":
    mcp.run(transport="stdio")

O uso de fcntl e fsync deixa este exemplo direcionado a POSIX, que também é o ambiente do teste com SIGKILL. Em Windows, substitua o lock e a sincronização de diretório por mecanismos equivalentes da plataforma.

Repare na separação de erros. Política negada levanta MCPError(code=INVALID_PARAMS, ...) antes de super().call_tool(), então o host recebe um erro JSON-RPC, não um resultado com isError. Um erro recuperável da execução, como reutilizar o mesmo request_id com outro texto, levanta ToolError de mcp.server.mcpserver.exceptions; o modelo recebe isError: true e pode corrigir os argumentos. Retornar a string "erro" seria sucesso com texto pessimista — um arranjo bastante eficiente para esconder falhas.

As anotações idempotent_hint, destructive_hint e read_only_hint ajudam o host a apresentar as tools, mas são dicas não confiáveis segundo a especificação. A garantia está em request_id, no commit atômico e no gate, não na anotação.

Execute-verify-stop pertence ao harness

Execute-verify-stop é um padrão da aplicação, não uma operação do MCP. A sequência é deliberadamente assimétrica:

1. write_note tenta mudar o estado;
2. read_ledger lê o estado que de fato ficou;
3. o harness compara estado esperado e observado;
4. o harness — não o modelo — escolhe continuar, parar ou iniciar rollback.

Um harness mínimo pode usar o cliente em memória do SDK v2:

# harness.py
import asyncio

from mcp import Client
from server import mcp


async def main() -> None:
    request_id = "req-0001"
    expected = {"request_id": request_id, "text": "gate aplicado"}

    async with Client(mcp) as client:
        mutation = await client.call_tool("write_note", expected)
        if mutation.is_error:
            raise RuntimeError("mutação falhou; stop")

        observation = await client.call_tool("read_ledger", {})
        notes = observation.structured_content["notes"]

        if expected not in notes:
            raise RuntimeError("estado observado divergiu; stop ou rollback")

        # Só daqui em diante o fluxo pode chamar a próxima tool mutante.


asyncio.run(main())

Em produção, a decisão pode ser uma máquina de estados explícita: CONTINUE se a pós-condição fechou, STOP quando o estado é inconclusivo e ROLLBACK apenas quando existe operação compensatória testada. “A tool retornou sucesso” não substitui observação do recurso. Para acompanhar isso sem depender da conversa do agente, emita métricas por decisão e correlacione request_id entre tracing, auditoria e o recurso observado.

Auditoria JSONL com cadeia SHA-256 editorial

A cadeia de hashes do exemplo é uma escolha editorial da aplicação. MCP não padroniza esse formato nem promete evidência antiviolação. Cada linha contém o hash da linha anterior e seu próprio hash:

sha256 = SHA-256(prev_sha256 || canonical_json(registro_sem_sha256))

Isso permite detectar edição, remoção no meio e reordenação ao percorrer o arquivo. Não protege contra um invasor que consiga substituir o arquivo inteiro e o hash-raiz ao mesmo tempo; para esse risco, publique periodicamente o hash final em armazenamento separado e imutável.

O verificador é pequeno:

# verify_chain.py
import hashlib
import json
from pathlib import Path

from server import canonical

previous = "0" * 64
for number, line in enumerate(Path(".ledger/audit.jsonl").read_text().splitlines(), 1):
    record = json.loads(line)
    received = record.pop("sha256")
    assert record["prev_sha256"] == previous, f"elo inválido na linha {number}"
    expected = hashlib.sha256(
        previous.encode("ascii") + canonical(record)
    ).hexdigest()
    assert received == expected, f"hash inválido na linha {number}"
    previous = received

Para provar que a recusa não tocou disco, faça uma observação anterior e outra posterior do recurso-alvo: hash, tamanho, mtime_ns e existência. Depois chame rm_rf e confirme três sinais juntos:

• tools/list não contém rm_rf;
• tools/call falha com -32602 e a auditoria registra decision=deny, outcome=not_executed;
• hash, tamanho, mtime_ns e existência do alvo permanecem iguais.

A remediação para uma tool proibida é o gate anterior ao dispatch; a observação segura é comparar o recurso e a linha de auditoria. Logar apenas a recusa prova que o gate falou “não”, não que outro trecho do processo deixou o arquivo quieto.

Teste dirigido: matar antes do commit e repetir

O transporte stdio considera que requests em voo se perdem quando o processo morre; o cliente pode reiniciar e repetir. Sem idempotência, repetir uma mutação é repetir o efeito.

No servidor, o caminho final deriva do SHA-256 de request_id. A escrita vai para um temporário exclusivo e os.replace() publica o arquivo no mesmo filesystem. Se o processo morrer antes do replace, sobra no máximo um temporário órfão; a próxima execução ainda não vê o destino e conclui o commit. Se morrer depois, o replay lê o destino e não cria uma segunda nota. Para uma API externa, o equivalente seria uma chave de idempotência persistida no mesmo limite transacional do efeito.

O teste abaixo usa o CRASH_MARKER do exemplo para parar exatamente entre fsync do temporário e os.replace:

# crash_test.py
import json
import os
import signal
import subprocess
import sys
import tempfile
import time
from pathlib import Path

META = {
    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
    "io.modelcontextprotocol/clientCapabilities": {},
}


def call(process, request_id):
    message = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "tools/call",
        "params": {
            "name": "write_note",
            "arguments": {"request_id": request_id, "text": "uma vez"},
            "_meta": META,
        },
    }
    process.stdin.write(json.dumps(message, separators=(",", ":")) + "\n")
    process.stdin.flush()


def start(environment):
    return subprocess.Popen(
        [sys.executable, "server.py"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
        env=environment,
    )


with tempfile.TemporaryDirectory() as directory:
    root = Path(directory)
    policy = root / "policy.json"
    policy.write_text(Path("policy.json").read_text(), encoding="utf-8")
    marker = root / "before-commit"

    environment = os.environ | {
        "LEDGER_DIR": str(root / "ledger"),
        "POLICY_PATH": str(policy),
        "CRASH_MARKER": str(marker),
    }

    first = start(environment)
    call(first, "req-crash-0001")
    deadline = time.monotonic() + 5
    while not marker.exists() and time.monotonic() < deadline:
        time.sleep(0.01)
    assert marker.exists(), "a tool não chegou ao ponto de corte"
    os.kill(first.pid, signal.SIGKILL)
    first.wait()

    environment.pop("CRASH_MARKER")
    second = start(environment)
    call(second, "req-crash-0001")
    response = json.loads(second.stdout.readline())
    second.stdin.close()
    second.wait(timeout=5)

    assert response["result"]["resultType"] == "complete"
    assert not response["result"].get("isError", False)

    notes = list((root / "ledger" / "notes").glob("*.json"))
    assert len(notes) == 1
    assert json.loads(notes[0].read_text())["request_id"] == "req-crash-0001"

Rode tudo com:

uv run python harness.py
uv run python verify_chain.py
uv run python crash_test.py

O sinal operacional importante é a razão entre committed, replay, deny e falhas inesperadas, sempre por tool e sem colocar argumentos sensíveis em logs. Um aumento de replay indica retry, crash ou timeout a investigar; um aumento de deny pode indicar prompt ruim, catálogo desatualizado ou tentativa indevida. A correção continua no servidor, e a investigação começa no request_id.

Armadilhas que mudam o resultado

Também vale validar o schema, limitar taxa e sanitizar saídas. O SDK gera e aplica o schema dos handlers decorados, mas predicados de negócio como comprimento, prefixo permitido ou tenant continuam pertencendo à política. Middleware do SDK v2 é provisional; por isso o exemplo mantém o gate no caminho explícito de call_tool e no topo do handler.

Quando adotar e quando recuar

A DevDojo adotaria este desenho quando um servidor stdio expusesse mutações locais, comandos de build/deploy ou acesso a recursos cujo efeito precisa sobreviver a retry e revisão. O mínimo para entrar seria política versionada, recusa antes do dispatch, pós-condição observável, idempotência e auditoria correlacionada.

Recuaria para tools somente de leitura quando não houvesse pós-condição confiável, chave de idempotência ou rollback seguro. Também não colocaria este JSONL local como autoridade de auditoria em múltiplas réplicas: nesse caso, o próximo passo é um ledger transacional central, retenção definida e ancoragem externa do hash final.

Comece com as três verificações mais baratas: confirme que rm_rf não aparece, prove que a chamada direta não altera o alvo e mate write_note antes do commit. Se uma delas falhar, não entregue mais autonomia ao modelo; fortaleça o servidor primeiro.