wytness

Wytness wrapper for Microsoft's Agent Governance Toolkit (AGT). Adds the two security properties AGT does not ship:

• Non-repudiation — per-event Ed25519 signature with a customer-held private key. Asymmetric verification: anyone with the public key can verify, no shared secret.
• Zero-knowledge PII — HMAC-SHA256 pseudonyms + X25519 + ChaCha20-Poly1305 token-map encryption. Wytness never sees raw PII.

These compose with AGT's own primitives (Ed25519 agent identity, SHA-256 Merkle hash chain). See docs/agt-wrapper-design.md §2.5 for the full layered model.

Mirrors wytness_ai (TypeScript) — same wire format, same init() / shutdown() lifecycle, same env-var contract.

Status

Current version: 1.8.0. Depends on agent-governance-toolkit[full]==4.1.0 (hard dependency, installed automatically; the latest published Python artifact of the AGT 5.0 train). Mirrors @wytness/ai 1.8.0 (TypeScript) — same wire format, same env-var contract.

Public surface: init(), shutdown(), WytnessConfig, WytnessAuditSink, WytnessEventSink, stats(), health(), wire_status(), AuditLog (re-exported from AGT), and the typed exception hierarchy (WytnessSDKError → WytnessConfigError / WytnessInitError / WytnessTransportError). Wire envelope is envelope_version: 2; signature covers canonical-JSON of {data, id, time, type}.

Tests: a pytest suite covering config validation (XOR PII pairing, signing-key prefix/length tolerance, env-var fallback), envelope signing + tamper-detection, PII pseudonymisation + per-sink token-map encryption + auto-attach, transport batching + retry + dead-letter file + jittered backoff, init/shutdown idempotency, stats/health pre-init safety, framework-adapter compatibility against AGT 4.1.0's 12 integrations (LangChain, CrewAI, LangGraph, MCP, Haystack, Flowise, Langflow, Swarm, AI Card, A2A, Django middleware, HTTP middleware), and event-processor resolution regression tests covering the agent_os.event_sink candidate path.

Install

pip install wytness-ai

agent-governance-toolkit[full]==4.1.0 is a hard dependency — installed automatically. No extras to remember. Python >=3.11.

Hello world

from wytness_ai import init, shutdown, AuditLog

init(
    api_key="wyt_...",
    signing_key="...",     # browser-generated Ed25519 (always required)
    pii_pubkey="...",      # browser-generated X25519 (optional, paired with pii_secret)
    pii_secret="...",      # browser-generated HMAC secret (optional, paired with pii_pubkey)
    pii_fields=["customer.email"],
)

# AGT owns identity + Merkle chain. The wrapper has already monkey-patched
# AuditLog — every entry below also flows to Wytness.
audit = AuditLog()
audit.log(
    event_type="tool_invocation",
    agent_did="my-first-agent",
    action="find_espresso",
    data={
        "customer": {"email": "alice@example.com"},
        "query": "best espresso",
    },
)

shutdown()

AuditLog is re-exported from wytness_ai for convenience, so a single import line covers both lifecycle and audit. The original from agentmesh.governance.audit import AuditLog is identical and still works.

