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="you@example.com") 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="sarah@acme.com",
    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("client@acme.com", 1200.00)

API reference

track_llm(action, fn, opts)

track_chain(steps, opts)

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

Common opts fields

Exceptions

Environment variables