Metadata-Version: 2.4
Name: bulwark-sdk
Version: 0.2.0
Summary: Bulwark SDK: governance, audit and policy decisions for AI agents
Author: Bastion Shield Technologies Ltd
License-Expression: MIT
Project-URL: Homepage, https://bastionshieldtechnologies.com
Project-URL: Changelog, https://github.com/bastionshieldtechnologies/bulwark/blob/main/packages/sdk-python/CHANGELOG.md
Project-URL: Source, https://github.com/bastionshieldtechnologies/bulwark
Keywords: ai,agents,governance,audit,compliance,mcp,anthropic,bedrock,openai,langchain
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Intended Audience :: Developers
Classifier: Topic :: Security
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Provides-Extra: offline
Requires-Dist: cryptography>=42; extra == "offline"
Provides-Extra: dev
Requires-Dist: pytest>=7.4; extra == "dev"
Requires-Dist: cryptography>=42; extra == "dev"
Dynamic: license-file

# bulwark-sdk (Python)

Governance, audit and policy decisions for AI agents.

**Zero runtime dependencies** — stdlib only. Python 3.10+ (tested on 3.10, 3.11 and 3.12).

## Install

```bash
pip install bulwark-sdk
```

Python 3.10+, zero runtime dependencies. `pip install "bulwark-sdk[offline]"` adds
`cryptography` for the signed offline bundle.

Prefer not to add a dependency at all? Two supported alternatives:

- **Use the REST API directly.** Three calls, no install. See `docs/API.md` §2.
- **Vendor it.** Copy `src/bulwark/` into your project. Standard library only.

## The one thing to understand first

There are two calls, and they do different jobs:

| Call | Endpoint | What happens | Counts as a decision? |
|---|---|---|---|
| `observe()` | `POST /v1/events` | Appended to the audit chain. **No policy is evaluated.** Buffered + batched. | No |
| `decide()` | `POST /v1/decide` | Policy is evaluated. Returns allow/deny/escalate, can raise a human approval. | Yes |

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

```python
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 checked on Team and above, recorded only on Free.** On a Team or Business
workspace, an action outside the agent's declared `tools` is escalated as `action_outside_scope`:

```python
bw.register_agent(..., tools=["crm.read"])
bw.decide(agent_id, "crm.write")   # Team+: escalate, "action_outside_scope"; Free: allow
```

An empty `tools` list turns the check off for that agent. Action names glob, so `crm.*` covers
`crm.read` and `crm.write`.

**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:**

```python
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

The same in this SDK and the TypeScript SDK (`bulwark-sdk` on npm), so an integration ported
between languages keeps its security behaviour.

- `observe()` never raises, in any mode. Undeliverable events stay queued; beyond `max_queue` the
  oldest are dropped and counted (`bw.stats()["dropped_events"]`).
- `decide()` has a hard total deadline, `decide_timeout` (default **1.5 s**, retries included).
- When Bulwark cannot answer, `decide()` uses the cached signed bundle (`source == "cached_offline"`),
  and with no usable bundle applies `fail_mode` (`source == "fail_mode"`):

| `fail_mode` | No usable bundle |
|---|---|
| `"fail_closed"` (default) | deny |
| `"fail_open"` | allow, with the `log_local_backfill` obligation |
| `"risk_tiered"` | deny for `high`, `critical` and unknown tiers; allow for `low` and `medium` |

Both kinds of offline verdict are marked `degraded=True` and queued as decision events so they reach
your audit trail when Bulwark is back. Never treat a degraded verdict as a live governance record.

What each response means for `decide()`:

| Status | SDK behaviour | Raises |
|---|---|---|
| 2xx | live verdict | |
| 400 | not retried, never answered offline | `BulwarkValidationError` |
| 401 | not retried, never answered offline (bad, expired or revoked key) | `BulwarkAuthError` |
| 402 | not retried, never answered offline (plan) | `BulwarkUpgradeRequired` |
| 403 | not retried, never answered offline | `BulwarkPermissionError` |
| 429 | waits `retry-after` and retries only if that fits the deadline; otherwise offline | |
| 5xx | offline at once, no retry in the hot path | |
| network error, deadline | offline at once | |

