wytness-ai 1.1.0

Wytness — the blackbox recorder for AI agents. Customer-held PII keys, per-event Ed25519 signing, audit-ready evidence packs. Built on Microsoft's Agent Governance Toolkit.

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.0.2. Depends on agent-governance-toolkit[full]==4.0.0 (hard dependency, installed automatically). Mirrors @wytness/ai 1.0.2 (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: 106 pytest green 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.0.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.0.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_id="my-first-agent",
    data={
        "customer": {"email": "[e-mail]"},
        "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`.

`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": "…"},
})

CLI

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

Command	What it does
wytness-ai verify	Round-trip one signed event end-to-end and confirm it arrives in the dashboard.
wytness-ai tail	Stream recent dashboard events to your terminal. Flags: --agent-id, --limit, --interval, --once, --json.
wytness-ai validate-chain	Pull recent events and walk the AGT hash chain locally, proving link integrity without trusting the server. Flags: --agent-id, --limit, --json.

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:

Surface	Python	TypeScript	Notes
AGT AuditLog auto-wire	✓ — patches agentmesh.governance.audit.AuditLog	✓ — patches AuditLogger.prototype.log from @microsoft/agent-governance-sdk	Both default auto_wire=True / autoWire: true.
AGT GovernanceEvent auto-wire	✓ — registers WytnessEventSink on GovernanceEventProcessor	✗ — manual (getEventSink().emit(...))	AGT-TS 4.0.0 has no GovernanceEventProcessor registry. Tracked as V2-AGT-TS-EVENT-SINK; will wire automatically when Microsoft ships the registry.
Wire-format byte parity	✓	✓	Cross-language fixture in sdk-typescript-agt/tests/integration/wireFormat.parity.test.ts.
Ed25519 envelope signing	✓	✓	Identical canonical-JSON encoding + sig scope.
PII pseudonymisation	✓	✓	Identical HMAC + X25519 + ChaCha20-Poly1305 primitives.
Framework adapter compatibility tests	✓ — 12 integrations × 3 axes (import / monkey-patch survival / PII scrub)	✗ — AGT-TS ships zero framework adapters in 4.0.0

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_id="my-first-agent",
    data={
        "entry_id": "tool-invocation-2026-06-01-uuid-abc",
        "customer": {"email": "[e-mail]"},
        "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).