Tool Authorization and Argument Validation

Consider a hypothetical order-support assistant for an online store. A customer chats with it, and the language model behind it can request four operations: read an order, refund part of an order, change the shipping address of an order that has not shipped, and email the customer an order status. The model executes none of them. It emits a tool call, a tool name plus arguments, and the application decides what happens next.

Suppose the model proposes issue_refund with order ord_1002 and an amount of 8,000 cents. Four separate questions follow. Is the call well-formed? May this customer touch that order? Does the order’s current state allow a refund of that size? And is this an action the system should perform with nobody looking? A JSON schema answers only the first. The rest need checks that know who is asking, what the data looks like now, and what the action costs if it is wrong.

OWASP’s LLM06:2025 entry, “Excessive Agency,” names three root causes for damage done through model-driven tools: excessive functionality, excessive permissions, and excessive autonomy. Each cause has a matching mechanism. A typed tool contract and a narrow tool set limit functionality. Least privilege and authorization in the context of the end user limit permissions. Argument validation keeps malformed or hostile values away from the backend. A consequence policy decides how much autonomy each action gets. Integration tests keep all of it true as the code changes.

The article on model API integration supplies the starting point: a reply is validated in layers before it is used, a failed validation gets a bounded repair or a fallback, and a repeated call needs a key so that repetition does not repeat the effect. Here the thing being validated is a tool call, and the thing being protected is a backend that changes state. The article on reproducible evaluation supplies the method for measuring whether the model proposes the right calls. All programs below are Python 3.12 using only the standard library. The backend is an in-memory fake, the “model” is a list of scripted calls, and the data is synthetic.

One call, nine checks

Every tool call passes through the same sequence before anything changes. The order matters, so it is worth stating once as a whole.

model output: tool name + arguments            (untrusted proposal)
  1. tool exists?                       no  -> unknown_tool
  2. tool exposed in this session?      no  -> not_available
  3. session holds the required scope?  no  -> forbidden
  4. normalize strings, check schema    bad -> invalid_arguments (errors go back to the model)
  5. object exists and belongs to the caller?
                                        no  -> not_found (same answer for both causes)
  6. a completed call with the same key?
                                        yes -> return the stored result
  7. does the object's current state allow it?
                                        no  -> invalid_state
  8. consequence policy                 -> auto, or needs_approval
  9. execute, record the effect, update session budgets

The sequence runs from cheap and general to specific and expensive, and no stage reveals more than the caller is entitled to know. Checks 1 to 3 need only the session and the tool name. Check 4 needs the arguments but no data. Check 5 touches the data, and it comes before the state check so that a caller can never learn the state of an order that is not theirs. Check 6 sits before check 7 because a completed refund has changed the order’s state, and a retry of that refund would otherwise fail the state check with a misleading error instead of returning the original result.

Typed tool contracts

A tool contract is the agreement between three parties: the model that proposes calls, the runtime that executes them, and the people who must be able to reason about what a tool can do. It has two halves, and they play different roles.

Half	Fields	Who reads it	What it is for
Model-facing	name, description, input schema	The model, as part of its input	Persuades the model to choose the tool and fill the arguments correctly
Runtime-only	effect class, required scope, state rules, impact, execution	The dispatcher	Enforces what the tool may do, whatever the model proposes

The split matters because the model-facing half is text that influences a statistical system, while the runtime-only half is code that cannot be talked out of anything. A permission written into a description (“only use this for the customer’s own orders”) is advice. The same rule in the dispatcher is a check.

What the providers standardize

Provider APIs converge on the same three model-facing fields. Anthropic’s documentation for defining tools lists a name, a description, and an input_schema that is a JSON Schema object, and recommends long, specific descriptions covering what the tool does, when to use it, and what each parameter means. OpenAI’s function calling guide describes a function by a name, a description, and a parameters JSON Schema, with a strict flag. With strict mode enabled, every object in the schema must set additionalProperties to false and every property must be listed as required, with optional parameters expressed as a null-able type. The Model Context Protocol (MCP) tool definition has a name, description, and inputSchema, plus an optional outputSchema and optional annotations that describe tool behavior. The MCP specification adds two rules that bear on this article: servers must validate all tool inputs and implement access controls, and clients must treat tool annotations as untrusted unless they come from a trusted server.

Strict or constrained modes reduce malformed arguments, as the article on decoding and structured output explains for generation in general, but they enforce a subset of JSON Schema and only structure. Whether the order belongs to the caller, whether the refund fits the order, and whether the session has already refunded too much are application facts that no schema can express. The dispatcher therefore validates again, whatever the provider promises.

Design rules for the model-facing half

Five rules recur in practice.

• **Make tools task-shaped and narrow, and merge only along permission boundaries.** Anthropic’s guidance recommends consolidating related operations into fewer tools with an action parameter to reduce selection ambiguity, and OpenAI’s guide suggests keeping the number of available functions small. Both address how reliably the model chooses. They do not address authorization. A tool with action: "read" | "refund" makes the privilege level an argument that the model, and any text that influences it, controls. Merging two read variants into one tool is harmless. Merging a read and a write behind one enum means the scope check has to inspect the arguments to know which scope applies.
• **Keep identity and destination out of the arguments.** The customer’s id comes from the authenticated session, not from a customer_id parameter, because a parameter can be filled with any string the model is persuaded to produce. The same rule applies to destinations: send_status_email in the code below takes only an order id, and the recipient is read from the order record. A tool with a free to field turns every prompt injection into a data exfiltration channel.
• **Use closed vocabularies.** Enums for reasons, patterns for identifiers, integers in minor currency units for money, and additionalProperties: false everywhere. Anything the model may choose freely is something the validator must later judge on its merits.
• **Return the minimum.** A tool result goes back into the model’s context, so it is content that the model reads and that other parties may have written. get_order returns status and totals, not the email address or street address it does not need. Results from tools that fetch third-party text are an injection channel and belong to the boundary discussed in Prompt Injection and Untrusted Content Boundaries.
• **Define the error vocabulary.** The runtime returns a small fixed set of codes (invalid_arguments, not_found, invalid_state, needs_approval, and so on) with optional machine-readable detail. A model can act on a code. It cannot act on a stack trace, and a stack trace can leak internals. MCP separates protocol errors, such as an unknown tool, from tool execution errors reported in the result with isError set; the same distinction appears in the status codes below.

A contract in code

The first module defines the effect classes, a small schema checker, and the ToolSpec that joins both halves. The checker implements only the JSON Schema keywords the tools use: object, string, integer, enum, minimum and maximum, length limits, and pattern. A real system would use a maintained JSON Schema validator; the point of a hand-written one is that each rule is visible.

"""Tool contracts for a hypothetical order-support assistant. Python 3.12, standard library only."""
from __future__ import annotations
import re
from dataclasses import dataclass
from enum import Enum
from typing import Callable

class Effect(Enum):
    READ = "read"
    WRITE_REVERSIBLE = "write_reversible"
    WRITE_IRREVERSIBLE = "write_irreversible"

