Security¶

What can go wrong when a model drives your tools, what the adapter stops on its own, and what your app must still do.

flowchart LR
  M["model output<br/>(untrusted)"] --> V["validate args<br/>strict schema"]
  V --> G["GuardrailHook<br/>your checks"]
  G --> A{"confirm?"}
  A -- yes --> P["approval port<br/>one decision, one call"]
  A -- no --> T["tool"]
  P --> T
  T --> R["result<br/>crash text hidden"]
  R --> L["logs and spans<br/>no args, no content"]

Threat model¶

Threat	Example	Control
Model-supplied arguments	wrong types, extra keys, huge values	every call is validated against the tool's schema first (`tool-args-invalid`, values never echoed)
Prompt injection via tool results	a fetched page says "now pay AC-666"	`confirm=` on risky tools, `GuardrailHook` on `args`, a `registry` subset per surface
Replayed approval	a resume sent twice, or by someone else	one decision runs one call, once; it is bound to the run, call id, tool and argument digest
Approval forged by a framework	an SDK flag says "approved"	never counts alone (below)
Arguments swapped after review	same call id, new account	refused: `call id reused with different tool/arguments`
Secrets in logs and traces	API keys in arguments, crash text with tokens	redacting `LoggingHook`; crash text only at `DEBUG`; content off in sinks
Another user's run (HTTP)	guessing a `run_id` or an `Idempotency-Key`	principal-scoped ids and run owners (below)

Framework approvals¶

A framework's own "approved" signal never runs a call by itself.

A bridge may only add a gate, never remove one.
For a confirm call, a bridge passes approval only as a one-use token that the adapter issued after its own approval port approved.
The token is bound to the run (or the framework's scope), the native call id, the tool and a digest of the validated arguments. It is taken once, atomically.
A call with no native call id gets no token: it is denied, or the executor asks its own port.
Without a token, the executor asks its port. MCP's client_approval=True is the one opt-in exception: the MCP client's user decides.

HTTP server¶

Rule	In plain words
Principal-scoped ids	the server derives call, run and thread ids from (who you are, your key): you cannot pick another user's id
One owner per run	the first principal to use a run owns it; others get 403 on invoke, events, cost and approvals
Who may decide	the run's owner, or a principal with the `approvals:decide` scope; pass your own rule for four-eyes
Re-checks on decide	`authorize` runs again for the decider (edits) and for the invoker (final arguments)
Anonymous mode	all anonymous callers share one identity and never join an authenticated run; it serves loopback peers only unless `allow_anonymous_public=True`

Every decision and residual risk is listed in the HTTP server security model.

See it fail closed¶

@registry.tool(confirm=True, effects="destructive")
def pay(account: str, amount: float) -> str:
    """Pay an account."""
    return f"paid {amount} to {account}"


@registry.tool
def fetch(url: str) -> str:
    """Fetch a page from an internal API."""
    raise RuntimeError(f"connect failed to {url} with token=abc123")  # text never leaks


def only_internal(stage: str, payload: Any) -> bool | str:
    """Blocks URLs a prompt injection may have planted in the arguments."""
    url = payload.get("url") if isinstance(payload, dict) else None
    return url is None or urlparse(url).hostname in ALLOWED_HOSTS or "host not allowed"

store = MemoryApprovalStore()
executor = Executor(
    registry,
    approval=DeferredApproval(store),  # no port at all would deny every confirm call
    hooks=[GuardrailHook(only_internal), LoggingHook(logger=log)],
)

ctx = executor.context(run_id="run_sec")
call = ToolCall(name="pay", arguments={"account": "AC-1", "amount": 10}, id="c1")
pending = await executor.run_call(call, ctx)
approval_id = pending.error.details["approval_id"] if pending.error else ""
await store.decide(approval_id, Decision(approved=True, decided_by="ana"))
print((await executor.run_call(call, ctx)).value)  # runs once
print((await executor.run_call(call, ctx)).status)  # a replay asks again
print(json.loads(lines.lines[0].split(" ", 1)[1]).get("args_withheld"))  # confirm args

swapped = ToolCall(name="pay", arguments={"account": "AC-666", "amount": 10}, id="c1")
refused = await executor.run_call(swapped, ctx)  # same call id, other arguments
print(failure(refused))

planted = ToolCall(name="fetch", arguments={"url": "http://evil.example/x"})
internal = ToolCall(name="fetch", arguments={"url": "http://api.internal.example/x"})
for result in await executor.run_many([planted, internal], ctx):
    print(failure(result))

paid 10.0 to AC-1
pending
True
tool-usage-error call id reused with different tool/arguments (tool 'pay', call 'c1')
guardrail-blocked guardrail blocked args for tool 'fetch': host not allowed
tool-failed tool 'fetch' failed after 1 attempt

The token in the crash text (token=abc123) reaches neither the result nor any log line above DEBUG.

What the adapter does, what your app does¶

The adapter guarantees	Your app must
validation before any tool code runs	mark risky tools `confirm=` and choose a real port
no approval port: every `confirm` call is denied	keep reviewers real: verify webhook signatures, map users to reviewers
one decision runs at most one call	use a durable store (`SQLiteApprovalStore`, `RedisApprovalStore`, `SQLAlchemyApprovalStore`, `DynamoDBApprovalStore`; Database stores) across processes
crash text and arguments stay out of results and loud logs	never put secrets in tool descriptions or prompts; keep `capture_content=False` in production
per-principal ids and run owners on the HTTP server	turn authentication on; set `allowed_hosts` on public hosts; use TLS and rate limits at a proxy
fail closed on corrupt store rows (`ApprovalStoreCorrupt`)	back up and protect the store like any audit table

What the adapter does not do¶

Not provided	Use
sandboxing tool code	containers, separate processes, least-privilege credentials
a secrets store	your vault; read secrets inside the tool, never take them as model arguments
content moderation	`GuardrailHook` on `llm_request` / `llm_response` with your classifier
rate limiting per user on the server	a reverse proxy; `RateLimitHook` per process

Files¶

File	Does
`examples/guide_security.py`	a confirm tool, a guardrail, a replay, a swapped call and a hidden crash
`tests/docs/test_guides.py`	runs it offline and checks the output