PII is pseudonymised before AGT computes its chain hash (so AGT's hashes commit to the redacted bytes — chain verification still works), the canonical-JSON envelope is signed with your Ed25519 key, and the batch lands at `api.wytness.ai/ingest`. Failed POSTs land in `~/.wytness/wytness-deadletter.jsonl` — recover them with `wytness-ai replay-deadletter` (add `--dry-run` to list first): it re-POSTs the already-signed envelopes through the normal transport and is safe to re-run, because the backend deduplicates replayed event ids.

`init()` reads `WYTNESS_API_KEY` / `WYTNESS_SIGNING_KEY` / `WYTNESS_PII_PUBKEY` / `WYTNESS_PII_SECRET` / `WYTNESS_ENDPOINT` from the environment when kwargs aren't supplied. Browser-generated keys come from the [Wytness dashboard's Keys page](https://app.wytness.ai/keys) (Signing Key, PII Encryption Key, and PII HMAC Key cards); the dashboard never sees your private halves.

A complete runnable example lives at [`examples/hello-world/`](examples/hello-world/) — it generates ephemeral demo keys if no `WYTNESS_*` env vars are set, so you can `python examples/hello-world/main.py` immediately after install.

### Manual wiring (no monkey-patch)

If you'd rather wire explicitly, pass `auto_wire=False` and use the audit sink directly. The sink accepts AGT `AuditEntry` instances + dicts shaped like one.

```python
from wytness_ai import init, get_transport
from wytness_ai.sinks import WytnessAuditSink

config = init(auto_wire=False, signing_key="...", api_key="...")
transport = get_transport()
sink = WytnessAuditSink(transport=transport, config=config)
sink.write({
    "event_type": "tool_invocation",
    "agent_id": "agent-1",
    "data": {"query": "…"},
})

Automatic LLM capture

Point init() at your model client and every Anthropic Messages API (including the beta namespace) and OpenAI Chat Completions and Responses call — streaming included — is recorded automatically — prompt, reply, reasoning (Anthropic extended thinking), latency, and token usage — with no AuditLog().log(...) in your code. Token usage needs no opt-in: on OpenAI streams the SDK requests it and hides the resulting usage-only chunk, so your loop is unchanged. Other providers are not captured. Captured calls ride the same signed, secret-scrubbed sink as manual audit entries; when PII pseudonymisation is enabled the conversation is tokenised in-process before it leaves the machine.

import anthropic
import wytness_ai

wytness_ai.init(
    wrap=anthropic,                                 # capture Anthropic calls
    agent_did="did:web:acme.example:support-bot",   # attributes captured events
)

client = anthropic.Anthropic()
client.messages.create(                             # recorded automatically
    model="claude-haiku-4-5",
    max_tokens=512,
    messages=[{"role": "user", "content": "Summarise ticket #4821"}],
)

wytness_ai.shutdown()

wrap= accepts a provider module (anthropic / openai), its name as a string ("anthropic" / "openai"), a list of either, or True / "auto" to auto-detect the providers you have installed:

wytness_ai.init(wrap=[anthropic, openai], agent_did="did:web:acme.example:bot")
wytness_ai.init(wrap=True, agent_did="did:web:acme.example:bot")   # auto-detect

Anthropic and OpenAI are supported today. agent_did attributes the captured calls to an agent — set it, or captured events land unattributed. Streaming calls are captured too: create(stream=True) returns a transparent tee that records the assembled response when the stream is exhausted, errors, is close()d, or its with block exits (a bare break without any of those skips capture), and client.messages.stream(...) captures from the SDK's own message snapshot on exit — helpers like .text_stream and .get_final_message() are untouched. Streamed captures carry parameters.stream: true. OpenAI streamed usage needs no opt-in — the SDK sets stream_options when you have not, and never overrides one you set yourself.

Parity note. Python auto-patches the provider at init(wrap=…); the TypeScript SDK uses an explicit wrapAnthropic(client) / wrapOpenAI(client) because its provider-SDK classes aren't safely patchable at init. The recorded event is identical — only the one-line setup differs.

Capture-status reporting. The SDK periodically reports which surfaces it is recording to Wytness so your dashboard can state what is watched. The report carries surface names, capture states, and delivery counters — never prompts, payloads, or credentials — over the same endpoint and API key as your events. Fail-soft (a failed report is retried later, capture is never affected); dry_run disables it.

CLI

python -m wytness_ai <command> (or the wytness-ai console script). All commands read the same WYTNESS_* env vars and never generate keys.

Dry run

Set WYTNESS_DRY_RUN=1 (or init(dry_run=True)) to build and Ed25519-sign every envelope but skip the POST to /ingest — the signed envelope is written to stderr instead. The init-time signing-key registration probe is skipped too, so it runs fully offline. Ideal for CI, unit tests, and local development without polluting production.

Framework recipes

Where to call init() / shutdown() in FastAPI, Django, and AWS Lambda: docs/recipes.md.

Feature parity with wytness_ai (TypeScript)

Both SDKs ship the same wire format (snake_case canonical shape, identical envelope structure, byte-comparable for the same logical input). Behavioural surface differences:

Wire format

Each POST batch is a JSON array of CloudEvents envelopes:

[
  {
    "specversion": "1.0",
    "id": "<uuid>",
    "source": "wytness",
    "type": "AuditEntry",
    "datacontenttype": "application/json",
    "time": "2026-05-25T00:00:00.000Z",
    "data": {
      "timestamp": "2026-05-25T00:00:00.000Z",
      "agent_id": "my-first-agent",
      "event_type": "tool_invocation",
      "entry_hash": "<AGT-supplied SHA-256 hex>",
      "previous_hash": "<prior entry's hash>",
      "data": { "query": "…", "customer": { "email": "EMAIL_<token>" } }
    },
    "wytness_envelope": {
      "envelope_version": 2,
      "key_id": "<16 chars base64url SHA-256(pubkey)>",
      "signature_ed25519": "<base64 Ed25519 sig over canonicalJson({data, id, time, type})>",
      "pseudonymization_version": "1"
    }
  }
]

Content-Type: application/vnd.wytness.agt+json — every envelope is signed. Wytness has no unsigned mode; signing is the product's value prop.

Idempotency / deduplication

The SDK gives you network-retry idempotency for free: when a POST fails and is retried, the same envelope id is re-sent, so the backend deduplicates the duplicate write.

The SDK does not automatically deduplicate customer-side retries — a logical action retried twice in your app code produces two distinct envelopes with two distinct entry_ids, which means two CH rows. If you need that guarantee, thread a stable per-action key through your event payload:

audit.log(
    event_type="tool_invocation",
    agent_did="my-first-agent",
    action="find_espresso",
    data={
        "entry_id": "tool-invocation-2026-06-01-uuid-abc",
        "customer": {"email": "alice@example.com"},
        "query": "best espresso",
    },
)

When data["entry_id"] is present at the boundary the SDK preserves it through canonicalisation; the backend then uses (org_id, entry_id) as the dedup tuple. Absent the field, the SDK synthesises a fresh UUID per write() call.

License

MIT. Built on agent-governance-toolkit (MIT, Microsoft).