def check_schema(value, schema: dict, path: str = "$") -> list[str]:
    """A small JSON Schema subset: object, string, integer, enum, minimum, maximum, pattern, lengths."""
    t = schema.get("type")
    if t == "object":
        if not isinstance(value, dict):
            return [f"{path}: expected object"]
        props = schema.get("properties", {})
        errs = [f"{path}.{k}: required" for k in schema.get("required", []) if k not in value]
        if schema.get("additionalProperties") is False:
            errs += [f"{path}.{k}: unexpected property" for k in value if k not in props]
        for k, sub in props.items():
            if k in value:
                errs += check_schema(value[k], sub, f"{path}.{k}")
        return errs
    if t == "integer":
        if not isinstance(value, int) or isinstance(value, bool):   # bool is a subclass of int
            return [f"{path}: expected integer"]
        errs = []
        if "minimum" in schema and value < schema["minimum"]: errs.append(f"{path}: below minimum {schema['minimum']}")
        if "maximum" in schema and value > schema["maximum"]: errs.append(f"{path}: above maximum {schema['maximum']}")
        return errs
    if t == "string":
        if not isinstance(value, str):
            return [f"{path}: expected string"]
        errs = []
        if "enum" in schema and value not in schema["enum"]: errs.append(f"{path}: not one of {schema['enum']}")
        if "minLength" in schema and len(value) < schema["minLength"]: errs.append(f"{path}: too short")
        if "maxLength" in schema and len(value) > schema["maxLength"]: errs.append(f"{path}: too long")
        if "pattern" in schema and not re.search(schema["pattern"], value): errs.append(f"{path}: does not match {schema['pattern']}")
        return errs
    raise ValueError(f"unsupported schema type {t!r}")

@dataclass(frozen=True)
class ToolSpec:
    # Model-facing half: these three fields are what the model is shown.
    name: str
    description: str
    input_schema: dict
    # Runtime-only half: the model never sees these and cannot argue with them.
    effect: Effect
    permission: str                                   # scope the session must hold
    order_id: Callable                                # (args) -> id of the one order the call touches
    check_state: Callable                             # (record, args) -> list of error strings
    impact_cents: Callable                            # (record, args) -> money at stake
    execute: Callable                                 # (store, ctx, record, args) -> result dict

    def model_view(self) -> dict:
        return {"name": self.name, "description": self.description, "input_schema": self.input_schema}

Two details in the checker address known failure modes. Python’s bool is a subclass of int, so a plain isinstance(value, int) would accept true as an amount; the checker excludes it explicitly, as the validator in the API integration article does. And the pattern keyword in JSON Schema is not implicitly anchored: it matches if the pattern is found anywhere in the string. The patterns below therefore carry explicit ^ and $ anchors, and the checker uses re.search to follow that semantics.

The second module holds a fake backend, the session object, and four tool specifications. Money is an integer number of cents. The backend keeps an effect ledger of every state change, which the tests later read.

"""A dispatcher that stands between model-proposed tool calls and a fake backend (Python 3.12, stdlib)."""
from __future__ import annotations
import hashlib, hmac, json, secrets, unicodedata
from dataclasses import dataclass
from contracts import Effect, ToolSpec, check_schema

class Store:
    """Fake backend. `effects` is a ledger of every state change, which tests can inspect."""
    def __init__(self):
        self.orders = {
            "ord_1001": {"customer": "cu_ana", "total": 4200,  "refunded": 0, "status": "delivered",
                         "email": "ana@example.test", "address": "1 Elm St, 94107"},
            "ord_1002": {"customer": "cu_ben", "total": 15900, "refunded": 0, "status": "delivered",
                         "email": "ben@example.test", "address": "9 Oak Ave, 10001"},
            "ord_1003": {"customer": "cu_ana", "total": 12000, "refunded": 0, "status": "processing",
                         "email": "ana@example.test", "address": "1 Elm St, 94107"},
            "ord_1004": {"customer": "cu_ana", "total": 15000, "refunded": 0, "status": "delivered",
                         "email": "ana@example.test", "address": "1 Elm St, 94107"},
        }
        self.effects: list[tuple] = []

@dataclass
class Session:
    user_id: str                    # from the authenticated session, never from model output
    scopes: frozenset
    tools: frozenset                # the tools exposed to the model in this session
    writes: int = 0
    irreversible_total: int = 0

ORDER_ID = {"type": "string", "pattern": r"^ord_[0-9]{4,8}$"}

The tool definitions follow. Each one fills both halves of the contract. After the effect class and scope, the last four fields of each spec are the order id the call touches, the state rules, the impact in cents, and the execution function.

def _refund(store, ctx, rec, a):
    rec["refunded"] += a["amount_cents"]
    store.effects.append(("refund", a["order_id"], a["amount_cents"]))
    return {"refunded_cents": rec["refunded"]}

def _address(store, ctx, rec, a):
    rec["address"] = f'{a["line1"]}, {a["postal_code"]}'
    store.effects.append(("address", a["order_id"], rec["address"]))
    return {"address": rec["address"]}

def _email(store, ctx, rec, a):
    store.effects.append(("email", a["order_id"], rec["email"]))   # recipient comes from the record
    return {"sent": True}

TOOLS = {t.name: t for t in [
    ToolSpec("get_order", "Return status and payment totals for one order of the current customer.",
             {"type": "object", "properties": {"order_id": ORDER_ID}, "required": ["order_id"],
              "additionalProperties": False},
             Effect.READ, "orders:read", lambda a: a["order_id"],
             lambda rec, a: [], lambda rec, a: 0,
             lambda st, c, rec, a: {"order_status": rec["status"], "total_cents": rec["total"],
                                    "refunded_cents": rec["refunded"]}),
    ToolSpec("issue_refund", "Refund part or all of a delivered order to the original payment method.",
             {"type": "object", "properties": {
                 "order_id": ORDER_ID,
                 "amount_cents": {"type": "integer", "minimum": 1, "maximum": 100000},
                 "reason": {"type": "string", "enum": ["damaged", "not_as_described", "late", "other"]}},
              "required": ["order_id", "amount_cents", "reason"], "additionalProperties": False},
             Effect.WRITE_IRREVERSIBLE, "refunds:write", lambda a: a["order_id"],
             lambda rec, a: ([] if rec["status"] == "delivered" else ["order is not delivered"])
                          + ([] if a["amount_cents"] <= rec["total"] - rec["refunded"]
                             else [f'amount exceeds refundable {rec["total"] - rec["refunded"]}']),
             lambda rec, a: a["amount_cents"], _refund),
    ToolSpec("update_shipping_address", "Change the delivery address of an order that has not shipped.",
             {"type": "object", "properties": {
                 "order_id": ORDER_ID,
                 "line1": {"type": "string", "pattern": r"^[^\x00-\x1f\x7f]{1,80}$"},
                 "postal_code": {"type": "string", "pattern": r"^[0-9]{5}$"}},
              "required": ["order_id", "line1", "postal_code"], "additionalProperties": False},
             Effect.WRITE_REVERSIBLE, "orders:write", lambda a: a["order_id"],
             lambda rec, a: [] if rec["status"] == "processing" else ["order has already shipped"],
             lambda rec, a: 0, _address),
    ToolSpec("send_status_email", "Email the order status to the customer's address on file.",
             {"type": "object", "properties": {"order_id": ORDER_ID}, "required": ["order_id"],
              "additionalProperties": False},
             Effect.WRITE_IRREVERSIBLE, "email:send", lambda a: a["order_id"],
             lambda rec, a: [], lambda rec, a: 0, _email),
]}

