DEV Community

Emery Chen
Emery Chen

Posted on

Pin the Tool Contract Before You Trust the Agent Loop

An agent that assumes is not actually clever. It is only an untyped RPC client. You should pin every tool before the loop starts. Extra prompt text will not replace a schema.

The model is the last variable

You do not fail because the model looks dumb. You fail because the adapter accepts fiction. The agent invents a field. Your mutate path then executes it.

Money can move. Tickets can close. Files can vanish. Stop asking the model to be careful. Make the tool refuse illegal shapes.

Then give the loop a hard side-effect budget. This is not a taste debate. This is basic production hygiene.

What assuming looks like in logs

You will see the same three lies. They appear before any fancy reasoning trace.

  1. A required key is missing, so the model forges one.
  2. A retry repeats a paid mutation without shame.
  3. A string enum is ignored, so a rogue action lands.

None of those start as prompt problems. They start as contract problems. If you cannot name allowed keys, you cannot debug the call. If you cannot name the budget, you cannot stop the loop.

Read the raw tool JSON first. Do not read the assistant prose first. The prose is theater. The payload is the incident.

A contract is a schema plus brakes

Keep the contract boring on purpose. Boring is easier to test.

  • Schema: names, types, enums, and required fields
  • Idempotency key: one mutation equals one key
  • Side-effect budget: max paid calls per run
  • Timeout: wall clock, never "until it feels done"

You pin these in code. You do not pin them in a system prompt. A prompt is a suggestion. A fence is a gate.

Artifact: a tool fence you can run tonight

The code below is a labeled proposal. Treat it as a local harness. It does not call a vendor until you add transport.

# tool_fence.py
from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Any, Callable, Mapping

ALLOWED_ACTIONS = {"create_ticket", "add_comment", "noop"}

TICKET_SCHEMA = {
    "type": "object",
    "required": ["action", "ticket_id", "idempotency_key"],
    "additionalProperties": False,
    "properties": {
        "action": {"type": "string", "enum": list(ALLOWED_ACTIONS)},
        "ticket_id": {"type": "string", "minLength": 1, "maxLength": 64},
        "body": {"type": "string", "maxLength": 2000},
        "idempotency_key": {"type": "string", "minLength": 8, "maxLength": 128},
    },
}


class ContractError(ValueError):
    pass


def validate_shape(payload: Mapping[str, Any], schema: dict) -> None:
    if not isinstance(payload, dict):
        raise ContractError("payload must be an object")
    for key in schema.get("required", []):
        if key not in payload:
            raise ContractError(f"missing key: {key}")
    props = schema["properties"]
    if schema.get("additionalProperties") is False:
        extra = set(payload) - set(props)
        if extra:
            raise ContractError(f"unknown keys: {sorted(extra)}")
    action = payload.get("action")
    if "action" in payload and action not in props["action"]["enum"]:
        raise ContractError(f"illegal action: {action}")
    ticket_id = str(payload.get("ticket_id", ""))
    if "ticket_id" in payload and not (1 <= len(ticket_id) <= 64):
        raise ContractError("ticket_id length out of range")
    ident = str(payload.get("idempotency_key", ""))
    if "idempotency_key" in payload and not (8 <= len(ident) <= 128):
        raise ContractError("idempotency_key length out of range")


@dataclass
class SideEffectBudget:
    max_mutations: int
    used: int = 0

    def consume(self, action: str) -> None:
        if action == "noop":
            return
        if self.used >= self.max_mutations:
            raise ContractError("side-effect budget exhausted")
        self.used += 1


@dataclass
class ToolFence:
    schema: dict
    budget: SideEffectBudget
    seen_keys: set[str] = field(default_factory=set)
    audit: list[dict] = field(default_factory=list)
    clock: Callable[[], float] = time.time

    def run(self, payload: Mapping[str, Any], mutate: Callable[[dict], dict]) -> dict:
        started = self.clock()
        validate_shape(payload, self.schema)
        action = str(payload["action"])
        ident = str(payload["idempotency_key"])
        if ident in self.seen_keys:
            self.audit.append({"status": "duplicate", "key": ident})
            return {"ok": True, "duplicate": True}
        self.budget.consume(action)
        self.seen_keys.add(ident)
        result = mutate(dict(payload))
        self.audit.append(
            {
                "status": "applied",
                "action": action,
                "key": ident,
                "ms": int((self.clock() - started) * 1000),
            }
        )
        return result
