Skip to main content

BudgetGuard

class actguard.BudgetGuard(
    *,
    user_id: str,
    token_limit: int | None = None,
    usd_limit: float | None = None,
)
Context manager that tracks cumulative token and USD usage for LLM API calls made within its block. Supports both sync (with) and async (async with) usage.

Parameters

ParameterTypeDefaultDescription
user_idstrIdentifier for the budget owner. Included in BudgetExceededError messages and on the exception object.
token_limitint | NoneNoneMaximum cumulative tokens (input + output combined). Raises BudgetExceededError when tokens_used >= token_limit.
usd_limitfloat | NoneNoneMaximum cumulative USD cost. Raises BudgetExceededError when usd_used >= usd_limit.
At least one limit should be set. If both are None, usage is tracked but no error is ever raised.

Properties

PropertyTypeDescription
tokens_usedintTotal tokens consumed so far (input + output across all calls).
usd_usedfloatTotal USD cost so far.
user_idstrThe user_id passed to the constructor.
token_limitint | NoneThe token_limit passed to the constructor.
usd_limitfloat | NoneThe usd_limit passed to the constructor.

BudgetExceededError

class actguard.BudgetExceededError(Exception)
Raised by BudgetGuard when a token or USD limit is exceeded.

Attributes

AttributeTypeDescription
user_idstrThe user ID from the active BudgetGuard.
tokens_usedintTotal tokens consumed at the moment the limit was hit.
usd_usedfloatTotal USD cost at the moment the limit was hit.
token_limitint | NoneThe token limit that was set (may be None if no token limit).
usd_limitfloat | NoneThe USD limit that was set (may be None if no USD limit).
limit_typeLiteral["token", "usd"]Which limit triggered the error.

Tool Runtime

RunContext

class actguard.RunContext(*, run_id: str | None = None)
Context manager for tool runtime state (max_attempts, idempotent). Supports sync and async usage.

Constructor params

ParameterTypeDefaultDescription
run_idstr | Noneauto-generated UUIDOptional explicit run identifier used by runtime exceptions

Methods

MethodSignatureDescription
enter__enter__() -> RunContextActivate run state
exit__exit__(exc_type, exc_val, exc_tb) -> NoneRestore previous run state
async enter__aenter__() -> RunContextAsync enter
async exit__aexit__(exc_type, exc_val, exc_tb) -> NoneAsync exit
attemptsget_attempt_count(tool_id: str) -> intReturn current attempt count for a tool id in this context

Tool Guards

configure()

actguard.configure(config: str | None = None) -> None
Wires in the ActGuard gateway for global enforcement reporting. Call once at startup; optional.

@rate_limit

actguard.rate_limit(
    *,
    max_calls: int = 10,
    period: float = 60.0,
    scope: str | None = None,
)
Decorator that enforces a sliding-window call-rate limit on sync and async functions.
ParameterTypeDefaultDescription
max_callsint10Maximum calls allowed within period
periodfloat60.0Sliding-window length in seconds
scopestr | NoneNoneArgument name used as counter key; None means one global counter

FailureKind

class actguard.FailureKind(str, Enum)
Stable failure taxonomy used by @circuit_breaker. Members:
  • TRANSPORT
  • TIMEOUT
  • OVERLOADED
  • THROTTLED
  • AUTH
  • INVALID
  • NOT_FOUND
  • CONFLICT
  • UNKNOWN

Preset constants

actguard.FAIL_ON_DEFAULT
actguard.IGNORE_ON_DEFAULT
actguard.FAIL_ON_STRICT
actguard.FAIL_ON_INFRA_ONLY
Set-like presets of FailureKind values.

@circuit_breaker

actguard.circuit_breaker(
    *,
    name: str,
    max_fails: int = 3,
    reset_timeout: float = 60.0,
    fail_on: set[FailureKind] = FAIL_ON_DEFAULT,
    ignore_on: set[FailureKind] = IGNORE_ON_DEFAULT,
)
Circuit breaker decorator for sync and async functions.
ParameterTypeDefaultDescription
namestrrequiredDependency name
max_failsint3Counted failures before OPEN
reset_timeoutfloat60.0Seconds before calls are allowed again
fail_onset[FailureKind]FAIL_ON_DEFAULTKinds that increment/open
ignore_onset[FailureKind]IGNORE_ON_DEFAULTKinds that do not affect breaker state

@max_attempts

actguard.max_attempts(*, calls: int)
Decorator that limits invocations per tool per active RunContext.
ParameterTypeDefaultDescription
callsintrequiredMax allowed attempts per run
Raises MissingRuntimeContextError if no RunContext is active.

@timeout

actguard.timeout(seconds: float, executor: Executor | None = None)
Decorator that bounds tool wall-clock runtime.
ParameterTypeDefaultDescription
secondsfloatrequiredTimeout threshold in seconds
executorExecutor | NoneNoneOptional executor for sync tool execution
Raises ToolTimeoutError when exceeded.

shutdown()

actguard.shutdown(wait: bool = True) -> None
Shuts down the shared timeout executor used by @timeout when no custom executor is passed.

@idempotent

actguard.idempotent(
    *,
    ttl_s: float = 3600,
    on_duplicate: Literal["return", "raise"] = "return",
    safe_exceptions: tuple = (),
)
Decorator enforcing at-most-once execution per (tool_id, idempotency_key) in an active RunContext.
ParameterTypeDefaultDescription
ttl_sfloat3600TTL for idempotency entries
on_duplicate"return" | "raise""return"Return cached result or raise
safe_exceptionstuple()Exceptions that clear state and allow retry
Requires the decorated function to include an idempotency_key parameter.