The model sees only model_view(). For the refund tool that is the following object, printed from the code above.

{
  "name": "issue_refund",
  "description": "Refund part or all of a delivered order to the original payment method.",
  "input_schema": {
    "type": "object",
    "properties": {
      "order_id": {
        "type": "string",
        "pattern": "^ord_[0-9]{4,8}$"
      },
      "amount_cents": {
        "type": "integer",
        "minimum": 1,
        "maximum": 100000
      },
      "reason": {
        "type": "string",
        "enum": [
          "damaged",
          "not_as_described",
          "late",
          "other"
        ]
      }
    },
    "required": [
      "order_id",
      "amount_cents",
      "reason"
    ],
    "additionalProperties": false
  }
}

Everything else in ToolSpec, namely the effect class, the scope, the state rules, and the impact function, stays on the runtime side. A manipulated model can produce any string in the schema’s space, and the contract’s job is to make sure that no point in that space does more than the session is allowed to do.

Least privilege and authorization

Who is calling

A tool call involves three actors, and only two of them have an identity. The customer is authenticated by the application. The application’s service account holds credentials for the backend. The model has neither: it is a text generator whose output depends on the system prompt, the conversation, retrieved passages, and tool results, any of which may carry text written by someone other than the user. Treating the model as a principal that is “trusted to decide” what to do moves the trust boundary into the one component whose behavior cannot be specified exactly.

This setup invites a classic mistake. The runtime holds a service credential that can read and refund any order. A model-proposed call arrives with an order id, and the runtime satisfies it with the service credential. The runtime has become what Norm Hardy called a confused deputy: a program holding authority from two sources, its own and the one its caller supplies, that cannot tell which one a request should be judged under. In Hardy’s account, a compiler that was allowed to write to a protected system file overwrote a billing file when a user named that file as the destination for debugging output. The same structure appears when the “user” is a model and the “file name” is an order id. The remedy in this setting is the one OWASP lists as user context authorization: execute actions on behalf of a user in the context of that specific user, with the minimum privilege necessary. In the code, the session carries the authenticated user_id, and every object lookup is judged against it.

Three levels of check

Authorization decisions for a tool call fall into three levels, and they answer different questions. The authorization glossary entry covers the general concept; the levels below are how it applies to tools.

Level	Question	In the code	Typical failure when missing
Function	May this session use this tool at all?	Tool exposed in the session, scope held	A read-only session issues refunds
Object	May this session touch this particular record?	Order must belong to ctx.user_id	One customer reads or changes another’s order
Condition	Is this action allowed now, with these values?	State rules, amount limits, session budgets	Refund exceeds what was paid; address changed after shipment

The object level has its own entry in the OWASP API Security Top 10: Broken Object Level Authorization, the first item of the 2023 list, where attackers manipulate an identifier in a request to reach another user’s object. OWASP’s prevention advice is to check authorization in every function that uses a client-supplied identifier to access a record, and to write tests that fail the build when it is missing. A model-driven tool is a client-supplied identifier in the most literal sense, since the identifier is whatever string the model emitted. The condition level is the one most often forgotten, because it looks like business logic. It is authorization too: the question “may this user refund 8,000 cents of this order today” has a different answer from “may this user refund anything.”

Least privilege on four axes

Least privilege means each identity holds only the permissions its task needs. For tool-using systems that principle applies along four axes, summarized in the least privilege glossary entry for the general case. Saltzer and Schroeder list least privilege together with fail-safe defaults and complete mediation among their design principles for protection mechanisms, and all three show up in the dispatcher.

• **Which tools are exposed.** The session’s tool list is built per task. A status-lookup session needs get_order only. Fewer tools also mean fewer tokens in the prompt and fewer wrong choices.
• **Which scopes the session holds.** The dispatcher requires refunds:write for the refund tool. Scopes that come from roles are the role-based side of access control; ownership of an order is a relationship attribute. Real systems combine both.
• **Which objects.** Ownership, tenant, region, or whatever the data model uses to partition records.
• **Which downstream credentials.** The handlers call real services. A read tool should hold a credential that cannot write, a refund handler a credential that can refund but not delete, and where the downstream system supports delegated or user-scoped tokens, those are better than a shared service account. Tools that run code or touch files belong inside a sandbox, which limits what the tool can reach and does not decide what each caller may ask of it.

Exposing fewer tools to the model is a convenience, not a control. The model can name a tool that was never offered, either by hallucination or because injected text told it to, and the dispatcher must refuse such a call itself. This is the complete mediation principle: every access is checked, whatever earlier layers did. Likewise the defaults fail closed. An unknown tool, a missing scope, or an unrecognized stage outcome is a refusal and never an implicit allow.

One more detail concerns what a refusal reveals. If the dispatcher answered “forbidden” for another customer’s order and “not found” for a nonexistent one, a caller could enumerate which order ids exist. The code returns the same not_found for both.

The dispatcher

The remaining part of the module is the dispatcher. It implements the nine-stage sequence above in one function. Stages 4, 7, and 8 belong to the sections on validation and consequences; they are in the listing so that the sequence can be read in one piece.

AUTO_LIMIT, SESSION_CAP, MAX_WRITES = 5_000, 10_000, 3

def canonical(args: dict) -> dict:
    """Normalize every string once, before any check: Unicode NFKC and trimmed ends."""
    return {k: unicodedata.normalize("NFKC", v).strip() if isinstance(v, str) else v for k, v in args.items()}

def args_hash(args: dict) -> str:
    return hashlib.sha256(json.dumps(args, sort_keys=True, separators=(",", ":")).encode()).hexdigest()[:16]

class Runtime:
    def __init__(self, store: Store, enforce_owner: bool = True):
        self.store, self.enforce_owner = store, enforce_owner
        self._key = secrets.token_bytes(16)          # approval key, generated per run
        self._done: dict = {}                        # (user, tool, call_key) -> (args_hash, result)

    def issue_approval(self, user_id: str, tool: str, args: dict) -> str:
        """Stands in for the human-facing approval service. The token binds user, tool and exact arguments."""
        msg = json.dumps([user_id, tool, canonical(args)], sort_keys=True).encode()
        return hmac.new(self._key, msg, hashlib.sha256).hexdigest()[:16]

    def decide(self, spec: ToolSpec, impact: int, ctx: Session) -> str:
        if spec.effect is Effect.READ:
            return "auto"
        if spec.effect is Effect.WRITE_REVERSIBLE:
            return "auto" if ctx.writes < MAX_WRITES else "approve"
        return "auto" if impact <= AUTO_LIMIT and ctx.irreversible_total + impact <= SESSION_CAP else "approve"

    def call(self, ctx: Session, name: str, raw_args, call_key: str | None = None, approval: str | None = None) -> dict:
        err = lambda code, **kw: {"status": "error", "code": code, **kw}
        spec = TOOLS.get(name)
        if spec is None:                                   return err("unknown_tool")
        if name not in ctx.tools:                          return err("not_available")
        if spec.permission not in ctx.scopes:              return err("forbidden")
        if not isinstance(raw_args, dict):                 return err("invalid_arguments", detail=["$: expected object"])
        args = canonical(raw_args)
        errors = check_schema(args, spec.input_schema)
        if errors:                                         return err("invalid_arguments", detail=errors)
        rec = self.store.orders.get(spec.order_id(args))
        if rec is None or (self.enforce_owner and rec["customer"] != ctx.user_id):
            return err("not_found")                        # same answer for "missing" and "not yours"
        if spec.effect is not Effect.READ:
            if not call_key:                               return err("missing_call_key")
            seen = self._done.get((ctx.user_id, name, call_key))
            if seen:
                if seen[0] != args_hash(args):             return err("key_reused")
                return {**seen[1], "replayed": True}
        problems = spec.check_state(rec, args)
        if problems:                                       return err("invalid_state", detail=problems)
        impact = spec.impact_cents(rec, args)
        if self.decide(spec, impact, ctx) == "approve":
            if not (approval and hmac.compare_digest(approval, self.issue_approval(ctx.user_id, name, args))):
                return {"status": "needs_approval", "preview": {"tool": name, "args": args, "impact_cents": impact}}
        result = {"status": "ok", **spec.execute(self.store, ctx, rec, args)}
        if spec.effect is not Effect.READ:
            ctx.writes += 1
            if spec.effect is Effect.WRITE_IRREVERSIBLE:
                ctx.irreversible_total += impact
            self._done[(ctx.user_id, name, call_key)] = (args_hash(args), result)
        return result

