Guard
Manual checks
Call st.guard(...) directly when you're dispatching tools yourself — a custom agent framework, a background job, a hook outside the patched integrations.
import staso as st
tool_input = {"amount": 4200, "user_id": "u_42"}
result = st.guard("process_refund", tool_input)
if result.action == "block":
raise RuntimeError(result.reason)
if result.action == "modify":
tool_input = result.modified_input
actual_run_refund(**tool_input)Signature
st.guard(
tool_name: str,
tool_input: dict[str, Any],
*,
context: dict[str, Any] | None = None,
wait_for_escalation: bool = False,
escalation_poll_interval: float = 3.0,
escalation_timeout: float = 300.0,
timeout: float = 10.0,
fail_closed: bool | None = None,
) -> GuardResult| Parameter | Default | Description |
|---|---|---|
tool_name | required | Name of the tool. |
tool_input | required | Tool arguments. |
context | None | Override conversation_id, agent_name, trace_id, environment, workspace_id. |
wait_for_escalation | False | Poll until a human resolves an escalation. |
escalation_poll_interval | 3.0 | Seconds between polls. |
escalation_timeout | 300.0 | Max seconds to wait. |
timeout | 10.0 | HTTP request timeout. Override globally with STASO_GUARD_TIMEOUT. |
fail_closed | None | On transport failure, return block instead of allow. Defaults to STASO_GUARD_FAIL_CLOSED. |
GuardResult
| Field | Description |
|---|---|
action | allow, block, modify, escalate. |
reason | Human-readable reason from the rule. |
rule_name | First rule that triggered. |
severity | low, medium, high, critical. |
results | Per-rule breakdown. |
modified_input | Rewritten input (when modify). |
modifications | Diff of changes. |
escalation_id | Escalation handle (when escalate). |
latency_ms | Evaluation latency. |
rules_triggered | All rules that contributed. |
Fail-open vs fail-closed
Default: transport failures (network down, timeout, backend unreachable) return action="allow". A flaky observability backend should never brick production agents.
Opt into fail-closed for destructive or irreversible actions:
result = st.guard("delete_user", {"id": "u_42"}, fail_closed=True)Or process-wide:
export STASO_GUARD_FAIL_CLOSED=1With fail_closed=True, transport failure returns block with severity="high". In patched integrations, this raises GuardBlocked.
Catching GuardBlocked
Patched Anthropic and OpenAI raise staso.GuardBlocked when a tool call is blocked.
try:
resp = client.messages.create(...)
except st.GuardBlocked as e:
for bt in e.blocked_tools:
print(f"blocked {bt.tool_name}: {bt.result.reason}")The exception is raised after the entire batch is evaluated, so e.blocked_tools lists every blocked call — not just the first.
| Field | Description |
|---|---|
tool_name | First blocked tool's name. |
result | GuardResult for the first blocked tool. |
reason | Summary string. |
rules_triggered | Deduplicated union across all blocked tools. |
blocked_tools | List of BlockedTool(tool_name, result). |
Patterns
- Recover gracefully. Log the reason, return a safe default.
- Fall back to a safer tool. If
send_emailwas blocked for PII, trysend_email_redacted. - Re-ask the user. Surface
e.reason: "I can't do that because <reason>. Want me to try something else?" - Log and alert. Push
e.rules_triggeredinto your alerting pipeline. Frequent blocks mean a misbehaving agent or an over-eager rule.