API Reference

Main entry point. Connects to Postgres and initializes the enabled enforcement modules. Use as an async context manager for automatic start/close. Every module is opt-in via an enable_* flag — when False, the module is not constructed and accessing sdk.<module> raises AttributeError. verify_chain_on_read (v0.5.3) verifies the HMAC chain on every get_events() call. warn_on_no_wrappers (v0.5.3) emits a structlog warning at start() when no wrap_openai / wrap_anthropic has been registered. default_max_tokens (v0.5.3) sets the ceiling used by the projected budget gate when callers do not declare max_tokens. As of v0.6.2, halt enforcement spans scope + cost + gates + the wrap_openai / wrap_anthropic wrappers; calling any of those against a halted agent raises AgentHaltedError. cost_strict_unknown_models (v0.6.2, default True) flips CostModule to raise UnknownModelError on pricing lookups for unknown model IDs instead of silently zero-costing — set to False (plus a fallback rate) to restore v0.6.1 observe-only behaviour.

Returns: GovernanceSDK instance with the modules constructed for the enabled flags

Raises: ValueError if neither database_url nor api_key is provided, if the audit secret is too short / too weak, or if default_max_tokens is < 1

Note: v0.6.2 adds cost_strict_unknown_models, cost_unknown_model_fallback_usd_per_million, and enable_coverage. Halt enforcement now covers scope + cost + gates + wrappers (was scope-only through v0.5.4/v0.6.1). See the /governance/upgrade-to-0.6.2 guide for the four BREAKING-block defaults.

# Full enforcement (all modules, wrapper-coverage warning, projected budget ceiling)
async with GovernanceSDK(
    database_url=os.environ["DATABASE_URL"],
    default_max_tokens=4096,
    verify_chain_on_read=True,
) as sdk:
    await sdk.audit.log(AuditEvent(agent_id="x", kind="y"))

# Lightweight — no loop detection, no presence, audit-only
async with GovernanceSDK(
    database_url=os.environ["DATABASE_URL"],
    enable_loop=False,
    enable_presence=False,
    warn_on_no_wrappers=False,
) as sdk:
    ...  # sdk.loop / sdk.presence raise AttributeError on access

Synchronous wrapper around GovernanceSDK. Spins up a daemon thread with a dedicated asyncio event loop. All async module methods (audit.log, cost.track, scope.check, etc.) are exposed as blocking sync calls. Supports context-manager usage. The background thread is a daemon thread so it does not prevent interpreter shutdown.

Returns: GovernanceSDKSync instance with .audit, .scope, .cost, .gates, .loop, .presence, .contracts modules (all sync)

Raises: RuntimeError if the background event loop stops unexpectedly

from codeatelier_governance import GovernanceSDKSync, AuditEvent

with GovernanceSDKSync(database_url=os.environ["DATABASE_URL"]) as sdk:
    sdk.audit.log(AuditEvent(agent_id="my-agent", kind="tool.call"))
    sdk.scope.check("my-agent", tool="send_email")
    sdk.cost.check_or_raise("my-agent", session_id)

Log an audit event. Computes the HMAC chain atomically with the insert (advisory lock per session). Non-breaking: never raises on internal failure — returns a placeholder record instead.

Returns: AuditEventRecord (check record.is_placeholder to detect degraded state)

Note: Observation surface — never breaks the host call.

from codeatelier_governance.audit import AuditEvent

record = await sdk.audit.log(AuditEvent(
    agent_id="billing-agent",
    kind="llm.call",
    model="gpt-4o-mini",
    metadata={"prompt_tokens": 150},
))

Decorator that logs entry/exit events for an async function. Computes input_hash from args and output_hash from the return value. On exception, logs a .error event and re-raises.

Returns: The wrapped function's return value (unchanged)

Note: Async functions only. Observation surface.

@sdk.audit.track(kind="charge_card", agent_id="billing-agent")
async def charge(amount: int) -> str:
    return await stripe.charge(amount)

Walk the HMAC chain for an entire session. Re-verifies every row's HMAC against the secret. The compliance officer's one-click verification.

