Metadata-Version: 2.5
Name: agentaudit-python
Version: 0.1.1
Summary: EU AI Act Article 12 audit trails for AI agents
Author: Sathya Prakash J
License: MIT
License-File: LICENSE
Keywords: agents,ai,audit,compliance,eu-ai-act,langchain,logging,observability
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
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: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Provides-Extra: dev
Requires-Dist: build>=1.2; extra == 'dev'
Requires-Dist: langchain-core>=0.3; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: ruff>=0.8; extra == 'dev'
Requires-Dist: twine>=6; extra == 'dev'
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.3; extra == 'langchain'
Description-Content-Type: text/markdown

# agentaudit — Python SDK

Tamper-evident audit logging for AI agents, built for EU AI Act Article 12
record-keeping. Three lines of code get events flowing to your AgentAudit
backend; a hash chain over every event makes tampering detectable after the
fact.

```bash
pip install agentaudit-python   # PyPI distribution name; import is still `agentaudit`
```

```python
import agentaudit

agentaudit.init(api_key="aa_live_sk_...", agent_id="my-support-agent")
agentaudit.log_event(event_type="decision", action="route_to_human")
```

## Get your API key

From your AgentAudit dashboard: sign up, create a project, then go to
**Settings → API Keys → Create key**. The raw key (starts with
`aa_live_sk_`) is shown exactly once — copy it immediately. It is never
recoverable after that; losing it means revoking it and creating a new one.

Store it as an environment variable, not in source:

```python
import os
import agentaudit

agentaudit.init(api_key=os.environ["AGENTAUDIT_API_KEY"], agent_id="my-support-agent")
```

## Three ways to log events

### `@agentaudit.trace` — simplest, framework-agnostic

```python
@agentaudit.trace
def search_documents(query: str, top_k: int = 5) -> list:
    return vector_db.similarity_search(query, k=top_k)
```

Logs a `tool_call` event per call: `action` is the function's qualified
name, `input_hash`/`output_hash` are SHA-256 digests of the arguments and
return value (never raw), `metadata.duration_ms` is execution time. A raised
exception is logged as `severity="critical"` with `metadata.error`, and
always still propagates — `@trace` only observes, it never changes what the
wrapped function does. Works on sync and async functions.

### `LangChainHandler` — for LangChain / LangGraph

```python
from agentaudit.integrations import LangChainHandler  # pip install agentaudit-python[langchain]

chain.invoke(inputs, config={"callbacks": [LangChainHandler()]})
```

Handles `on_llm_start`/`on_llm_end` (and `on_chat_model_start`, which chat
models call instead), `on_llm_error`, `on_tool_start`/`on_tool_end`,
`on_tool_error`, and `on_chain_error`. **LLM calls log model name and token
usage only — no prompt or response content, hashed or otherwise.** This is
stricter than the SDK's default elsewhere: a callback wired into every LLM
call in a chain has no per-call decision behind it, so it defaults to the
safest reading. Tool calls are hashed, matching `@trace`.

All three error paths log the identical shape: `event_type="error"`,
`severity="critical"`, `metadata={"error": {"type": ..., "message": ...}}`.
`on_tool_error` fires even when the surrounding chain catches the exception
and completes normally — a failure the chain recovered from is still worth
an audit trail.

Not implemented: a start/end event for the chain run as a whole (that's
session-grouping territory, not a per-call audit gap — every LLM call, tool
call, and error inside the chain is already captured on its own).

### `agentaudit.log_event()` — manual, for anything custom

```python
agentaudit.log_event(
    event_type="decision",
    action="route_to_human",
    metadata={"reason": "low_confidence", "confidence_score": 0.23},
    severity="warning",
)
```

`event_type` is one of `tool_call`, `llm_call`, `decision`, `data_access`,
`error`, `custom`. `severity` is one of `info`, `warning`, `critical`
(default `info`). Pass `inputs=`/`outputs=` to get the same hashing
guarantee `@trace` applies automatically.

## Privacy

Inputs and outputs are SHA-256 hashed, never sent raw, by default
(`hash_inputs=True`). `metadata` is the one exception — it is not hashed,
because it's yours to control; whatever you put there is sent and stored
as-is. Pass `hash_inputs=False` to `init()` to opt into raw logging instead;
raw values then travel under the reserved keys `metadata.raw_input` and
`metadata.raw_output`, since the event schema has no top-level field for
them. `init(environment=...)` is likewise recorded as `metadata.environment`.

## Delivery

Capture never blocks: `log_event` appends to an in-memory queue and returns
(~0.05 ms). A daemon thread owning its own asyncio loop and `httpx.AsyncClient`
drains that queue every `flush_interval` seconds, or as soon as `batch_size`
events are waiting — whichever comes first. One implementation serves both
sync and async callers; nothing touches your event loop if you have one.

- `flush(timeout=10.0)` forces a send and blocks until it completes.
- Failed batches are retried `max_retries` times with exponential backoff and
  jitter, then spilled to a JSONL file in the temp directory. The buffer is
  keyed to the API key and `agent_id`, so a restarted process finds its backlog.
- While the backend is down, new events queue behind the buffered ones so
  delivery order matches production order. Buffered events replay first.
- A repeated failure starts a cooldown between cycles, so a long outage does
  not cost a full retry budget every `flush_interval`. An explicit `flush()`
  ignores it.
- Permanent failures (`400` malformed event, `401` revoked key) are discarded
  rather than buffered — they would fail identically forever. A revoked key
  is latched after the first failure, so it fails once, not on a loop.
- An `atexit` hook flushes whatever is queued at process exit.

## Verify your integration

```python
agentaudit.flush()
print(agentaudit.status())
# {'connected': True, 'events_sent': 47, 'events_queued': 0, 'events_buffered': 0,
#  'events_dropped': 0, 'last_flush': '2026-08-18T10:30:00.000Z', 'last_error': None}
```

`events_dropped` above zero means audit events were lost — check
`last_error`, raise `max_queue_size`, or investigate why the backend is
unreachable. Then open your AgentAudit dashboard's Timeline — you should see
the same events there.

## Configuration

```python
agentaudit.init(
    api_key="aa_live_sk_...",
    agent_id="my-agent",
    environment="production",       # -> metadata.environment on every event
    batch_size=50,                  # events per batch (default: 50)
    flush_interval=5.0,             # seconds between flushes (default: 5.0)
    max_queue_size=10_000,          # max events in memory queue (default: 10000)
    timeout=10.0,                   # HTTP timeout in seconds (default: 10.0)
    max_retries=3,                  # retries per batch after the first attempt (default: 3)
    base_url="https://your-agentaudit-backend",  # where your team's backend runs
    debug=False,                    # print debug logs to stderr
    enabled=True,                   # set False to disable without removing code
)
```

**Implemented (0.1.0):** `init`, `log_event`, `set_session`, `status`,
`flush`, `shutdown`, `@trace`, `LangChainHandler`, input/output hashing,
event builder, bounded queue, batch sender, offline buffer, retries, atexit
hook.
