Skip to main content

​
Basic usage

from agent_sentinel import guarded_action

@guarded_action(name="search_web", cost_usd=0.02, tags=["tool", "search"])
def search_web(query: str) -> dict:
    return {"results": ["..."], "query": query}
Calling search_web(...) will:
  • Measure duration
  • Record inputs/outputs
  • Append a ledger entry

​
Async functions

from agent_sentinel import guarded_action

@guarded_action(name="call_llm", cost_usd=0.05, tags=["llm"])
async def call_llm(prompt: str) -> str:
    return "..."

​
Exceptions

Agent Sentinel does not swallow your exceptions. If your function raises, the exception propagates normally — the SDK records the action as outcome="error" and logs the error string.

​
Fail-open vs fail-closed

  • Fail-open: ledger writes and remote sync should never crash your agent.
  • Fail-closed: policy violations intentionally raise before execution (see Policies).

​
Decorator arguments

@guarded_action accepts the following keyword-only arguments. Anything beyond name, cost_usd, and tags is optional — sensible defaults apply.
ArgumentTypePurpose
namestrLogical action name (defaults to function name)
cost_usdfloatEstimated per-call cost; deducted from budgets
tagslist[str]Free-form tags; matched against policy approval_tags
requires_human_approvalboolPause and request human approval before execution
approval_descriptionstrHuman-readable description shown in the approval inbox
agent_id / task_id / mission_idstrAttribution overrides; default to current ExecutionContext
produces_evidenceboolMark this action as producing evidence other actions can require
is_commitboolMark as a commit action (subject to evidence + grounding checks)
requireslist[str]Names of evidence actions that must precede this one
argument_constraintsdictJSON Schema applied to kwargs before execution
evidence_max_age_secondsintMaximum age of required evidence before it expires
grounding_rulesdictField-level grounding constraints (action arg → evidence field)
risk_levelstrOne of critical, high, medium, low, minimal
idempotency_keystr | CallableStatic key or (args, kwargs) -> key callable for replay caching
idempotency_ttl_secondsfloatTTL for the cached result (default 3600s)
@guarded_action(
    name="charge_card",
    cost_usd=0.01,
    risk_level="high",
    requires=["lookup_customer"],
    grounding_rules={"amount_cents": {"source_action": "lookup_customer", "source_field": "balance_cents"}},
    idempotency_key=lambda *_, **kw: kw["payment_id"],
    idempotency_ttl_seconds=900,
)
def charge_card(*, payment_id: str, customer_id: str, amount_cents: int) -> dict:
    return billing.charge(payment_id, amount_cents)

​
Replay mode integration

If replay mode is active, decorated actions return recorded outputs instead of executing (and are logged with outcome="replayed"). See Replay.