Returns: list[AuditEventRecord] in chain order

Raises: ChainIntegrityError at the first tampered row

chain = await sdk.audit.trace_session_chain(session_id)
for event in chain:
    print(f"{event.kind} — {event.agent_id} — hmac verified")

On-demand HMAC chain verification. Stable public API for compliance attestation and post-incident review. Supports partial-range verification via from_seq and to_seq — useful for verifying only the window relevant to a specific incident without paying the O(n) cost on the whole chain. Two-pass verification: recomputes each row's HMAC and checks prev_hash linkage (fork detection). Never returns False — clean chain returns True, any failure raises.

Returns: True on a clean chain

Raises: ChainIntegrityError with the offending sequence number on the first broken HMAC or prev_hash linkage

Note: Added in v0.5.3. Set verify_chain_on_read=True on GovernanceSDK to verify the chain automatically on every get_events() call.

# Full-chain verification for a compliance attestation run
await sdk.audit.verify_chain()

# Window verification — only events [100..200] in one session
await sdk.audit.verify_chain(session_id=sid, from_seq=100, to_seq=200)

# Automatic on-read verification
sdk = GovernanceSDK(database_url=..., verify_chain_on_read=True)
events = await sdk.audit.get_events(session_id=sid)  # raises on tamper

Register a scope policy for an agent. Call at app startup. Policies are also persisted to Postgres (best-effort) so the console GUI can display them.

from codeatelier_governance.scope import ScopePolicy

sdk.scope.register(ScopePolicy(
    agent_id="billing-agent",
    allowed_tools=frozenset({"read_invoice", "send_email"}),
    allowed_apis=frozenset({"GET https://api.stripe.com/v1/*"}),
    hidden_tools=frozenset({"delete_all", "admin_reset"}),
))

Check if the agent is allowed to call the given tool or API. Default-deny: an agent with no registered policy fails every check.

Raises: ScopeViolation (auto audit-logged) or PolicyNotRegistered

Note: Enforcement surface — raises by contract.

await sdk.scope.check(agent_id="billing-agent", tool="read_invoice")  # passes
await sdk.scope.check(agent_id="billing-agent", tool="delete_user")   # raises ScopeViolation

Register a budget policy for an agent. Caps: per_session_usd, per_session_tokens, per_agent_usd_daily, per_agent_tokens_daily, per_session_seconds. At least one cap required.

from codeatelier_governance.cost import BudgetPolicy

sdk.cost.register(BudgetPolicy(
    agent_id="billing-agent",
    per_session_usd=0.50,
    per_session_seconds=300,  # 5-minute time limit
    per_agent_usd_daily=10.00,
))

Pre-call enforcement: raises BudgetExceeded if any cap is hit, including session time limits. Fail-closed by default — if the cost store is unreachable, the call is denied.

Raises: BudgetExceeded (auto audit-logged)

Note: Enforcement surface — raises by contract. Pass cost_fail_open=True at SDK init for availability over safety.

await sdk.cost.check_or_raise("billing-agent", session_id)
# If this line runs, the budget is not exceeded — proceed with the LLM call

Post-call: record actual usage with explicit token count and USD. Counters are monotonic (negative deltas rejected). Multi-process correct via Postgres UPSERT. Use track_usage() instead if you want auto USD computation.

Note: Observation surface — never raises.

await sdk.cost.track("billing-agent", session_id, tokens=150, usd=0.003)

Register a loop detection policy for an agent. Defines the sliding window size, maximum calls allowed, and action to take on detection.

from codeatelier_governance.loop import LoopPolicy

sdk.loop.register(LoopPolicy(
    agent_id="my-agent",
    window_seconds=60,
    max_calls=5,
    action="raise",  # or "log"
))

Record a tool call and check for loops. If the same tool has been called more than max_calls times within the sliding window for this session, the configured action fires. Tool names are normalized to lowercase for case-insensitive detection.

Raises: LoopDetected (when action="raise" and threshold exceeded)

Note: Enforcement surface when action="raise". Emits a loop.detected audit event on detection.