The identity never enters through the arguments. ctx.user_id is set by the application from the authenticated session, and the order lookup in stage 5 compares it with the record’s owner. The following program exercises stages 1 to 5 for one customer.

from harness import fresh, R
from runtime import Session

st, rt, ana = fresh()
for label, tool, args in [("own order",            "get_order", {"order_id": "ord_1001"}),
                          ("another customer's",   "get_order", {"order_id": "ord_1002"}),
                          ("nonexistent",          "get_order", {"order_id": "ord_9999"})]:
    print(f"{label:20}", rt.call(ana, tool, args))

print()
no_write = Session("cu_ana", frozenset({"orders:read"}), ana.tools)          # same user, read-only scope
print("refund without scope  ", rt.call(no_write, "issue_refund", R(), "k1"))
catalog_only = Session("cu_ana", ana.scopes, frozenset({"get_order"}))         # refund tool not exposed
print("refund tool not shown ", rt.call(catalog_only, "issue_refund", R(), "k2"))
print("ledger:", st.effects)

own order            {'status': 'ok', 'order_status': 'delivered', 'total_cents': 4200, 'refunded_cents': 0}
another customer's   {'status': 'error', 'code': 'not_found'}
nonexistent          {'status': 'error', 'code': 'not_found'}

refund without scope   {'status': 'error', 'code': 'forbidden'}
refund tool not shown  {'status': 'error', 'code': 'not_available'}
ledger: []

Reading the output: the customer’s own order is returned with only the fields the tool exposes. Another customer’s order and a nonexistent one produce byte-identical responses. A session without the refund scope is refused with forbidden, and a session that was never shown the refund tool is refused with not_available even though the call names a tool that exists. The ledger is empty, so none of the refused calls changed anything.

Argument validation

Arguments are untrusted input

Model-proposed arguments deserve the same suspicion as the fields of an HTTP request from a stranger. The difference is in the failure distribution. Most bad arguments from a well-behaved model look like honest mistakes: a number rendered as a string, an enum value that is a near miss, an identifier that was never in the conversation. Under prompt injection, or when a user steers the model, they look like attacks. One validator has to handle both, and it should not depend on telling them apart.

The OWASP Input Validation Cheat Sheet states the stance: define what the application accepts and reject everything else, rather than trying to recognize every malicious string, and validate both syntax and meaning. The layers below add one stage that the cheat sheet also names, canonicalization before validation.

Five layers

1. **Shape and type.** Is the argument an object with exactly the expected properties, of the expected types? A string where an integer is required, a boolean where a number is required, and an extra property are all rejections. Unknown properties are rejected, not ignored: an extra customer_id is the shape of an attempted identity override, and silently dropping it hides the attempt.
2. **Canonical form.** Normalize before checking, once, and use the normalized value from then on. The dispatcher applies Unicode NFKC normalization and trims string ends. Without it, full-width digits such as “９４１０７” would be rejected by a [0-9] pattern, or accepted by a \d pattern that matches any Unicode digit and stored in a form downstream code does not expect. Validating one representation and storing another is the inconsistency the OWASP cheat sheet warns about.
3. **Value rules.** Ranges, lengths, enums, and patterns written as allow-lists. The refund amount must be an integer from 1 to 100,000 cents; the reason is one of four words; the postal code is five ASCII digits.
4. **Reference.** Does the identifier name something that exists, and may this caller use it? Stage 5 of the dispatcher is this layer, merged with object-level authorization because the two are one lookup.
5. **State.** Does the current state of the object allow the action with these values? A refund must not exceed what is still refundable, and an address cannot change after shipment. These rules read live data, so they run after the object is loaded.

State checks have a race. The state read at check time may change before the write lands, for example when two sessions refund the same order at once. In a real backend the check and the write belong in one transaction or a conditional update (for example, an update that adds to the refunded amount only if the sum stays within the total). The fake backend is single-threaded and does not model this; the point is that the check in the dispatcher is an early, friendly rejection, and the backend must still enforce the invariant at the point of the write.

Money is an integer in minor units for a reason beyond type safety. A float such as 29.99 has no exact binary representation, and a model that writes “29.99 dollars”, “29.99”, or “2999” for the same amount makes the argument ambiguous. An integer count of cents with an explicit unit in the parameter name (amount_cents) leaves one reading.

Validation is not injection defense

Validation reduces what can reach a sink; it does not make a sink safe. The OWASP cheat sheet makes this point explicitly: parameterized queries, not input filtering, are the defense against SQL injection. In this article’s tools the order id is constrained by a strict pattern, but free text such as an address line could contain anything, and a tool backed by SQL, a shell, a file path, or an outbound URL has a sink for which validation is only the first of two controls. The next program shows the sink problem with an in-memory SQLite database. The string stands for an argument a manipulated model might pass along; the caller is Ana, and the target row belongs to Ben.

import sqlite3
db = sqlite3.connect(":memory:")
db.execute("CREATE TABLE orders (id TEXT, customer TEXT)")
db.executemany("INSERT INTO orders VALUES (?, ?)", [("ord_1001", "cu_ana"), ("ord_1002", "cu_ben")])
caller = "cu_ana"
arg = "x' OR customer = 'cu_ben' --"                   # a string a manipulated model could pass along
naive = f"SELECT id, customer FROM orders WHERE id = '{arg}' AND customer = '{caller}'"
print("concatenated :", db.execute(naive).fetchall())
print("parameterized:", db.execute("SELECT id, customer FROM orders WHERE id = ? AND customer = ?", (arg, caller)).fetchall())

concatenated : [('ord_1002', 'cu_ben')]
parameterized: []

The concatenated query turns the argument into SQL: the injected condition and comment marker make it return Ben’s row even though the query still names Ana as the caller. The parameterized query treats the whole string as a value, matches nothing, and returns an empty list. The same pattern applies to other sinks: join file paths against a fixed base directory and verify the resolved path stays inside it, resolve URLs against an allow-list of hosts before fetching, and pass command arguments as lists, not strings.

Cases and results

The table below sends nineteen calls through a fresh backend. Each row changes one thing in a valid call; the last column counts entries added to the ledger. The harness and the cases are short enough to read completely.