`decide()` has no idempotency key: a call that timed out on our side but reached the server may be
recorded there as well as answered offline. Treat duplicate decision records after an outage as
expected.

`fail_open=True/False` from 0.1 still works but is deprecated and warns once; it maps to
`fail_mode="fail_open"` / `"fail_closed"`. **The default changed from fail-open to fail-closed in
0.2.0** (see CHANGELOG.md).

## 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

```python
BulwarkClient(
    api_key,
    base_url="https://api.bastionshieldtechnologies.com",
    timeout=5.0,
    batch_size=100,       # API hard cap is 100 events per batch
    flush_interval=2.0,   # background flush cadence, seconds
    fail_mode="fail_closed",   # or "fail_open", "risk_tiered"
    decide_timeout=1.5,   # total deadline for one decision, seconds
    agent_risk_tiers=None,  # {agent_id: tier} for risk_tiered
    max_queue=10_000,     # buffer bound; oldest dropped beyond it
    max_retries=3,        # events: 429 and 5xx, honours retry-after (capped at 30 s)
    raise_on_flush_error=False,
)
```

## API

| Method | Purpose |
|---|---|
| `register_agent(name, purpose, risk_tier, environment, *, tools, data_sources, models_used, scope, category, owner_user_id, spend_limit_gbp)` | Register. Returns the agent dict (state `pending`). |
| `activate_agent(agent_id)` | Transition to `active`. |
| `set_agent_state(agent_id, state)` | `draft`→`pending`→`active`→`suspended`/`retired`. |
| `list_agents()` | All agents for your tenant. |
| `observe(agent_id, action, *, resource, on_behalf_of, context, type, usage)` | Buffered telemetry. Never raises. |
| `decide(agent_id, action_type, *, resource, data_classification, context, args, risk_tier)` | Policy verdict. |
| `report_effect(decision_id, *, executed, outcome, detail, cost_gbp, occurred_at, usage)` | What happened after a live verdict (Team+). |
| `gate_openai_tool_calls(agent_id, response)` | Decide every OpenAI tool call. |
| `gate_anthropic_tool_use(agent_id, message)` | Decide every Anthropic `tool_use` block; returns `tool_result` blocks for blocked ones. |
| `gate_bedrock_tool_use(agent_id, response)` | Decide every Bedrock Converse `toolUse` block; returns `toolResult` blocks for blocked ones. |
| `report_mcp_server(name, tools, *, transport, target, agent_id)` | Report an MCP server's `tools/list` to the registry. |
| `sync_bundle()` | Fetch, verify and cache the signed policy bundle. |
| `rotate_key(grace_minutes=60)` | Replace a short-lived key (Business). |
| `stats()` | `dropped_events`, `queued_events`, `bundle_version`, `last_bundle_error`. |
| `flush(timeout=None)` | Send everything buffered. |
| `close()` | Flush and stop the worker. Runs at exit automatically. |

### 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

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

## Rotating keys (Business)

A key minted with a rotation period can replace itself before it expires:

```python
out = bw.rotate_key(grace_minutes=60)  # old secret works 60 more minutes
store_secret("BULWARK_API_KEY", out["secret"])  # your secret manager
```

## Working when Bulwark is unreachable

Offline decisions use a signed policy bundle. Verifying it needs the
`offline` extra (the `cryptography` package; the base install stays
dependency-free):

```bash
pip install "bulwark-sdk[offline]"
curl -s https://api.bastionshieldtechnologies.com/v1/bundle/keys   # note kid and x, once
```

Pin the key and your tenant, then cache the bundle at startup and keep it fresh:

```python
bw = BulwarkClient(
    api_key="bw_live_...",
    tenant_id="ten_...",                 # yours; a bundle for any other tenant is refused
    bundle_keys={"<kid>": "<x>"},        # pin both kids during a key rotation
    bundle_cache_path="/var/lib/myagent/bulwark-bundle.json",  # optional, survives restarts
    bundle_refresh=60,                   # refresh every 60s (sooner if the server asks)
)
bw.sync_bundle()
```

