agentlattice 0.3.0

Governance SDK for AI agents — audit trails, policy engine, approval gates.

agentlattice — Python SDK

Governance-aware SDK for AI agents. Wraps agent actions with audit trails, approval gates, and circuit breaker integration.

Install

pip install agentlattice

Quickstart

import asyncio
import os
from agentlattice import AgentLattice

al = AgentLattice(api_key=os.environ["AL_API_KEY"])

async def main():
    # gate() blocks until approved — your agent code only runs if it passes
    await al.gate("pr.merge")
    await merge_pr(42)

asyncio.run(main())

Handle denials and timeouts:

from agentlattice import AgentLattice, AgentLatticeDeniedError, AgentLatticeTimeoutError

al = AgentLattice(api_key=os.environ["AL_API_KEY"])

async def main():
    try:
        await al.gate("code.commit")
    except AgentLatticeDeniedError as e:
        print(f"Denied: {e.reason}, policy: {e.policy}")
        print(f"Failed conditions: {e.conditions}")
    except AgentLatticeTimeoutError as e:
        print(f"Approval timed out: {e.approval_id}")

asyncio.run(main())

Sync usage

If you're not in an async context:

import os
from agentlattice import AgentLattice

al = AgentLattice(api_key=os.environ["AL_API_KEY"])
al.gate_sync("code.commit")

API

AgentLattice(api_key, *, base_url?, gate_poll_timeout_ms?, gate_poll_interval_ms?)

Parameter	Default	Description
api_key	required	Bearer token — pass via os.environ["AL_API_KEY"], never hardcoded
base_url	https://agentlattice.com	Override for self-hosted or staging
gate_poll_timeout_ms	8 * 60 * 60 * 1000 (8 hours)	How long gate() polls for approval
gate_poll_interval_ms	5000	Poll interval in ms

await al.execute(action_type, options?)

Submit an action and return immediately. Returns an ActionResult. Does not block.

from agentlattice import AgentLattice, ActionOptions

al = AgentLattice(api_key="al_...")
result = await al.execute("pr.open", ActionOptions(
    data_accessed=[{"type": "source_code", "count": 5, "sensitivity": "low"}],
    metadata={"pr_number": 42},
    event_id="my-idempotency-key",   # optional: dedup by your own key
))

if result.status == "executed":
    print(f"Auto-executed under policy: {result.policy_name}")
elif result.status == "requested":
    print(f"Awaiting approval: {result.approval_id}")
elif result.status == "denied":
    print(f"Denied ({result.denial_reason}): {result.conditions_evaluated}")

await al.gate(action_type, options?)

Submit an action and block until it is approved or denied.

try:
    await al.gate("pr.merge")
except AgentLatticeDeniedError as e:
    # e.reason:     "CONDITIONS_DENIED" | "POLICY_TAMPERED" | "DENIED_BY_REVIEWER"
    # e.policy:     name of the governing policy (if conditions-based)
    # e.conditions: list of ConditionResult (which rule failed and why)
    # e.approval_id: set if a human reviewer denied it
    ...
except AgentLatticeTimeoutError as e:
    # e.approval_id: the approval that timed out
    ...

al.gate_sync(action_type, options?) / al.execute_sync(action_type, options?)

Synchronous wrappers — identical semantics, no asyncio.run() needed.

ActionResult

@dataclass
class ActionResult:
    status: str                          # "executed" | "requested" | "denied" | ...
    audit_event_id: str
    approval_id: str | None              # set when status="requested"
    message: str | None
    timeout_at: str | None
    denial_reason: str | None            # "CONDITIONS_DENIED" | "POLICY_TAMPERED" | ...
    policy_name: str | None              # governing policy name
    conditions_evaluated: list[ConditionResult]  # which rules passed/failed

ConditionResult

@dataclass
class ConditionResult:
    field: str       # e.g. "pr_size"
    operator: str    # e.g. "lt"
    expected: Any    # e.g. 100
    result: bool     # True = passed, False = failed

Agents can read conditions_evaluated to understand exactly which governance rule blocked them and adapt their behavior accordingly.

Self-correcting agent pattern

conditions_evaluated is what makes this more than a compliance layer — it closes the feedback loop so agents can adapt, not just fail.

from agentlattice import AgentLattice, AgentLatticeDeniedError, ActionOptions

al = AgentLattice(api_key=os.environ["AL_API_KEY"])

async def commit_with_governance(files: list[str]) -> None:
    try:
        await al.gate("pr.open", ActionOptions(
            metadata={"pr_size": len(files), "repo": "acme/backend"},
        ))
        await open_pr(files)

    except AgentLatticeDeniedError as e:
        if e.reason == "CONDITIONS_DENIED":
            # Find which rules failed and why
            failed = [c for c in e.conditions if not c.result]
            # e.g. [ConditionResult(field="pr_size", operator="lt", expected=50, result=False)]

            for rule in failed:
                if rule.field == "pr_size":
                    # Policy says PRs must be < 50 files — split and retry
                    mid = len(files) // 2
                    await commit_with_governance(files[:mid])
                    await commit_with_governance(files[mid:])
                    return

        # Reviewer denial or policy tamper — no retry
        raise

Error hierarchy

AgentLatticeError          base — code, message, details
  AgentLatticeDeniedError  approval_id?, reason?, policy?, conditions?
  AgentLatticeTimeoutError approval_id

Reliability

execute() and gate() retry automatically on 5xx and network errors — up to 3 attempts with 1s/2s/4s exponential backoff. Each request has a 30-second timeout. Pass event_id to make retries idempotent.

execute_sync() and gate_sync() are safe in async runtimes (LangGraph, FastAPI, CrewAI) — they detect a running event loop and dispatch to a thread pool automatically.

License

MIT