from codeatelier_governance.loop import LoopDetected

try:
    await sdk.loop.record_call("my-agent", session_id, "search_web")
except LoopDetected:
    # Agent called search_web 5+ times in 60s — break the loop
    break

Read-only check: returns whether the agent is currently in a detected loop state for the session. Does not record a new call.

Returns: bool — True if a loop is currently detected

Note: Observation surface — never raises.

if await sdk.loop.check("my-agent", session_id):
    # Loop detected — take corrective action
    await escalate_to_human(session_id)

Mark an agent as live. Call periodically (e.g. every 30s) to maintain the "Live" status. Each heartbeat updates the last_seen timestamp in Postgres. Pass operator_id to bind the agent to a human operator — used by the console for self-approval prevention on HITL gates. Pass metadata for arbitrary JSON (max 64KB).

Note: Observation surface — never raises.

# In your agent's main loop:
await sdk.presence.heartbeat(
    "billing-agent",
    operator_id="alice@company.com",
    metadata={"version": "1.2.0"},
)

Explicitly transition an agent to "Idle" status. Use when the agent is waiting for work but still running.

Note: Observation surface — never raises.

# Agent finished processing, waiting for next task:
await sdk.presence.mark_idle("billing-agent")

Remove an agent from the presence table. Call during graceful shutdown. The agent will no longer appear in list_agents().

Note: Observation surface — never raises.

# Graceful shutdown:
await sdk.presence.close_agent("billing-agent")

Return all agents with their current status (Live, Idle, or Unresponsive) and last_seen timestamp.

Returns: list[AgentPresence] with .agent_id, .status, .last_seen

Note: Observation surface — never raises.

agents = await sdk.presence.list_agents()
for agent in agents:
    print(f"{agent.agent_id}: {agent.status} (last seen {agent.last_seen})")

Mark agents as "Unresponsive" if their last heartbeat is older than timeout_seconds. Call periodically from a health-check endpoint or cron job.

Returns: list[str] — agent IDs that were marked unresponsive

Note: Observation surface — never raises.

# Mark agents unresponsive if no heartbeat in 5 minutes:
stale = await sdk.presence.check_stale(timeout_seconds=300)
if stale:
    alert(f"Unresponsive agents: {stale}")

Return True if an operator has written a halt marker for this agent (via the console halt button or a direct write to the presence row). Reads from a 5-second TTL in-memory cache backed by governance_agent_presence.metadata_json — the hot path is one dict lookup, and the worst-case delay between an operator clicking halt and enforcement is bounded by the TTL. If the governance database is unreachable, the cache holds the last known state and a structlog warning is logged instead of raising. This is OBSERVATION ONLY — it does not raise or block anything. For actual enforcement, call assert_alive() instead (or let scope / cost / gates / wrappers call it for you — all four enforcement paths are wired in v0.6.2).

Returns: bool

Note: Primary name as of v0.6.0. is_killed is preserved as a back-compat alias for v0.6.x and REMOVED in v0.7. Does not raise on DB outage — preserves CLAUDE.md Invariant #1 (host app keeps running when governance DB is unreachable). Observation-only: use assert_alive() for enforcement.

# Observation only — use assert_alive() for actual enforcement.
# This pattern is correct when you want to HANDLE the halt explicitly
# (e.g. flush state, notify upstream) before aborting.
if await sdk.presence.is_halted("billing-agent"):
    await flush_partial_results()
    await notify_upstream("halted by operator")
    raise RuntimeError("billing-agent halted — aborting task")

# For enforcement, prefer assert_alive():
await sdk.presence.assert_alive("billing-agent")  # raises AgentHaltedError on halt

Raise AgentHaltedError if the agent has been halted, with the operator, timestamp, and reason from the halt marker. As of v0.6.2 this is the fail-closed gate called from sdk.scope.check(), sdk.cost.check_or_raise(), sdk.gates.request(), wrap_openai, and wrap_anthropic. Can also be called directly if you want to gate your own enforcement path against the halt switch.

Returns: None (raises AgentHaltedError on halt)