A bundle is used only if its Ed25519 signature verifies against a pinned key,
it names your tenant, it is inside its validity window and it is not older than
the last one accepted (kept in `<cache>.state.json`). A cached file is
re-verified on every load. A rejected bundle is never used; offline decisions
then follow `invalid_bundle_mode` (`"deny"` by default, never `fail_mode`), and
`bw.last_bundle_error` says why. Without pinned keys and a tenant id, or
without the `offline` extra, no bundle is used. `bundle_trust_on_first_use=True`
fetches and pins the key and tenant on first use: weaker, and off by default.

If Bulwark cannot be reached (network error, timeout, rate limit or server
error), `decide()` evaluates from the cached bundle while it is younger than
`max_bundle_age` (24 hours by default): revoked agents, blocked MCP servers,
delegation limits and enforce-mode rules, with time windows. Those verdicts
have `source == "cached_offline"` and `degraded == True`, and are queued so
they reach your audit trail when Bulwark is back. With no fresh bundle,
`fail_mode` decides (`source == "fail_mode"`). Authentication, validation and
plan errors are never answered offline. Risk scoring and approvals need the
live API.

## Argument conditions

`bw.decide(agent_id, "mcp.fs.read_file", args={"path": path})` lets policies
test arguments without Bulwark receiving them: the client sends only derived
attributes (type, size, keyed hashes, URL host, glob/regex results) for the
fields your rules reference. It needs a verified bundle (pinned keys, tenant
id, the `offline` extra); otherwise argument conditions are unknown and fail
closed. See docs/API.md, "Argument conditions".

## Framework helpers

Every helper gates each tool call with its name (plus `action_prefix`) as the action and passes the
tool input as `args`, so argument conditions apply while the raw input stays in your process.
Nothing is executed or dropped for you: run `allowed`, handle `blocked`.

**Anthropic Messages** (`tool_use`):

```python
msg = anthropic.messages.create(model=..., tools=tools, messages=messages)
out = bw.gate_anthropic_tool_use(agent_id, msg, action_prefix="tools.")
results = [run_tool(t) for t in out["allowed"]]          # your tool_result blocks
messages += [{"role": "assistant", "content": msg.content},
             {"role": "user", "content": results + out["tool_results"]}]
```

`out["tool_results"]` holds a `tool_result` block with `is_error: True` for each denied or escalated
call, so the model is told the tool did not run instead of guessing. With streaming, gate the final
message (after `message_stop`), never partial `input_json_delta` chunks.

**AWS Bedrock Converse** (`toolUse`): `bw.gate_bedrock_tool_use(agent_id, response)` has the same
contract; `out["tool_results"]` are `{"toolResult": {..., "status": "error"}}` content blocks.

**OpenAI**: `bw.gate_openai_tool_calls(agent_id, response)` returns `{"allowed": [...], "blocked": [...]}`
for a Chat Completions or Responses API response.

**LangChain**: `from bulwark.langchain import BulwarkCallbackHandler`, then pass
`callbacks=[BulwarkCallbackHandler(bw, agent_id)]`. A denied or escalated tool raises
`BulwarkToolBlocked` before it runs; outcomes are reported as effects. An escalation is a block
until a reviewer approves it: retry the tool after approval (on Business with context binding, pass
the approval as `context={"approvalId": ...}` on the retried decision).

## Token usage

Report what each model call consumed, as metadata (counts and a model id, never content):

```python
from bulwark import usage_from_response

usage = usage_from_response(msg)                       # Anthropic or OpenAI response
usage = usage_from_response(resp, model=model_id)      # Bedrock: pass the modelId you called
bw.report_effect(verdict.decision_id, executed=True, outcome="success",
                 usage={**usage, "cost": 0.0042, "currency": "USD"})
bw.observe(agent_id, "ai.model.call", usage=usage)    # or as an event
```

`cost` is stored in the `currency` you send; Bulwark never treats USD as GBP. `cost_gbp` on
`report_effect` remains the field the spend-ceiling detector reads.

## Licence

MIT. See [LICENSE](LICENSE). Copyright (c) 2026 Bastion Shield Technologies Ltd. Using the SDK with the hosted Bulwark service is also subject to the Bulwark Terms of Service.
