API Reference

class actguard.Client(
    *,
    api_key: str | None = None,
    gateway_url: str | None = None,
    event_mode: str = "verbose",
    flush_interval_ms: int = 1000,
    max_batch_events: int = 100,
    max_batch_bytes: int = 256_000,
    max_queue_events: int = 10_000,
    timeout_s: float | None = None,
    max_retries: int | None = None,
    budget_timeout_s: float | None = None,
    budget_max_retries: int | None = None,
    event_timeout_s: float | None = None,
    event_max_retries: int | None = None,
    backoff_base_ms: int = 200,
    backoff_max_ms: int = 10_000,
)

actguard.Client.from_env() -> Client
actguard.Client.from_file(path: str | os.PathLike[str]) -> Client

client.run(user_id: str | None = None, run_id: str | None = None)
client.budget_guard(
    *,
    user_id: str | None = None,
    name: str | None = None,
    usd_limit: float | None = None,
    run_id: str | None = None,
    plan_key: str | None = None,
)
client.close() -> None

client.reserve_budget(
    *,
    run_id: str,
    usd_limit_micros: int | None,
    plan_key: str | None = None,
    user_id: str | None = None,
) -> str
client.settle_budget(
    *,
    reserve_id: str,
    provider: str,
    provider_model_id: str,
    input_tokens: int,
    cached_input_tokens: int,
    output_tokens: int,
) -> None

actguard.emit_event(category: str, name: str, payload: dict, **fields) -> None
actguard.emit_violation(error: Exception, **context_overrides) -> None

actguard.rate_limit(...)
actguard.circuit_breaker(...)
actguard.max_attempts(...)
actguard.timeout(...)
actguard.idempotent(...)
actguard.prove(...)
actguard.enforce(...)
actguard.tool(...)
actguard.shutdown(wait: bool = True)

actguard.session(id: str, scope: dict[str, str] | None = None)