bulwark-sdk (Python)

Governance, audit and policy decisions for AI agents.

Zero runtime dependencies — stdlib only. Python 3.9+.

Install

Not yet on PyPI. Install from git:

pip install bulwark-sdk

If that fails with "No matching distribution found", the package is not on PyPI yet. Two alternatives, both fully supported:

• Use the REST API directly. Three calls, no install. See docs/API.md §2.
• Vendor it. Copy src/bulwark/ into your project. Zero dependencies, nothing to install.

The one thing to understand first

There are two calls, and they do different jobs:

If you only call observe(), your dashboard shows activity but zero decisions, zero flags, and a risk score driven only by registry facts. That is working as designed. Call decide() at the point your agent is about to do something that matters.

Bulwark never blocks your agent. decide() returns a verdict; acting on it is your code's job.

Quickstart

from bulwark import BulwarkClient

bw = BulwarkClient(api_key="bw_live_...")  # Settings → API keys

# 1. Register once (at deploy time, not per request)
agent = bw.register_agent(
    name="Portfolio Assistant",
    purpose="Answers investor questions about portfolio performance",
    risk_tier="medium",
    environment="prod",
    tools=["portfolio.read", "ai.assistant.answer"],
)
bw.activate_agent(agent["agentId"])   # agents are created "pending"

# 2. Record what the agent did
bw.observe(agent["agentId"], "ai.assistant.answer", resource="portfolio")

# 3. Ask for a verdict before something consequential
verdict = bw.decide(agent["agentId"], "crm.write", resource="customers")
if verdict.pending:
    log.warning("Awaiting human approval: %s", verdict.approval_id)
elif not verdict.allowed:
    log.warning("Bulwark would block this: %s", verdict.reason)

Three things that surprise people

1. Agents start pending, not active. decide() denies anything from a non-active agent with agent_not_active. Call activate_agent() after registering.

2. tools is currently descriptive, not enforced. Declaring tools=["crm.read"] records the agent's intended allowlist and drives the access-matrix view, but Bulwark does not yet reject or escalate an action outside it. Today this returns allow:

bw.register_agent(..., tools=["crm.read"])
bw.decide(agent_id, "crm.write")   # -> allow, "no matching policy — default allow"

To enforce scope today, author an explicit policy (POST /v1/policies) with a rule matching the actions you want escalated. Declare tools anyway — it is the allowlist scope enforcement will read once it ships.

3. Events for an unregistered agent are dead-lettered, not rejected. A typo'd agent_id fails quietly — the event lands in /v1/events/dead-letter rather than erroring. The SDK emits a warning when the API reports dead-lettered events.

Serverless

The background flusher does not run once your process is frozen or torn down. Flush before you return:

def handler(event, context):
    bw.observe(agent_id, "doc.process", resource=event["key"])
    ...
    bw.flush()      # or: with BulwarkClient(...) as bw:
    return {"ok": True}

flush() guarantees that everything buffered when you called it has been sent before it returns.

Failure behaviour

fail_open=True (the default) is the monitor-mode contract: governance being down must never take your agent down.

• observe() never raises. Failed batches are dropped and counted in bw.dropped_events.
• decide() returns an allow marked degraded=True. Never treat a degraded verdict as a real governance record — it means no policy was evaluated.

Set fail_open=False to surface errors as exceptions instead.

Don't send content

Never pass prompts, completions, or document bodies in context. Bulwark strips content fields at ingest by design, but the cheapest data to protect is the data you never transmit. Send metadata: identifiers, action names, resource names, counts.

Configuration

BulwarkClient(
    api_key,
    base_url="https://bulwark-api.fly.dev",
    timeout=5.0,
    batch_size=100,       # API hard cap is 100 events per batch
    flush_interval=2.0,   # background flush cadence, seconds
    fail_open=True,
    max_queue=10_000,     # buffer bound; oldest dropped beyond it
    max_retries=3,        # 429 and 5xx only, honours retry-after
)

API

Exceptions

All inherit BulwarkError. BulwarkAuthError (401), BulwarkPermissionError (403), BulwarkUpgradeRequired (402, has .feature), BulwarkRateLimited (429, has .retry_after), BulwarkPayloadTooLarge (413), BulwarkValidationError (400, .body["details"]), BulwarkTransportError (network).

An API key maps to the developer role: it can register agents, send events and call decide(), but cannot mint keys, manage team members, or change the plan — those need an admin session.

Tests

pip install -e ".[dev]"
python -m pytest