from runtime import Store, Runtime, Session, TOOLS

SCOPES = frozenset({"orders:read", "refunds:write", "orders:write", "email:send"})

def fresh(user="cu_ana"):
    """A new backend, a runtime over it, and a session for one customer with every tool exposed."""
    st = Store()
    return st, Runtime(st), Session(user, SCOPES, frozenset(TOOLS))

def R(**kw):
    """Refund arguments for a delivered order, with overrides."""
    return {"order_id": "ord_1001", "amount_cents": 3000, "reason": "damaged", **kw}

from harness import fresh, R

cases = [
    ("valid refund",              "issue_refund", R()),
    ("amount as string",          "issue_refund", R(amount_cents="3000")),
    ("amount as float",           "issue_refund", R(amount_cents=29.99)),
    ("amount as bool",            "issue_refund", R(amount_cents=True)),
    ("negative amount",           "issue_refund", R(amount_cents=-500)),
    ("amount above refundable",   "issue_refund", R(amount_cents=4300)),
    ("reason outside enum",       "issue_refund", R(reason="customer is angry")),
    ("extra property",            "issue_refund", R(customer_id="cu_ben")),
    ("missing property",          "issue_refund", {"order_id": "ord_1001", "amount_cents": 100}),
    ("another customer's order",  "issue_refund", R(order_id="ord_1002")),
    ("order does not exist",      "issue_refund", R(order_id="ord_9999")),
    ("injection-shaped id",       "get_order",    {"order_id": "ord_1001' OR '1'='1"}),
    ("order not delivered",       "issue_refund", R(order_id="ord_1003")),
    ("unknown tool",              "delete_order", {"order_id": "ord_1001"}),
    ("postal code, full-width",   "update_shipping_address", {"order_id": "ord_1003", "line1": "5 Pine Rd", "postal_code": "９４１０７"}),
    ("postal code, trailing \\n","update_shipping_address", {"order_id": "ord_1003", "line1": "5 Pine Rd", "postal_code": "94107\n"}),
    ("postal code, embedded \\n","update_shipping_address", {"order_id": "ord_1003", "line1": "5 Pine Rd", "postal_code": "94107\n99999"}),
    ("address after shipment",    "update_shipping_address", {"order_id": "ord_1001", "line1": "5 Pine Rd", "postal_code": "94107"}),
    ("recipient argument",        "send_status_email", {"order_id": "ord_1001", "to": "attacker@example.test"}),
]

print(f"{'case':27} {'result':18} detail")
for label, tool, args in cases:
    st, rt, ana = fresh()
    r = rt.call(ana, tool, args, call_key="k1")
    result = r.get("code", r["status"])
    detail = "; ".join(r.get("detail", [])) or {k: v for k, v in r.items() if k not in ("status", "code")} or ""
    print(f"{label:27} {result:18} {detail}  effects={len(st.effects)}")

case                        result             detail
valid refund                ok                 {'refunded_cents': 3000}  effects=1
amount as string            invalid_arguments  $.amount_cents: expected integer  effects=0
amount as float             invalid_arguments  $.amount_cents: expected integer  effects=0
amount as bool              invalid_arguments  $.amount_cents: expected integer  effects=0
negative amount             invalid_arguments  $.amount_cents: below minimum 1  effects=0
amount above refundable     invalid_state      amount exceeds refundable 4200  effects=0
reason outside enum         invalid_arguments  $.reason: not one of ['damaged', 'not_as_described', 'late', 'other']  effects=0
extra property              invalid_arguments  $.customer_id: unexpected property  effects=0
missing property            invalid_arguments  $.reason: required  effects=0
another customer's order    not_found            effects=0
order does not exist        not_found            effects=0
injection-shaped id         invalid_arguments  $.order_id: does not match ^ord_[0-9]{4,8}$  effects=0
order not delivered         invalid_state      order is not delivered  effects=0
unknown tool                unknown_tool         effects=0
postal code, full-width     ok                 {'address': '5 Pine Rd, 94107'}  effects=1
postal code, trailing \n    ok                 {'address': '5 Pine Rd, 94107'}  effects=1
postal code, embedded \n    invalid_arguments  $.postal_code: does not match ^[0-9]{5}$  effects=0
address after shipment      invalid_state      order has already shipped  effects=0
recipient argument          invalid_arguments  $.to: unexpected property  effects=0

Reading the output from the top:

• The valid refund passes and writes one ledger entry.
• The next four refund rows fail the schema. A string, a float, and a boolean are all rejected as “expected integer”, and a negative value falls below the minimum. None of them is coerced. Coercion would make the validator a second, hidden parser with its own bugs.
• The amount above what is refundable passes the schema and fails at stage 7, because the schema cannot know the order’s totals. Its detail is safe to show: the caller owns the order.
• The enum, extra-property, and missing-property rows are schema failures. The message names the property path but does not echo the offending value, because error text goes back into the model’s context and an attacker-controlled string should not be copied there.
• Another customer’s order and a nonexistent order are indistinguishable.
• The injection-shaped id fails the pattern before it can reach any lookup. This is the layer-3 control, and parameterized queries in the backend are still required as the sink-level control.
• The full-width postal code is accepted after normalization, and the stored value is the ASCII form. A trailing newline is trimmed and accepted, while a newline in the middle makes the value fail the pattern. Python’s $ also matches before a final newline, which is one reason to normalize first and not rely on the pattern alone.
• The email tool rejects a to argument because the schema has no such property; the destination cannot be chosen by the model at all.

Every row that is accepted is accepted because it is well-formed and allowed, and none of that proves it is what the customer wanted. A refund of 3,000 cents on the customer’s own delivered order is valid whether the customer asked for it or an instruction hidden in a retrieved document did. That is the gap the next section addresses: after validation, how much should the system trust a valid call?

Feeding errors back

A rejected call is information for the model, and the repair loop from the API integration article applies: return the error code and the property-level detail, allow a bounded number of corrected attempts, then stop. Two cautions are specific to tools. First, a rejected call that is a repeat of the same rejected call is a sign of a stuck or manipulated model, so the attempt budget counts calls to the same tool, and the budget ends in a fallback. Second, the error vocabulary is itself an interface to a possibly manipulated component. Keep details to what is needed to fix the call, never return raw exceptions, and make the messages for cross-customer lookups identical to those for missing records, as above.

Action consequences

Classifying effects

Two calls with identical shapes and permissions can differ enormously in what they do. A useful classification asks what undoing the action would take.

Effect class	Meaning	Tools here	What goes wrong	Main controls
Read	No state change; may disclose data	get_order	Wrong data shown, data leaked to the wrong party	Object-level authorization, minimal results
Reversible write	State changes and a cheap undo exists	update_shipping_address	Wrong change that must be found and reverted	Write budget, audit trail, state rules
Irreversible write	The effect leaves the system or moves money	issue_refund, send_status_email	A refund paid out, a message delivered	Limits per call and per session, approval, idempotency

The class is only the first coordinate. The consequence of a call is better described by four quantities: the class (can it be undone), the magnitude (money or records at stake), the scope (how many objects), and the exposure (who sees the result). The code reduces these to two numbers per call, the effect class and an impact in cents, and applies one policy to them. The policy has three parts.