Chain-of-custody APIs

session()

actguard.session(id: str, scope: dict[str, str] | None = None) -> GuardSession
Context manager for chain-of-custody state used by @prove and @enforce. Supports sync and async usage.
ParameterTypeDefaultDescription
idstrrequiredSession identifier (request/run correlation id)
scopedict[str, str] | NoneNoneOptional scope dimensions (for example {"user_id": "u42"})

GuardSession

Context manager returned by session(...).
MethodSignatureDescription
enter__enter__() -> GuardSessionActivate session state
exit__exit__(...) -> NoneRestore previous session state
async enter__aenter__() -> GuardSessionAsync enter
async exit__aexit__(...) -> NoneAsync exit

@prove

actguard.prove(
    kind: str,
    extract: str | Callable,
    ttl: float = 300,
    max_items: int = 200,
    on_too_many: str = "block",
)
Decorator that mints verified facts from a tool’s return value.
ParameterTypeDefaultDescription
kindstrrequiredFact kind/category
extractstr | CallablerequiredField/attribute name or callable extractor
ttlfloat300Fact TTL in seconds
max_itemsint200Maximum values to mint per invocation
on_too_many"block" | "truncate""block"Block with GuardError or truncate extracted values
Requires an active actguard.session(...). Without it, raises GuardError(code="NO_SESSION").

@enforce

actguard.enforce(rules: list[Rule])
Decorator that checks chain-of-custody rules before tool execution.
ParameterTypeDefaultDescription
ruleslist[Rule]requiredRule objects evaluated in order
Requires an active actguard.session(...). Without it, raises GuardError(code="NO_SESSION").

RequireFact

actguard.RequireFact(arg: str, kind: str, hint: str = "")
Rule requiring argument value(s) to have been proven in active session scope.

Threshold

actguard.Threshold(arg: str, max: float)
Rule enforcing a maximum numeric value for an argument.

BlockRegex

actguard.BlockRegex(arg: str, pattern: str)
Rule blocking argument values that match a regex pattern. In-memory store semantics:
  • Fact verification state is in-memory, process-local, and ephemeral.
  • Facts are isolated by session id and scope hash.
  • State is not durable across restarts and is not shared across processes.

@tool

actguard.tool(
    *,
    rate_limit: dict | None = None,
    circuit_breaker: dict | None = None,
    max_attempts: dict | None = None,
    timeout: float | None = None,
    timeout_executor: Executor | None = None,
    idempotent: dict | None = None,
    policy: ... = None,
)
Unified decorator that composes multiple guards.
KwargTypeDescription
rate_limitdict | Nonemax_calls, period, scope
circuit_breakerdict | Nonename, max_fails, reset_timeout, fail_on, ignore_on
max_attemptsdict | Nonecalls
timeoutfloat | NoneTimeout in seconds
timeout_executorExecutor | NoneCustom executor for sync timeout execution
idempotentdict | Nonettl_s, on_duplicate, safe_exceptions
policyReserved stub
Execution order: idempotent -> max_attempts -> circuit_breaker -> rate_limit -> timeout -> fn

Exceptions

ToolGuardError

class actguard.ToolGuardError(Exception)
Base class for guard-blocked tool execution.

GuardError

class actguard.GuardError(ToolGuardError)
Raised by @prove and @enforce.
AttributeType
codestr
messagestr
detailsdict
fix_hintstr | None
Common code values: NO_SESSION, MISSING_FACT, TOO_MANY_RESULTS, THRESHOLD_EXCEEDED, PATTERN_BLOCKED.

ToolExecutionError

class actguard.ToolExecutionError(Exception)
Base class for tool execution failures.

RateLimitExceeded

class actguard.RateLimitExceeded(ToolGuardError)
AttributeType
func_namestr
scope_valuestr | None
max_callsint
periodfloat
retry_afterfloat

CircuitOpenError

class actguard.CircuitOpenError(ToolGuardError)
AttributeType
dependency_namestr
reset_atfloat
retry_afterfloat

MissingRuntimeContextError

class actguard.MissingRuntimeContextError(ToolGuardError)
Raised when run-state decorators are called without RunContext.

MaxAttemptsExceeded

class actguard.MaxAttemptsExceeded(ToolGuardError)
AttributeType
run_idstr
tool_namestr
limitint
usedint

ToolTimeoutError

class actguard.ToolTimeoutError(ToolExecutionError)
AttributeType
tool_namestr
timeout_sfloat
run_idstr | None

InvalidIdempotentToolError

class actguard.InvalidIdempotentToolError(ActGuardError)
Raised when a decorated function lacks idempotency_key.

MissingIdempotencyKeyError

class actguard.MissingIdempotencyKeyError(ToolGuardError)
Raised when idempotency_key is empty or missing.

IdempotencyInProgress

class actguard.IdempotencyInProgress(ToolGuardError)
Raised when the same key is currently in progress.

DuplicateIdempotencyKey

class actguard.DuplicateIdempotencyKey(ToolGuardError)
Raised on duplicate completed key when on_duplicate="raise".

IdempotencyOutcomeUnknown

class actguard.IdempotencyOutcomeUnknown(ToolGuardError)
Raised when previous unsafe failure left outcome unknown until TTL expiry.