ai-watcher 0.2.3

AI agent observability and control — AgentWatch SDK

ai-watcher

Python SDK for AIWatcher — AI agent observability and control.

Install

pip install ai-watcher

Quickstart

export AIWATCHER_API_KEY=aw_live_...

One-liner wrapper (Lambda / serverless)

from aiwatcher import track_llm

result = track_llm(
    'classify-document',
    lambda: openai.chat.completions.create(...),
    {
        'human_id': customer_id,
        'agent_name': 'doc-classifier',
        'model': 'gpt-4o',
        'framework': 'aws-lambda',
        'session_name': 'Classify: invoice',
        'input': {'doc_type': 'invoice', 'pages': 3},
    }
)

Chained model calls (multi-step pipeline)

from aiwatcher import track_chain

results = track_chain(
    steps=[
        {
            'action': 'extract',
            'model': 'gpt-4o',
            'fn': lambda: openai.chat.completions.create(...),
            'input': {'pages': 3},
        },
        {
            'action': 'classify',
            'model': 'claude-sonnet-4-20250514',
            'fn': lambda: anthropic.messages.create(...),
            'input': {'text': '...'},
        },
    ],
    opts={
        'human_id': customer_id,
        'agent_name': 'shipping-pipeline',
        'session_name': 'Process Shipping Document',
    }
)

Both wrappers are zero-dependency and never break your app — if AIWatcher is unreachable, the underlying function still runs.

Session context manager (full control)

from aiwatcher import Session, tool

with Session(human_id="[e-mail]") as session:

    @tool(session)
    def search_web(query: str) -> str:
        return "results..."

    result = search_web("AI agent security")

Full Session config

from aiwatcher import Session, tool

with Session(
    api_key="aw_live_...",
    human_id="[e-mail]",
    agent_name="billing-agent",
    agent_version="2.0.0",
    model="claude-sonnet-4-20250514",
    system_prompt="You are a billing assistant.",
    tools=["send_invoice", "fetch_invoice"],
    framework="langchain",
) as session:

    @tool(session, action_class="send", data_scope="financial")
    def send_invoice(recipient: str, amount: float) -> dict:
        return {"sent": True}

    send_invoice("[e-mail]", 1200.00)

API reference

track_llm(action, fn, opts)

Field	Type	Description
action	str	Name of the operation (e.g. "classify-document")
fn	Callable	Zero-argument callable that makes the model call
opts	dict	Session options (see below)

track_chain(steps, opts)

Each step: {'action': str, 'fn': callable, 'model': str, 'input': dict}

Common opts fields

Key	Default	Description
human_id	"anonymous"	User or customer identifier
agent_name	"agent"	Name of the agent/pipeline
model	"unknown"	Default model (overridden per step in track_chain)
framework	"python"	Runtime (e.g. "aws-lambda", "langchain")
session_name	action name	Human-readable session title in the dashboard
input	None	Input metadata to log with the event

Exceptions

Exception	When raised
ExecutionBlockedException	Policy blocked the tool call
HitlDeniedException	Human reviewer denied the action
AgentwatchAPIError	Non-2xx response from the API
AgentwatchConnectionError	Network error after retries

Environment variables

Variable	Default	Description
AIWATCHER_API_KEY	—	Required. Your aw_live_... key.
AIWATCHER_API_URL	https://agentwatch-pi.vercel.app	Override for self-hosted.