Note: Primary surface. Wired automatically into scope + cost + gates + wrappers when enable_presence=True (the default). The AgentKilledError name is still importable as a back-compat alias through v0.6.x; update to AgentHaltedError before v0.7.

from codeatelier_governance.presence import AgentHaltedError

try:
    await sdk.presence.assert_alive("billing-agent")
except AgentHaltedError as e:
    logger.error(
        "agent halted",
        agent_id=e.agent_id,
        halted_by=e.halted_by,
        halted_at=e.halted_at,
        reason=e.reason,
    )
    raise

Bypass the 5-second TTL and refresh the halted-agent cache immediately from the database. Used by tests and by push-based invalidation paths.

Returns: None

Note: Primary name as of v0.6.0. force_refresh_killed_cache is preserved as a back-compat alias through v0.6.x and REMOVED in v0.7. You should rarely need this in application code — the TTL cache is designed to be transparent. Primary use is tests.

# In a test after simulating a console halt:
await sdk.presence.force_refresh_halted_cache()
assert await sdk.presence.is_halted("test-agent")

Register a behavioral contract for an (agent_id, tool) pair. Call at app startup. One contract per (agent_id, tool) pair - registering again overwrites the previous contract.

from codeatelier_governance.contracts.models import Contract, PreCondition, PostCondition

sdk.contracts.register(Contract(
    agent_id="billing-agent",
    tool="charge_customer",
    pre=[
        PreCondition(check="hitl_approved", message="Charges require human approval"),
        PreCondition(check="budget_available", message="Budget must not be exceeded"),
    ],
    post=[
        PostCondition(check="audit_logged", message="Charge must be audit-logged"),
    ],
))

Async context manager that runs check_pre before the body and check_post after. Unifies scope, cost, HITL, and custom checks into one declarative surface. If any pre-condition fails, raises ContractViolation before your code runs. If any post-condition fails, raises ContractViolation after.

Raises: ContractViolation (auto audit-logged as contract.pre_violation or contract.post_violation)

Note: Enforcement surface - raises by contract. No-op if no contract registered for this (agent_id, tool).

async with sdk.contracts.enforce("billing-agent", session_id, "charge_customer"):
    # Pre-conditions verified - safe to proceed
    result = await stripe.charge(amount)
    # Post-conditions checked on exit

Evaluate all pre-conditions for this (agent_id, tool) pair. If any pre-condition fails, emits a contract.pre_violation audit event and raises. No-op if no contract is registered.

Raises: ContractViolation

Note: Enforcement surface - raises by contract.

await sdk.contracts.check_pre("billing-agent", session_id, "charge_customer")
# All pre-conditions passed - safe to proceed

Evaluate all post-conditions for this (agent_id, tool) pair. If any post-condition fails, emits a contract.post_violation audit event and raises. No-op if no contract is registered.

Raises: ContractViolation

Note: Enforcement surface - raises by contract.

await sdk.contracts.check_post("billing-agent", session_id, "charge_customer")
# All post-conditions verified

Constructor for the compliance report generator. Provide database_url for Postgres queries or audit_store for in-memory/test usage. Pass audit_module=sdk.audit (v0.5.3+) to enable chain verification inside reports — required when calling generate_article12(verify_chain=True) or generate_summary(verify_chain=True).

Note: audit_module parameter added in v0.5.3.

from codeatelier_governance.compliance.report import ReportGenerator

# For CLI / archival reports
generator = ReportGenerator(database_url=os.environ["DATABASE_URL"])

# For chain-verified reports — wire the audit module
generator = ReportGenerator(
    database_url=os.environ["DATABASE_URL"],
    audit_module=sdk.audit,
)

Generate an EU AI Act Article 12 evidence report. Queries audit events with the given filters and maps them to the seven Article 12 automatic logging requirements: event registration, duration of use, reference database, input data, functioning, human oversight, and post-market monitoring. The report provides evidence for actions the SDK observed; it does not assert compliance for the overall deployment.

Returns: ComplianceReport with seven ReportSection entries, each with a compliant/partial/non_compliant status, plus coverage_caveat, chain_integrity_status, and coverage_pct fields

