MCP Server with Execute-Verify-Stop and Tool Gating: What the Model Must Not Decide Alone

A coding agent can pass the repository gate, stay under the cost ceiling, and still call a destructive tool halfway through execution. The PR shows the final state; it does not necessarily show every effect produced between two calls to the model.

The result we want is verifiable, not a promise tucked into the prompt:

• tools/list exposes write_note and read_ledger, but not rm_rf;
• a direct call to rm_rf gets a JSON-RPC error before touching the disk;
• write_note writes exactly one note per request_id;
• read_ledger observes the state after the mutation;
• a test kills the server before commit and retries the call without duplicating the effect;
• every decision goes into an append-only JSONL file with an editorial SHA-256 chain.

This complements the gates before the first PR and the cost circuit breaker. The new part is the tool-call boundary: that is where the server can say "no" without negotiating with the model.

Minimum context: MCP 2026-07-28 and Python SDK 2.1.1

The example uses Python 3.11, works on Python 3.10 and later, and pins mcp==2.1.1, released on August 25, 2026. In SDK v2, the high-level class is MCPServer; FastMCP belongs to the v1 line. This combination is documented in the SDK v2 notes and on the 2.1.1 package page.

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

With the stdio transport, the host starts the server as a subprocess. stdin and stdout carry JSON-RPC messages, one per line, without Content-Length; logs go to stderr. So, no print() calls in the server. A harmless-looking print("subiu") eventually becomes part of the protocol. Protocols, as usual, do not care for improvisation.

The current anatomy is short:

1. the client may query server/discover;
2. it calls tools/list to get the available tools;
3. it sends tools/call with a name and arguments;
4. it receives either a result or a JSON-RPC error.

In the MCP 2026-07-28 revision, there is no initialization handshake. Every request carries the protocol version and client capabilities in _meta, and every server must implement 1. If you write JSON by hand, results also need a resultType, usually "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":{}}}}

Some coding agents are still stuck on the 2025-11-25 revision, whose flow begins with initialize. MCPServer 2.1.1 supports both eras. In a custom stdio client, the correct compatibility strategy is to probe server/discover and fall back to initialize only after a non-modern error or timeout, never after -32022, which already identifies a modern server with an incompatible version. That is a fallback, not the tutorial's baseline.

The tools specification defines tools/list, tools/call, schemas, and results. It does not provide an ACL file, audit chain, idempotency, or execute-verify-stop. OAuth is also an HTTP transport concern; a stdio process should not pretend it authenticated its parent process.

Versioned policy before the side effect

Create 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}
    }
  }
}

The policy is loaded once when the process starts. Changing it requires a controlled restart; if the team needs reloads, it can add SIGHUP later, along with a test for atomic replacement. Reading the JSON again in the middle of every call and calling the resulting race condition "dynamic configuration" is not much of a feature.

The server below applies defense in depth:

• it filters tools/list, so the model does not plan to use rm_rf;
• it intercepts tools/call, so a manual call is denied as well;
• it repeats the gate at the top of every handler;
• it records allow only after the effect is durable.

# 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")

Using fcntl and fsync makes this example POSIX-specific, which is also the environment used by the SIGKILL test. On Windows, replace the lock and directory synchronization with equivalent platform mechanisms.

Notice the error separation. A policy denial raises MCPError(code=INVALID_PARAMS, ...) before super().call_tool(), so the host receives a JSON-RPC error rather than a result with isError. A recoverable execution error, such as reusing the same request_id with different text, raises ToolError from mcp.server.mcpserver.exceptions; the model receives isError: true and can correct the arguments. Returning the string "erro" would be a success carrying pessimistic text, a remarkably efficient arrangement for hiding failures.

The idempotent_hint, destructive_hint, and read_only_hint annotations help the host present the tools, but the specification treats them as untrusted hints. The guarantee comes from request_id, the atomic commit, and the gate, not the annotation.

Execute-verify-stop belongs in the harness

Execute-verify-stop is an application pattern, not an MCP operation. The sequence is deliberately asymmetric:

1. write_note tries to change the state;
2. read_ledger reads the state that actually remained;
3. the harness compares the expected and observed states;
4. the harness, not the model, chooses whether to continue, stop, or start a rollback.

A minimal harness can use the SDK v2 in-memory client:

# 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())

In production, the decision can be an explicit state machine: CONTINUE when the postcondition matches, STOP when the state is inconclusive, and ROLLBACK only when a tested compensating operation exists. "The tool returned success" is no substitute for observing the resource. To track this without relying on the agent's conversation, emit metrics for each decision and correlate request_id across tracing, auditing, and the observed resource.

JSONL auditing with an editorial SHA-256 chain

The example's hash chain is an editorial application choice. MCP does not standardize this format or promise tamper-proof evidence. Each line contains the previous line's hash and its own hash:

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

This lets you detect edits, removal from the middle, and reordering while walking the file. It does not protect against an attacker who can replace the entire file and the root hash at the same time; for that risk, periodically publish the final hash to separate, immutable storage.

The verifier is small:

# 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

To prove the denial did not touch the disk, observe the target resource before and after the call: hash, size, mtime_ns, and existence. Then call rm_rf and confirm three signals together:

• tools/list does not contain rm_rf;
• tools/call fails with -32602, and the audit records decision=deny, outcome=not_executed;
• the target's hash, size, mtime_ns, and existence remain unchanged.

The remedy for a forbidden tool is the gate before dispatch; the safe observation is a comparison of the resource and the audit line. Logging only the denial proves that the gate said "no," not that some other part of the process left the file alone.

Directed test: kill before commit and retry

The stdio transport considers in-flight requests lost when the process dies; the client can restart and retry. Without idempotency, retrying a mutation means repeating the effect.

On the server, the final path is derived from the SHA-256 of request_id. The write goes to an exclusive temporary file, and os.replace() publishes the file on the same filesystem. If the process dies before the replace, at most it leaves behind an orphaned temporary file; the next execution still does not see the destination and completes the commit. If it dies afterward, the replay reads the destination and does not create a second note. For an external API, the equivalent would be an idempotency key persisted within the same transactional boundary as the effect.

The test below uses the example's CRASH_MARKER to stop exactly between the temporary file's fsync and 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"

Run everything with:

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

The important operational signal is the ratio among committed, replay, deny, and unexpected failures, always broken down by tool and without putting sensitive arguments in logs. An increase in replay points to a retry, crash, or timeout worth investigating; an increase in deny may point to a bad prompt, a stale catalog, or an improper attempt. The fix stays on the server, and the investigation starts with request_id.

Pitfalls that change the outcome

It is also worth validating the schema, rate limiting, and sanitizing outputs. The SDK generates and applies the schema for decorated handlers, but business predicates such as length, allowed prefix, or tenant still belong in the policy. SDK v2 middleware is provisional, so the example keeps the gate in the explicit call_tool path and at the top of the handler.

When to adopt it and when to step back

DevDojo would adopt this design when a stdio server exposed local mutations, build or deploy commands, or access to resources whose effects must survive retries and review. The minimum bar would be a versioned policy, denial before dispatch, an observable postcondition, idempotency, and correlated auditing.

We would step back to read-only tools when there is no reliable postcondition, idempotency key, or safe rollback. We also would not make this local JSONL file the audit authority across multiple replicas: in that case, the next step is a central transactional ledger, defined retention, and external anchoring of the final hash.

Start with the three cheapest checks: confirm that rm_rf does not appear, prove that the direct call does not change the target, and kill write_note before commit. If any of them fail, do not give the model more autonomy; strengthen the server first.