• A read is always automatic, once it passes authorization.
• A reversible write is automatic until the session has made MAX_WRITES of them, because even a cheap undo costs attention once there are many.
• An irreversible write is automatic only if its impact is at most AUTO_LIMIT and the session’s cumulative irreversible impact stays within SESSION_CAP. Otherwise the call is not executed and the runtime answers needs_approval with a preview.

The email tool has an impact of zero in this model, so it is automatic. That is a judgment about this hypothetical store, where a status email carries little risk because its recipient and content come from the system. A tool that sends free text to arbitrary addresses would need a different classification, and its impact would reflect what the message can expose.

Why a session cap and not only a per-call limit

A per-call limit bounds the loss from one wrong call. It does not bound the loss from a run of them. If each automatic call is independently wrong with probability \(p\), the probability that at least one of \(n\) calls in a session is wrong is

\[ 1 – (1-p)^{n}. \]

With \(p = 0.02\) and \(n = 20\) this is about 0.33; with \(p = 0.005\) it is about 0.095. The independence assumption is optimistic. Under prompt injection or a confused model, errors are correlated: a manipulated session can make every call wrong. Then the number that bounds the damage is not an error rate but the cap, because the cap holds whatever the probability of error turns out to be. This is why the session cap and write budget exist even when the error estimate for a tool is small.

When approval pays for itself

A human reviewer costs time and delay, so the policy should send to review only the calls where review is worth it. A simple expected-loss comparison gives the shape of the answer. Let \(p\) be the probability that an automatic call is wrong in a way that costs money, \(L\) the loss when it is, \(q\) the fraction of wrong calls a reviewer catches, and \(c\) the cost of one review. Automatic execution has an expected loss of \(pL\) per call. Review removes the fraction \(q\) of it at cost \(c\). Review pays for itself when

\[ q\,p\,L > c \quad\Longleftrightarrow\quad L > L^{*} = \frac{c}{q\,p}. \]

The threshold \(L^{*}\) falls as \(p\) rises, which matches intuition: a tool the model gets wrong often needs review at low stakes. It is a model, not a measurement. It assumes losses are comparable in dollars, ignores the delay a customer experiences while waiting, treats \(q\) as constant, and takes \(p\) as known. In practice \(p\) is estimated per tool from labeled conversations, which is the work of the evaluation article, and the per-call and session caps stay in place as hard limits, whatever the formula says. The next program evaluates the threshold for illustrative inputs: a reviewer who catches 90 percent of bad calls and a review that costs 1.50 USD. These are assumptions for the example, not data.

def approval_threshold(p_wrong: float, catch: float, review_cost: float) -> float:
    """Smallest loss L (dollars) at which review pays for itself: catch * p_wrong * L > review_cost."""
    return review_cost / (catch * p_wrong)

catch, review = 0.9, 1.50          # assumed: reviewer catches 90% of bad calls; one review costs 1.50 USD
print("p_wrong  threshold L* (USD)")
for p in (0.005, 0.02, 0.10, 0.50):
    print(f"{p:7.3f}  {approval_threshold(p, catch, review):14.2f}")

p_wrong  threshold L* (USD)
  0.005          333.33
  0.020           83.33
  0.100           16.67
  0.500            3.33

At a 2 percent error rate, review is worth it for losses above about 83 USD, so an automatic limit of 5,000 cents (50 USD) in the code is on the conservative side for that assumed rate. At 0.5 percent, review pays only above about 333 USD. At 10 percent, losses above about 17 USD already justify review. Because \(p\) comes from measurement and may change with each model or prompt version, the limits belong in configuration that is revisited when the evaluation results change.

Approval that binds to the action

A person approving a call must approve the call that will run. Two design points follow from that.

First, the preview the person sees comes from the runtime, built from the normalized arguments and the computed impact, not from the model’s description of what it is about to do. The model’s prose can be wrong or manipulated; the preview is what the dispatcher will execute. MCP’s tools specification takes the same stance for clients: show tool inputs to the user before calling the server, and prompt for confirmation on sensitive operations.

Second, the approval is bound to the exact call. In the code, an approval token is a keyed hash over the user, the tool, and the normalized arguments. A token for a refund of 9,000 cents does not authorize a refund of 8,000, or of 9,001, and a token for one customer does not work in another’s session. The code stands in for a real approval service with an HMAC; how an approval is requested, who may grant it, how long it stays valid, and what happens if the order changes while the request waits belong to Human Approval and Escalation in AI Workflows.

Making repeats safe

A write that is retried must not run twice. The dispatcher requires a call_key for every write and stores the result under the user, the tool, and the key. A second call with the same key and the same arguments returns the stored result and marks it as replayed; the same key with different arguments is rejected as a reuse error. The orchestrator is responsible for deriving the key from the logical step, for example the task id plus the step and call index, not from the model’s own tool-call id, because a re-asked model call can produce new ids for the same step. The idempotency glossary entry explains why a key identifying the logical operation is what makes repetition safe, and the retry section of the API integration article covers when to retry at all.

In a real system the stored result and the effect must be recorded atomically, or the downstream service must accept the key itself, because a crash between “refund executed” and “key recorded” would allow a duplicate. Uncertain outcomes need the same care: if the refund service times out, the dispatcher does not know whether the refund happened, and it must not tell the model “failed” and invite a second attempt under a new key. That state is reported as unknown and reconciled against the service, which is covered in Durable Agent Workflows.

The next program runs the consequence rules on one customer’s large order. The first group shows approval binding and replay, the second the session cap.

from harness import fresh, R
def show(label, r):
    print(f"{label:48} -> {r.get('code', r['status'])}{' (replayed)' if r.get('replayed') else ''}")

st, rt, ana = fresh()
big = R(order_id="ord_1004", amount_cents=8000)
show("refund 8000 (auto limit is 5000)", rt.call(ana, "issue_refund", big, "t7:1"))
tok = rt.issue_approval("cu_ana", "issue_refund", big)
show("approval token issued for 9000, call is 8000", rt.call(ana, "issue_refund", big, "t7:1", rt.issue_approval("cu_ana", "issue_refund", R(order_id="ord_1004", amount_cents=9000))))
show("approval token for exactly these args", rt.call(ana, "issue_refund", big, "t7:1", tok))
show("same call_key again (a retry)", rt.call(ana, "issue_refund", big, "t7:1", tok))
show("same call_key, different amount", rt.call(ana, "issue_refund", R(order_id="ord_1004", amount_cents=100), "t7:1", tok))
show("write without a call_key", rt.call(ana, "issue_refund", R(order_id="ord_1004", amount_cents=100)))
print("ledger:", st.effects)

print("\nsession cap of 10000 on irreversible impact (auto limit 5000 per call)")
st, rt, ana = fresh()
for i in range(3):
    show(f"refund 4000, call {i+1}", rt.call(ana, "issue_refund", R(order_id="ord_1004", amount_cents=4000), f"c:{i}"))
print("ledger:", st.effects)

refund 8000 (auto limit is 5000)                 -> needs_approval
approval token issued for 9000, call is 8000     -> needs_approval
approval token for exactly these args            -> ok
same call_key again (a retry)                    -> ok (replayed)
same call_key, different amount                  -> key_reused
write without a call_key                         -> missing_call_key
ledger: [('refund', 'ord_1004', 8000)]