Raises: ValueError if verify_chain=True is requested but audit_module was not passed to the ReportGenerator constructor — no silent degradation. IMPORTANT: you must construct ReportGenerator with audit_module=sdk.audit before calling this with verify_chain=True. See the ReportGenerator constructor entry above.

Note: verify_chain keyword-only parameter added in v0.5.3. Available via both the CLI (governance compliance report) and direct import.

from codeatelier_governance.compliance.report import ReportGenerator
from datetime import datetime, timezone

# Chain-verified Article 12 attestation
generator = ReportGenerator(
    database_url=os.environ["DATABASE_URL"],
    audit_module=sdk.audit,
)

report = await generator.generate_article12(
    agent_id="billing-agent",
    date_from=datetime(2026, 1, 1, tzinfo=timezone.utc),
    date_to=datetime(2026, 4, 1, tzinfo=timezone.utc),
    verify_chain=True,  # runs HMAC chain verification
)
assert report.chain_integrity_status == "verified"
print(report.coverage_caveat)
for section in report.sections:
    print(f"{section.title}: {section.status}")

Generate a high-level summary compliance report for an agent. Includes three sections: agent overview (event counts, models, sessions), violation summary (scope/budget/loop violations), and human oversight (HITL gate activity).

Returns: ComplianceReport with three ReportSection entries

Raises: ValueError if verify_chain=True is requested but audit_module was not passed to the ReportGenerator constructor

Note: verify_chain keyword-only parameter added in v0.5.3.

report = await generator.generate_summary(
    agent_id="billing-agent",
    date_from=datetime(2026, 3, 1, tzinfo=timezone.utc),
    verify_chain=True,
)
for section in report.sections:
    print(f"{section.title}: {section.status}")

Patch an Anthropic client to emit governance audit events and track cost automatically. Supports both anthropic.Anthropic (sync) and anthropic.AsyncAnthropic (async). The client is monkey-patched in-place and returned so existing references continue to work. Each messages.create() call will: (1) run cost.check_or_raise before the LLM call, (2) audit-log the call, (3) extract token usage from the response, (4) compute USD cost via the built-in pricing table, and (5) track cost post-call.

Returns: The same client object, patched in-place

Raises: BudgetExceeded (enforcement surface - checked before every LLM call)

Note: Audit logging and cost tracking are observation surfaces and never raise. The budget pre-check is the only enforcement surface. Calling wrap_anthropic twice on the same client is a no-op (logs a warning).

import anthropic
from codeatelier_governance import GovernanceSDK
from codeatelier_governance.integrations.anthropic_wrap import wrap_anthropic

async with GovernanceSDK(database_url=os.environ["DATABASE_URL"]) as sdk:
    client = wrap_anthropic(anthropic.Anthropic(), sdk=sdk, agent_id="my-agent")

    # client.messages.create() now auto-audits + tracks cost
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": "Hello"}],
    )

Open an approval request. Returns a pending request with a signed token. Surface the request to a human (Slack, email, console GUI). The human calls grant(token) or deny(token).

Returns: ApprovalRequest with .request_id, .token, .expires_at

Note: Observation surface — never raises on internal failure.

req = await sdk.gates.request(
    kind="deploy.production",
    agent_id="deploy-agent",
    payload={"pr": 142, "environment": "production"},
)
print(f"Approval needed: {req.request_id}")

Resolve an approval request. Single-use: the same token cannot be used twice. Time-bound: expired tokens are rejected. Action-hash-bound: swapping the action invalidates the token.

Raises: ApprovalTokenError on forge, replay, expiry, or mismatch

await sdk.gates.grant(req.token)  # approval.granted audit row written

Block until the request is resolved. Polls the store every 500ms (with jitter). Multi-process correct: the grant can come from any worker.

Raises: ApprovalTimeout if nobody responds, ApprovalDenied if denied

Note: Enforcement surface — raises by contract.

granted = await sdk.gates.wait_for(req.request_id, timeout=600)
# If this line runs, a human approved the action