Enter fullscreen mode Exit fullscreen mode

Now add tests that fail on purpose. You want red tests first.

# test_tool_fence.py
from tool_fence import ContractError, SideEffectBudget, TICKET_SCHEMA, ToolFence


def apply(payload):
    return {"ok": True, "action": payload["action"]}


def make_fence(n=1):
    return ToolFence(TICKET_SCHEMA, SideEffectBudget(max_mutations=n))


def test_rejects_forged_action():
    fence = make_fence()
    bad = {
        "action": "delete_everything",
        "ticket_id": "T-1",
        "idempotency_key": "k-12345678",
    }
    try:
        fence.run(bad, apply)
        raise AssertionError("forged action passed")
    except ContractError as exc:
        assert "illegal action" in str(exc)


def test_retry_does_not_double_mutate():
    fence = make_fence(n=2)
    payload = {
        "action": "add_comment",
        "ticket_id": "T-1",
        "body": "ship after contract",
        "idempotency_key": "k-retry-01",
    }
    first = fence.run(payload, apply)
    second = fence.run(payload, apply)
    assert first["ok"] is True
    assert second["duplicate"] is True
    assert fence.budget.used == 1


def test_budget_kills_the_loop():
    fence = make_fence(n=1)
    first = {
        "action": "create_ticket",
        "ticket_id": "T-1",
        "idempotency_key": "k-aaaaaaa1",
    }
    second = {
        "action": "create_ticket",
        "ticket_id": "T-2",
        "idempotency_key": "k-aaaaaaa2",
    }
    fence.run(first, apply)
    try:
        fence.run(second, apply)
        raise AssertionError("budget did not fire")
    except ContractError as exc:
        assert "budget" in str(exc)
Enter fullscreen mode Exit fullscreen mode

Run the file with pytest.

python -m pip install pytest
python -m pytest test_tool_fence.py -q
Enter fullscreen mode Exit fullscreen mode

You now have a kill switch. The agent can still talk. The adapter can still refuse.

Why the tests are the product

A green chat demo proves almost nothing. A red contract test proves the fence works. Keep the illegal action test in CI forever. Keep the duplicate key test in CI forever.

If either test goes missing, the loop is uninsured. Treat a deleted fence test like a deleted auth test.

Map model output before the fence

Many stacks wrap tools as JSON strings. You should parse once. You should reject twice.

import json
from tool_fence import ContractError