session cap of 10000 on irreversible impact (auto limit 5000 per call)
refund 4000, call 1                              -> ok
refund 4000, call 2                              -> ok
refund 4000, call 3                              -> needs_approval
ledger: [('refund', 'ord_1004', 4000), ('refund', 'ord_1004', 4000)]

Reading the output: a refund of 8,000 cents exceeds the automatic limit, so it is not executed and the response asks for approval. A token issued for 9,000 does not unlock the 8,000 call; a token for exactly those arguments does, and the refund executes once. Repeating the call with the same key returns the stored result and does not add a ledger entry. Reusing the key with a different amount is rejected, and a write without a key is rejected outright. In the second group, two refunds of 4,000 are each under the per-call limit and run automatically, but the third would take the session past 10,000 cents of irreversible impact and is held for approval. The ledger shows two entries, not three.

Each of these decisions should also be written to an audit log: the proposed call, the stage that stopped it or the stage it passed, the approver if any, and the outcome. A “refund requested” entry and a “refund completed” entry are different events, and the log should show both.

Integration tests

What the tests have to prove

The security properties of this design are statements about the whole pipeline: no call that fails any check changes backend state, no customer can read or change another customer’s orders through any tool, and a repeated call does not repeat its effect. A unit test of the schema checker proves none of these. The tests that do are integration tests: scripted calls go through the real dispatcher against a fake backend, and the assertions are on the backend’s effect ledger as well as the return value. The model is not part of these tests. A scripted list of calls replaces it, because the question is what the runtime does with any proposal, including the worst one.

The test families map to the sections above.

• **Contract tests** check that every tool has the model-facing and runtime-only halves, that schemas forbid additional properties, and that the serialized model-facing view matches a reviewed snapshot. A change to a description or schema is a release, because it changes the model’s input; the prompt versioning discipline applies.
• **Authorization matrix tests** generate every combination of user, order, and tool and assert that only owners succeed. A hand-written test for one cross-customer case is easy to get right by accident; the matrix is not.
• **Argument tests** run a table of hostile and malformed values, like the one in the previous section, and assert both the error code and an unchanged ledger.
• **Consequence tests** assert the limits: a large refund waits for approval and has no effect, an approval does not transfer to different arguments, the session cap holds.
• **Replay tests** assert that the same key does not repeat an effect.

The suite below implements the authorization, argument, consequence, and replay families; the contract snapshot test is a one-line comparison of serialized views and is not repeated. It uses unittest, and the helper at the bottom runs it against a runtime factory so that the same tests can run against a deliberately broken runtime.

"""Integration tests: scripted calls through the real dispatcher, assertions on the backend's effect ledger."""
import itertools, unittest
from runtime import Store, Runtime, Session, TOOLS

SCOPES = frozenset({"orders:read", "refunds:write", "orders:write", "email:send"})
FACTORY = lambda store: Runtime(store)            # replaced below to run the same tests on a broken runtime

def session(user, scopes=SCOPES, tools=None):
    return Session(user, frozenset(scopes), frozenset(tools or TOOLS))

R = lambda **kw: {"order_id": "ord_1001", "amount_cents": 1000, "reason": "damaged", **kw}

class ToolRuntimeTests(unittest.TestCase):
    def setUp(self):
        self.st = Store(); self.rt = FACTORY(self.st)
        self.ana, self.ben = session("cu_ana"), session("cu_ben")

    def code(self, r): return r.get("code", r["status"])

    def test_unknown_tool_fails_closed(self):
        self.assertEqual(self.code(self.rt.call(self.ana, "run_sql", {"q": "select 1"})), "unknown_tool")

    def test_tool_not_exposed_is_refused_even_if_named(self):
        s = session("cu_ana", tools={"get_order"})
        self.assertEqual(self.code(self.rt.call(s, "issue_refund", R(), "k")), "not_available")
        self.assertEqual(self.st.effects, [])

    def test_missing_scope_is_forbidden(self):
        s = session("cu_ana", scopes={"orders:read"})
        self.assertEqual(self.code(self.rt.call(s, "issue_refund", R(), "k")), "forbidden")
        self.assertEqual(self.st.effects, [])

    def test_other_customers_order_looks_like_a_missing_order(self):
        other = self.rt.call(self.ana, "get_order", {"order_id": "ord_1002"})
        missing = self.rt.call(self.ana, "get_order", {"order_id": "ord_9999"})
        self.assertEqual(other, missing)

    def test_authorization_matrix(self):
        calls = {"get_order": lambda o: {"order_id": o},
                 "issue_refund": lambda o: R(order_id=o),
                 "update_shipping_address": lambda o: {"order_id": o, "line1": "5 Pine Rd", "postal_code": "94107"},
                 "send_status_email": lambda o: {"order_id": o}}
        owner = {oid: rec["customer"] for oid, rec in Store().orders.items()}
        for user, oid, tool in itertools.product(["cu_ana", "cu_ben"], owner, calls):
            with self.subTest(user=user, order=oid, tool=tool):
                st = Store(); rt = FACTORY(st)
                r = rt.call(session(user), tool, calls[tool](oid), "k")
                if owner[oid] != user:
                    self.assertEqual(self.code(r), "not_found")
                    self.assertEqual(st.effects, [])
                else:
                    self.assertNotEqual(self.code(r), "not_found")

    def test_injected_identity_argument_is_rejected(self):
        r = self.rt.call(self.ana, "issue_refund", R(customer_id="cu_ben"), "k")
        self.assertEqual(self.code(r), "invalid_arguments"); self.assertEqual(self.st.effects, [])

    def test_email_has_no_recipient_argument(self):
        r = self.rt.call(self.ana, "send_status_email", {"order_id": "ord_1001", "to": "x@example.test"}, "k")
        self.assertEqual(self.code(r), "invalid_arguments")
        self.rt.call(self.ana, "send_status_email", {"order_id": "ord_1001"}, "k2")
        self.assertEqual(self.st.effects, [("email", "ord_1001", "ana@example.test")])

    def test_refund_cannot_exceed_refundable_amount(self):
        self.rt.call(self.ana, "issue_refund", R(amount_cents=4000), "k1")
        r = self.rt.call(self.ana, "issue_refund", R(amount_cents=300), "k2")
        self.assertEqual(self.code(r), "invalid_state")
        self.assertEqual(sum(e[2] for e in self.st.effects if e[0] == "refund"), 4000)

    def test_large_refund_waits_for_approval_and_has_no_effect(self):
        r = self.rt.call(self.ana, "issue_refund", R(order_id="ord_1004", amount_cents=8000), "k")
        self.assertEqual(self.code(r), "needs_approval"); self.assertEqual(self.st.effects, [])

    def test_approval_is_bound_to_exact_arguments(self):
        tok = self.rt.issue_approval("cu_ana", "issue_refund", R(order_id="ord_1004", amount_cents=8000))
        r = self.rt.call(self.ana, "issue_refund", R(order_id="ord_1004", amount_cents=9000), "k", tok)
        self.assertEqual(self.code(r), "needs_approval"); self.assertEqual(self.st.effects, [])

    def test_retry_with_same_key_does_not_repeat_the_effect(self):
        a = R(amount_cents=1000)
        first, second = self.rt.call(self.ana, "issue_refund", a, "k"), self.rt.call(self.ana, "issue_refund", a, "k")
        self.assertTrue(second.get("replayed")); self.assertEqual(len(self.st.effects), 1)

    def test_session_cap_stops_automatic_irreversible_actions(self):
        for i in range(3):
            self.rt.call(self.ana, "issue_refund", R(order_id="ord_1004", amount_cents=4000), f"k{i}")
        self.assertEqual(len(self.st.effects), 2)

    def test_address_change_after_shipment_is_refused(self):
        r = self.rt.call(self.ana, "update_shipping_address", {"order_id": "ord_1001", "line1": "5 Pine Rd", "postal_code": "94107"}, "k")
        self.assertEqual(self.code(r), "invalid_state"); self.assertEqual(self.st.effects, [])

