clavenar-agent-sdk (Python)

Wrap your Anthropic / OpenAI Python client — async or sync. Every tool call the model emits is inspected by Clavenar before the agent loop can run it.

Sequence diagrams for the five primary paths — clavenar_wrap boot with sync/async fork, async non-streaming inspection, async streaming choice-end gating, ClavenarPending.resolve poll loop, and the standalone OpenAI Realtime helper — plus a request decision-tree flowchart, live in docs/SEQUENCES.md.

import asyncio
from anthropic import AsyncAnthropic

from clavenar_agent_sdk import clavenar_wrap, ClavenarDenied, ClavenarOptions

async def main() -> None:
    client = clavenar_wrap(
        AsyncAnthropic(),
        ClavenarOptions(endpoint="http://localhost:8080", mode="enforce"),
    )

    try:
        result = await client.messages.create(
            model="claude-opus-4-7",
            max_tokens=1024,
            tools=[...],
            messages=[{"role": "user", "content": "list my files"}],
        )
    except ClavenarDenied as e:
        print(f"clavenar denied {e.tool_name}: {e.reasons}")

asyncio.run(main())

OpenAI works the same way:

from openai import AsyncOpenAI
from clavenar_agent_sdk import clavenar_wrap, ClavenarOptions

client = clavenar_wrap(
    AsyncOpenAI(),
    ClavenarOptions(endpoint="http://localhost:8080"),
)

completion = await client.chat.completions.create(
    model="gpt-5",
    tools=[...],
    messages=[...],
)

Sync clients

anthropic.Anthropic and openai.OpenAI (non-async) are wrapped exactly the same way — clavenar_wrap detects sync vs. async by inspecting the underlying create method:

from anthropic import Anthropic
from clavenar_agent_sdk import clavenar_wrap, ClavenarOptions

client = clavenar_wrap(
    Anthropic(),
    ClavenarOptions(endpoint="http://localhost:8080"),
)

result = client.messages.create(model="claude-opus-4-7", ...)

Sync clients use httpx.Client under the hood. Both sync and async wrappers select the side-effect-free clavenar.decision/v1 contract with a UUID allocated before the first attempt; a multi-tool turn is one ordered atomic decision. Proxy 1.0.0 and Lite 1.0.0 reject unselected tool calls with HTTP 426; upgrade this SDK before the gateway by following https://clavenar.com/docs/sdk-migration/. Callbacks (on_verdict, on_policy_error) must be sync when wrapping a sync client.

Streaming

stream=True is intercepted: each event/chunk passes through in order, but the closing event (Anthropic content_block_stop, OpenAI finish_reason="tool_calls") is held until clavenar returns a verdict. A denied tool raises mid-iteration before partner code can act on it.

async with client.messages.create(stream=True, ...) as stream:
    async for event in stream:
        ...
# ClavenarDenied raised inside the async-for if a tool_use was blocked.

Both AsyncAnthropic + AsyncOpenAI streams and their sync counterparts (Anthropic, OpenAI) are supported.

The provider SDKs' .stream() convenience helpers (messages.stream(), chat.completions.stream()) are blocked by the wrapper: their rich event interfaces can't be wrapped faithfully, so tool calls made through them would bypass inspection entirely. Calling one raises ClavenarConfigError pointing at create(stream=True); set allow_uninspected_stream=True only if you explicitly accept uninspected streaming.

Pending → resolve

When clavenar parks a tool call for human review, ClavenarPending is raised. Catch it and await resolve() to block until an operator decides:

try:
    result = await client.messages.create(...)
except ClavenarPending as p:
    print(f"awaiting approval: {p.review_reasons}")
    await p.resolve(poll_interval_s=2.0, timeout_s=600.0)
    # Returns on approve; raises ClavenarDenied on deny.

Transient transport errors (5xx, network blips) are swallowed between polls. Malformed success bodies, correlation mismatches, and every other status re-raise immediately. Sync wrappers use p.resolve_sync(poll_interval_s=2.0, timeout_s=600.0).

Governed execution

Use execute_prepared_tool or execute_prepared_tool_sync when policy authorization and the provider effect must form a recoverable workflow. Their options require an application-owned durable store, a cryptographic authorization verifier, a receipt signer, and an executor that forwards the supplied idempotency ID to the provider. Authorization bindings are checked before an intent is committed or an effect is released.

On restart, a stored completion is integrity-checked and returned. A stored intent is passed to recover_effect; if recovery is absent or inconclusive, the SDK raises ClavenarRecoveryRequired instead of executing again. Completion plus receipt-outbox persistence is bounded by the finalization deadline.

SecureTransportProfile reuses its sync and async connection pools. Call reload() / reload_sync() after rotating credential files and aclose() / close() during application shutdown.

Debugging a denied tool call

ClavenarDenied carries reasons, layer, and correlation_id. To see which detector fired, run the gateway with CLAVENAR_PROXY_VERBOSE_VERDICTS=true (Lite: --verbose-verdicts) — the deny then carries a per-detector detail breakdown, exposed as err.detail and rendered to stderr when you set dev_mode=True:

client = clavenar_wrap(
    anthropic,
    ClavenarOptions(
        endpoint="https://clavenar.internal",
        dev_mode=True,  # dev/staging only — detailed denials are an attacker oracle
    ),
)
# On a deny, the SDK prints a panel to stderr:
#   ━━ clavenar denied: send_email ━━
#     layer=brain  intent=Exfiltration  correlation=abc-123
#     detectors:
#       persona_drift         0.12
#       injection             0.91  ⚠ flagged
#     degraded: injection

Programmatic access (no dev_mode needed):

try:
    await client.messages.create(...)
except ClavenarDenied as e:
    if e.detail:
        fired = [d for d in e.detail["detectors"] if d.get("flagged") or d["score"] >= 0.5]
        print("fired detectors:", fired)

detail is None unless the gateway opts in; without it the panel prints a hint to enable verbose verdicts. render_deny_panel(err) returns the string if you want it without writing to stderr.

Retries

Network errors and 5xx responses retry up to max_attempts with jittered exponential backoff. 200, 403, and other 4xx never retry. Defaults mirror the TS SDK at 1.1.0:

from clavenar_agent_sdk import ClavenarOptions, ClavenarRetryOptions

opts = ClavenarOptions(
    endpoint="...",
    retry=ClavenarRetryOptions(max_attempts=3, base_delay_s=0.1),
)

Set max_attempts=1 to disable retries.

Modes

Observe is the rollout knob — surface what clavenar would decide without breaking the agent. Flip to enforce per-call once verdicts are trusted.

Install

pip install clavenar-agent-sdk==1.5.3

Python 3.10+. Runtime dep is httpx only; the anthropic and openai packages are NOT imported by clavenar-agent-sdk — bring your own.

Configuration

Credential-bearing endpoints require HTTPS by default. Reserved protocol headers cannot be overridden through extra_headers; request and response sizes, batches, retry values, and identifiers are bounded before network work.

Wire contract

The HTTP shape this SDK speaks against the inspect endpoint (POST /mcp, the verdict envelope, the pending / resolve contract, and the X-Clavenar-* header set) is documented in the workspace's source of truth: clavenar-specs/TECH_SPEC.md. This SDK is a faithful client of that contract — if you observe a divergence, file the bug against the spec first.

The TypeScript sibling at clavenar-typescript-sdk implements the same wire contract with parity guarantees.

License

Apache-2.0.