def strip_fence(raw: str) -> str:
    text = raw.strip()
    if text.startswith("```

"):
        lines = text.splitlines()
        if lines and lines[-1].strip() == "

```":
            lines = lines[1:-1]
        return "\n".join(lines)
    return text


def parse_tool_json(raw: str) -> dict:
    text = strip_fence(raw)
    if not text:
        raise ContractError("empty tool payload")
    try:
        data = json.loads(text)
    except json.JSONDecodeError as exc:
        raise ContractError(f"invalid json: {exc}") from exc
    if not isinstance(data, dict):
        raise ContractError("tool payload must be an object")
    return data
Enter fullscreen mode Exit fullscreen mode

Never pass a list into mutate. Never pass a leftover markdown fence. Never coerce a near-match enum. Closest verb is how data gets deleted.

Decision table: block, retry, or escalate

Do not improvise inside the hot loop. Pre-decide every branch.

Signal You do You do not
Unknown key Block and log the payload Ask the model to try forever
Missing required field Block with a typed error Fill a default that spends money
Duplicate idempotency key Return the first result Charge or write twice
Budget exhausted Stop the run and page a human Raise max tokens and hope
Transport 5xx Retry with the same key Mint a new key
Enum mismatch Block and treat it as hostile Coerce to the nearest verb

Print this table beside the on-call doc. If a row is missing, you are guessing under pressure.

Soak the fence, not the chatbot

Free tokens tempt you to chat longer. That is the wrong drill for agents. You should spend tokens on illegal payloads.

Build a generator. Feed garbage on purpose.

# fuzz_tools.py
import json
import random
from tool_fence import ContractError, SideEffectBudget, TICKET_SCHEMA, ToolFence

ACTIONS = ["create_ticket", "add_comment", "noop", "drop_table", "", None]
IDS = ["T-1", "", "T" * 80, 12]


def junk():
    return {
        "action": random.choice(ACTIONS),
        "ticket_id": random.choice(IDS),
        "body": "x" * random.choice([0, 10, 5000]),
        "idempotency_key": random.choice(["short", "k-12345678", None]),
        "extra": "nope",
    }


def main(n=200):
    fence = ToolFence(TICKET_SCHEMA, SideEffectBudget(max_mutations=3))
    blocked = 0
    applied = 0
    for _ in range(n):
        payload = {k: v for k, v in junk().items() if v is not None}
        try:
            fence.run(payload, lambda p: {"ok": True})
            applied += 1
        except (ContractError, TypeError):
            blocked += 1
    print(json.dumps({
        "n": n,
        "blocked": blocked,
        "applied": applied,
        "used": fence.budget.used,
    }))


if __name__ == "__main__":
    main()
Enter fullscreen mode Exit fullscreen mode
python fuzz_tools.py
Enter fullscreen mode Exit fullscreen mode

You want blocked high. You want used tiny. If used climbs fast, your fence is theater.

Read the JSON like an incident

Do not admire the dump. Interrogate it. Ask three questions every run.

  1. Which illegal action almost passed?
  2. Did any duplicate key consume budget?
  3. Did wall time explode on noop?

Write the answers in the PR. If you cannot answer, you did not test. A silent fuzz run is just CPU heat.

Count mutations the way you count deploys

You already cap deploys. Cap agent writes the same way.

  • One run, three mutations, then stop
  • One ticket, one create, then comment only
  • One idempotency key per user intent

Put those caps in config, not in chat. A config file can be reviewed. A prompt cannot.

{
  "max_mutations": 3,
  "max_wall_ms": 8000,
  "allowed_actions": ["create_ticket", "add_comment", "noop"]
}
Enter fullscreen mode Exit fullscreen mode

Load that file at process start. Refuse to boot if it is missing. Missing brakes are not a runtime surprise. They are a deploy bug.

Where a scratch model pool actually helps

You still need a model in the loop. The model proposes JSON. The fence judges that JSON.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

MonkeyCode is an open source project. It offers free model access and a free server option. There is also a free 10 million token pool for this kind of drill. Use the pool to generate bad tool calls. Use the free server as a scratch box for the fence.

Do not treat those options as a production SLA. Point the loop at the scratch box. Save every proposed payload. Replay failures through test_tool_fence.py.

A useful command sequence looks like this:

mkdir -p artifacts
python fuzz_tools.py > artifacts/fuzz.json
python -m pytest test_tool_fence.py -q
Enter fullscreen mode Exit fullscreen mode

If the model cannot emit the schema, do not grow the prompt. Shrink the tool instead. A smaller enum is a feature. A larger story is not.

What this opinion rejects

This opinion rejects three common habits.

  1. More system prompt. Careful words do not type-check JSON.
  2. Unlimited retries. Retries without keys duplicate side effects.
  3. Eval as vibes. "It seemed helpful" is not a gate.

You can keep the agent. You cannot keep an open mutate path. An untyped payment RPC would never ship. An untyped agent should not ship either.

Limitations

This fence does not prove the model is smart. It only proves the adapter is strict.

It will not catch:

  • A valid shape aimed at the wrong ticket you own
  • Semantic lies inside a legal body string
  • Authz bugs sitting in the mutate function
  • Prompt injection that still fits the schema
  • Restarts that forget in-memory seen_keys

Store idempotency keys in SQLite or Redis later. Memory dies with the process. Persist the audit log before you add more tools.

Who should not use this

Skip this approach in a few cases.

  • You have no mutate path, only read-only search
  • You cannot name a finite enum of actions
  • You need multi-tenant authz and have none
  • You expect the model to invent new tool names daily

If you cannot list the tools, you are not ready. A loop without a catalog is improvisation. Improvisation is fine in a notebook. It is not fine against live tickets.

Ship the contract first

Put the schema in review. Put the budget in review. Then let the agent speak.

A loop without a fence is a slow script. You would not ship that script for payments. Do not ship it for tickets either.

If you already have a scratch server and a free token pool, run the red tests there before you add another prompt rule.

Top comments (0)