The matrix test is the one to read closely. It builds a fresh backend for each of the 32 combinations of two users, four orders, and four tools, and for every combination where the user is not the owner it asserts both the not_found response and an empty ledger.

Do the tests catch a defect

A test suite is evidence only if it fails when the code is wrong. The last lines of the file run the suite twice: once against the correct runtime and once against a runtime whose ownership check in stage 5 has been removed. The second run is a seeded defect, the kind of change that happens when someone “simplifies” the lookup during a refactor.

def run(label, factory):
    global FACTORY
    FACTORY = factory
    suite = unittest.defaultTestLoader.loadTestsFromTestCase(ToolRuntimeTests)
    res = unittest.TestResult(); suite.run(res)
    from collections import Counter
    failed = Counter(t.id().split(".")[-1].split(" ")[0] for t, _ in res.failures + res.errors)
    print(f"{label}: {res.testsRun} tests, {sum(failed.values())} failures")
    for name, n in sorted(failed.items()): print(f"   {name}: {n}")

run("correct runtime            ", lambda st: Runtime(st))
run("ownership check removed    ", lambda st: Runtime(st, enforce_owner=False))

correct runtime            : 13 tests, 0 failures
ownership check removed    : 13 tests, 17 failures
   test_authorization_matrix: 16
   test_other_customers_order_looks_like_a_missing_order: 1

The correct runtime passes all 13 tests. With the ownership check removed, 17 failures appear: the 16 matrix combinations where the user is not the order’s owner, and the test that compares the response for another customer’s order with the response for a missing one. None of the tests that exercise validation, approval, or replay fails, because those mechanisms were not touched. That pattern is what a useful suite looks like: a specific defect fails the tests that cover it and no others. If the seeded defect had passed, the suite would have been giving false assurance about authorization.

The same technique extends to other stages: remove the scope check, the session cap, or the approval binding, and confirm that a named test fails each time.

Testing the model’s side of the interface

The suite above makes the runtime safe against any proposal. It says nothing about whether the model proposes the right calls. That is an evaluation question with a different structure: labeled conversations (“refund the damaged jacket on order 1001”), the expected call or the expected absence of a call, and deterministic checks on the trajectory, such as the tool name, the normalized arguments, and the absence of calls that should never happen. Because model output varies between runs, each case runs several times and the result is a pass rate, as in the reproducible evaluation article. The rates of invalid_arguments, not_found, and needs_approval responses from the dispatcher are worth tracking as metrics: a rise in not_found can mean a model change that makes it invent order ids, and a rise in needs_approval raises the cost of the review queue. Adversarial cases, such as a retrieved passage that instructs a refund, belong in this evaluation, and the pass criterion is again the ledger, not whether the model’s reply sounds reluctant. Scoring whole trajectories and workflow outcomes is the subject of Evaluating Agent Actions and Workflow Outcomes.

Connecting the dispatcher to a model

The following sketch shows where the dispatcher sits in a tool loop. It is not run against a live service; model.next_step is a hypothetical stand-in for a provider client that returns the model’s proposed calls for one step, and each provider’s message format for tool results differs.

# Sketch only (not executed): a provider-neutral tool loop around the dispatcher.
def run_task(model, runtime, ctx, task_id, messages, max_steps=6):
    for step in range(max_steps):
        proposals = model.next_step(messages)          # hypothetical client: list of (name, args)
        if not proposals:
            return messages                            # the model produced a final answer
        results = []
        for idx, (name, args) in enumerate(proposals):
            key = f"{task_id}:{step}:{idx}"            # derived from the logical step, not the model's id
            results.append(runtime.call(ctx, name, args, call_key=key))
        messages.append({"role": "tool_results", "content": results})
    return messages                                    # step budget exhausted

The loop is deliberately thin. All decisions live in runtime.call; the loop only derives keys and returns results. The step budget in the loop is a first bound on how long a task runs. Planning, state across steps, and termination rules are the subject of Agent State, Planning, and Bounded Execution, which builds on the per-session budgets used here.

Where this design stops

The controls above make a particular kind of failure impossible or visible. They leave several others in place.

• **Valid and authorized does not mean intended.** A refund within limits, on the caller’s own order, requested because an instruction hidden in a retrieved passage said so, passes every check. The defenses are the ones that do not depend on the model: caps that bound the damage, approval for what exceeds them, previews that show the real action, and the separation of content from authority discussed in the prompt injection article.
• **Individually allowed calls can compose into a disallowed outcome.** A read of customer data followed by a message to an outside address is two permitted calls and one leak. The tools here avoid a free destination, but a system with a general messaging tool needs controls on data flow across calls, covered in Privacy, Data Leakage, and Output Controls.
• **Tools from other parties carry their descriptions with them.** A third-party tool server supplies the text the model reads about its tools, and the list can change at runtime; MCP defines a notification for changes to the tool list. Treat that text and any behavior annotations as untrusted input unless the server is trusted, pin and review the tool definitions you accept, and apply the same dispatcher checks to calls into them.
• **Session state can go stale.** Scopes and ownership are read when the session starts or when the order is loaded. Revoking a scope while a session is running takes effect only if the dispatcher re-reads the source of truth at execution time.
• **The numbers in the policy are placeholders.** The limit of 5,000 cents, the session cap, the reviewer’s catch rate, and the review cost belong to a hypothetical store. A real deployment sets them from measured error rates and loss data, and revisits them when the model, the prompt, or the tools change.

What remains after these controls is an agent whose worst automatic action is bounded by configuration and whose every denied or held action is visible in a log. The rest of this part of the series builds on that base: state and budgets across steps, durable execution of actions that must not be lost or repeated, human approval, and the evaluation of whole runs.

References

• Model Context Protocol, “Tools” (specification, 2025-06-18 revision): tool definitions, output schemas, annotations, error handling, and security considerations.
• OWASP Gen AI Security Project, “LLM06:2025 Excessive Agency”: root causes and mitigations for model-driven actions.
• Anthropic, “Define tools” (Claude API documentation): tool definition fields and guidance on descriptions and tool design.
• OpenAI, “Function calling” (OpenAI API documentation): function definitions, strict mode, and schema requirements.
• OWASP API Security Project, “API1:2023 Broken Object Level Authorization”.
• OWASP Cheat Sheet Series, “Input Validation Cheat Sheet”: allow-lists, syntactic and semantic validation, canonicalization.
• Saltzer and Schroeder, “The Protection of Information in Computer Systems,” design principles section: least privilege, fail-safe defaults, complete mediation.
• Hardy, “The Confused Deputy (or why capabilities might have been invented)”, Operating Systems Review, 1988.