Python SDK

Typed Python client for the Watcher API. Provides sync and async clients with automatic retry, pagination, and structured error handling.

Installation

The SDK is distributed as a Python wheel attached to each Watcher client release on GitHub, next to the client binaries. Download the wheel from the release matching your server version and install it:

curl -fsSLO https://github.com/ApolloResearch/watcher-bin/releases/download/v<version>/watcher_sdk-<version>-py3-none-any.whl
pip install ./watcher_sdk-<version>-py3-none-any.whl

The package is not on PyPI. Each release's checksums.txt includes the wheel's SHA-256 if you want to verify the download. Requires Python 3.12+.

Quick Start

These examples authenticate with an organization API key, which exists on SSO-mode deployments. On a proxy-mode deployment there are no Watcher API keys; see Proxy-mode deployments below.

Synchronous

import os

from watcher_sdk import WatcherClient

with WatcherClient(
    "https://watcher.example.com/api",
    api_key=os.environ["WATCHER_API_KEY"],
) as client:
    # List all sessions
    sessions = client.list_sessions()
    print(f"Total sessions: {sessions.total}")

    # Iterate over all sessions (auto-paginates)
    for session in client.list_sessions_iter():
        print(session.session_id, session.trajectory_count)

    # Get available graders
    graders = client.list_graders()
    for g in graders.graders:
        print(g.name, list(g.scores.additional_keys))

Asynchronous

import os

from watcher_sdk import AsyncWatcherClient

async with AsyncWatcherClient(
    "https://watcher.example.com/api",
    api_key=os.environ["WATCHER_API_KEY"],
) as client:
    sessions = await client.list_sessions()

    async for session in client.list_sessions_iter():
        print(session.session_id)

Configuration

Both WatcherClient and AsyncWatcherClient accept the same constructor parameters:

Parameter	Type	Default	Description
base_url	str	"http://localhost:8000"	Watcher API base URL
api_key	str | None	None	Organization API key, sent via x-api-key header
access_token	str | None	None	JWT access token from the sign-in flow, sent via Authorization: Bearer header
timeout	float	30.0	Request timeout in seconds
max_retries	int	3	Max retries for transient failures
httpx_client	httpx.AsyncClient | None	None	Pre-configured HTTP client, for auth the other parameters cannot express (see below)
manage_external_httpx_client_lifecycle	bool	True	Whether the SDK closes a provided httpx_client on exit

For a deployed Watcher, base_url is the API base URL including its /api prefix, e.g. https://watcher.example.com/api; the SDK appends the /v1/... endpoint paths itself. The localhost default suits a local development server.

Two auth modes are supported:

• api_key -- sends the x-api-key header. Use with organization API keys, created on the Analyzer's Organization page (the OpenAPI spec calls them WorkOS API keys).
• access_token -- sends the Authorization: Bearer header. Use with the JWT access token from a user's sign-in.

If both are provided, access_token takes precedence.

# Using an organization API key
client = WatcherClient(
    base_url="https://watcher.example.com/api",
    api_key="my-org-api-key",
    timeout=60.0,
    max_retries=5,
)

# Using a JWT access token
client = WatcherClient(
    base_url="https://watcher.example.com/api",
    access_token="eyJhbG...",
)

Proxy-mode deployments

Proxy-mode deployments have no Watcher API keys: requests must satisfy your reverse proxy instead. Pass a pre-configured httpx.AsyncClient carrying whatever the proxy accepts (both WatcherClient and AsyncWatcherClient take an httpx.AsyncClient; the sync client drives async internally):

import httpx
from watcher_sdk import WatcherClient

http = httpx.AsyncClient(
    base_url="https://watcher.example.com/api",
    auth=("service-account", "password"),        # e.g. basic auth
    # or headers={"CF-Access-Client-Id": "...",
    #             "CF-Access-Client-Secret": "..."},
    timeout=30.0,
)

with WatcherClient(httpx_client=http) as client:
    ...

The provided client replaces the SDK-managed one entirely, so set base_url, auth, and timeout on it; the constructor's base_url, api_key, access_token, and timeout parameters do not apply to it. Retries, pagination, and error mapping still work as usual.

The remaining examples on this page abbreviate construction as WatcherClient(); against a real deployment, pass base_url and api_key as above (or httpx_client in proxy mode).

Both clients must be used as context managers (or opened/closed manually) to manage HTTP connections:

# Context manager (recommended)
with WatcherClient() as client:
    ...

# Manual lifecycle
client = WatcherClient()
client.open()
try:
    ...
finally:
    client.close()

Ingest

Ingest trajectories for storage and deduplication:

from watcher_sdk import WatcherClient, ClaudeCodeTrajectory

with WatcherClient() as client:
    response = client.ingest(
        source="my-app",
        trajectories=[trajectory],  # list of ClaudeCodeTrajectory objects
    )
    print(f"Session: {response.session_id}")
    print(f"Trajectory IDs: {response.trajectory_ids}")
    print(f"Ingested: {response.ingested_count}, Duplicates: {response.duplicate_count}")

IngestResponse fields:

• session_id — UUID of the session containing the trajectories
• trajectory_ids — list of leaf trajectory hashes (unique identifiers)
• ingested_count — number of newly ingested trajectories
• duplicate_count — number of trajectories skipped as duplicates
• skipped_count — number of trajectories skipped (e.g., empty entries)

Grade

Grade trajectories using configured LLM-based graders:

with WatcherClient() as client:
    # Grade specific trajectories. grader_ids is required: name every grader
    # to run (list them with client.list_graders()).
    response = client.grade(
        session_id="your-session-uuid",
        trajectory_ids=["trajectory-hash-1", "trajectory-hash-2"],
        grader_ids=["grader-uuid-1"],
    )

    for grade in response.grades:
        print(f"Trajectory: {grade.trajectory_hash}")
        print(f"Grader: {grade.grader_id}")
        print(f"Scores: {grade.grades.to_dict()}")

GradeResponse fields:

• grades — list of GradeResult objects
• not_found_ids — trajectory IDs that were not found

GradeResult fields:

• trajectory_hash, grader_id, session_id, model — identifiers (model is null for grades produced through monitors, where the model is a server-side choice)
• grades — score values (dict-like object)
• created_at — when the grade was produced
• error — error message if grading failed for this trajectory

Grading with Monitors

Watcher-managed monitors are addressed by role instead of grader UUID — the server resolves everything about how the grade is produced:

with WatcherClient() as client:
    monitors = client.list_monitors()            # the monitors this server runs
    response = client.grade_by_role(
        "triage",
        messages=[[{"role": "user", "content": "..."}]],
    )

list_monitors raises WatcherNotFoundError on servers without managed monitors; grade by grader UUID there instead.

Ingest and Grade (Convenience)

Ingest and immediately grade in one call:

with WatcherClient() as client:
    ingest_response, grade_response = client.ingest_and_grade(
        source="my-app",
        trajectories=[trajectory],
        grader_ids=["grader-uuid-1"],  # required: the graders to run
    )

List Graders

List available graders and their score definitions:

with WatcherClient() as client:
    response = client.list_graders()
    for grader in response.graders:
        print(f"{grader.name} ({grader.grader_id})")
        print(f"  Description: {grader.description}")
        for score_name, score_def in grader.scores.additional_properties.items():
            print(f"  {score_name}: {score_def.type_} [{score_def.min_value}-{score_def.max_value}]")

GraderSummary fields:

• grader_id — UUID of the grader
• name — human-readable name
• description — what the grader evaluates
• scores — mapping of score names to ScoreDefinition (type_, description, min_value/max_value)

Sessions

List Sessions

List sessions with pagination and sorting:

with WatcherClient() as client:
    response = client.list_sessions(
        limit=50,          # max results per page (1-1000, default 100)
        offset=0,          # pagination offset
        sort_by="ended_at",  # or "started_at"
        sort_order="desc",   # or "asc"
    )
    print(f"Showing {len(response.sessions)} of {response.total} sessions")
    for session in response.sessions:
        print(f"  {session.session_id}: {session.trajectory_count} trajectories")

Search Sessions

Search with filters (date range, external IDs):

from datetime import datetime, timezone

with WatcherClient() as client:
    response = client.search_sessions(
        start_date=datetime(2026, 1, 1, tzinfo=timezone.utc),
        end_date=datetime(2026, 2, 1, tzinfo=timezone.utc),
        external_ids={"anthropic-claude-code": ["session-abc-123"]},
        limit=100,
    )

The external_ids parameter filters by source-specific identifiers. Keys are namespace strings (e.g., "anthropic-claude-code" for sessions recorded from Claude Code), values are lists of IDs to match.

start_date and end_date select sessions whose activity overlaps the window: sessions with activity at or after start_date that started at or before end_date. A session's recorded span runs from its first through its latest message, so a session still in progress matches if it has had activity at or after start_date. Sessions with no messages never match a search, whatever the filters.

Get Session Detail

Get a single session with its trajectories and grades:

with WatcherClient() as client:
    session = client.get_session("your-session-uuid")
    print(f"Session: {session.session_id}")
    print(f"Time range: {session.started_at} — {session.ended_at}")
    print(f"External IDs: {session.external_ids}")
    for traj in session.trajectories:
        print(f"  Trajectory {traj.trajectory_id}: {traj.message_count} messages")

SessionSummary fields (in list/search responses):

• session_id — UUID
• trajectory_count, message_count — counts
• started_at, ended_at — time range
• external_ids — source-specific identifiers (e.g., {"anthropic-claude-code": "..."})
• metadata — arbitrary metadata
• grades — most recent grade per grader (list of GradeResult)
• max_score — highest score across all grades (or None)

Trajectories

List Trajectories in a Session

with WatcherClient() as client:
    response = client.list_trajectories("your-session-uuid")
    for traj in response.trajectories:
        print(f"{traj.trajectory_id}: {traj.message_count} messages, source={traj.source}")

Get Trajectory Detail

Get the full conversation with messages and grades. The view parameter picks the payload shape and defaults to TrajectoryView.TRUNCATED, which elides heavy payloads behind marker fields; pass TrajectoryView.EXTENDED explicitly for storage-faithful message content:

from watcher_sdk import TrajectoryView

with WatcherClient() as client:
    traj = client.get_trajectory("session-uuid", "trajectory-hash", view=TrajectoryView.EXTENDED)
    print(f"Trajectory: {traj.trajectory_id}")
    print(f"Source: {traj.source}")
    print(f"Messages ({len(traj.messages)}):")
    for msg in traj.messages:
        print(f"  [{msg.additional_properties.get('role', '?')}] {str(msg.to_dict())[:100]}...")
    print(f"Grades: {len(traj.grades)}")

TrajectoryDetailResponse fields:

• trajectory_id — leaf hash identifier
• messages — list of message objects (role, content, metadata)
• grades — list of GradeResult objects
• metadata — arbitrary metadata from the last message
• session_id, source, created_at

Health and Version

with WatcherClient() as client:
    # GET /v1/health: "healthy", or "unhealthy" when the deployment's
    # sign-in provider is unreachable (SSO mode)
    health = client.health()
    print(f"Status: {health.status} at {health.timestamp}")

    # GET /v1/version: the build stamped into the connected deployment
    build = client.version()
    print(f"Server: {build.version} ({build.git_sha})")

Pagination

Auto-paginate through all results with iterators:

with WatcherClient() as client:
    # Iterate all sessions (fetches pages on demand)
    for session in client.list_sessions_iter(page_size=50):
        print(session.session_id)

    # Iterate search results
    for session in client.search_sessions_iter(
        start_date=datetime(2026, 1, 1, tzinfo=timezone.utc),
        end_date=datetime(2026, 2, 1, tzinfo=timezone.utc),
        page_size=100,
    ):
        print(session.session_id)
        if some_condition:
            break  # stops pagination early

Async equivalents use async for:

async with AsyncWatcherClient() as client:
    async for session in client.list_sessions_iter(page_size=50):
        print(session.session_id)

Pages are fetched lazily — the iterator only makes HTTP requests as you consume items. Breaking out of the loop stops pagination immediately.

Error Handling

All errors inherit from WatcherError:

WatcherError
├── WatcherConnectionError      # Failed to connect to the API
├── WatcherTimeoutError         # Request timed out
└── WatcherAPIError             # API returned an error HTTP response
    ├── WatcherValidationError      # 400, 422 — bad request
    ├── WatcherAuthenticationError  # 401 — invalid/missing API key
    ├── WatcherNotFoundError        # 404 — resource not found
    └── WatcherRateLimitError       # 429 — rate limited

Status Code Mapping

HTTP Status	Exception
400	WatcherValidationError
401	WatcherAuthenticationError
404	WatcherNotFoundError
422	WatcherValidationError
429	WatcherRateLimitError
5xx	WatcherAPIError

Error Attributes

All WatcherAPIError subclasses have:

Attribute	Type	Description
status_code	int	HTTP status code
error	str	Error message from the API
message	str | None	Human-readable message (for structured error bodies, the message; else the stringified detail)
code	str | None	Stable machine-readable error code from a structured body — branch on this
body	dict | None	Raw parsed response body, lossless (None for SDK-synthesized errors)
request_id	str | None	Request ID for tracing

Structured error bodies ({"detail": {"code": ..., "message": ...}}) are flattened into code/message; the original shape stays available on body.

WatcherRateLimitError also has retry_after: float | None indicating how long to wait.

Examples

from watcher_sdk import WatcherClient, WatcherNotFoundError, WatcherAPIError

with WatcherClient() as client:
    try:
        session = client.get_session("nonexistent-id")
    except WatcherNotFoundError:
        print("Session not found")
    except WatcherAPIError as e:
        print(f"API error {e.status_code}: {e.error}")
        if e.request_id:
            print(f"Request ID: {e.request_id}")

Catch specific errors for fine-grained handling:

from watcher_sdk import (
    WatcherClient,
    WatcherValidationError,
    WatcherAuthenticationError,
    WatcherRateLimitError,
)

with WatcherClient() as client:
    try:
        client.grade(session_id="bad-uuid", trajectory_ids=["traj-1"], grader_ids=["grader-uuid-1"])
    except WatcherValidationError as e:
        print(f"Invalid request: {e.message or e.error}")
    except WatcherAuthenticationError:
        print("Check your API key")
    except WatcherRateLimitError as e:
        if e.retry_after:
            print(f"Rate limited. Retry after {e.retry_after}s")

Retry Behavior

The SDK automatically retries transient failures with exponential backoff and jitter.

Retried automatically:

• Connection errors (WatcherConnectionError)
• Timeouts (WatcherTimeoutError)
• Rate limits (WatcherRateLimitError / HTTP 429)
• Bad gateway (502), service unavailable (503), gateway timeout (504)

Not retried (client errors):

• Validation errors (400, 422)
• Authentication errors (401)
• Not found (404)
• Other 4xx errors

Backoff strategy: Exponential with jitter, starting at 0.5s, max 10s between retries. Default 3 retries (configurable via max_retries).

To disable retry:

client = WatcherClient(max_retries=0)  # no retries (1 attempt only)

Sync vs Async

	WatcherClient (sync)	AsyncWatcherClient (async)
Use when	Scripts, CLIs, Jupyter notebooks	Async applications, FastAPI routes
Context manager	with client:	async with client:
Pagination	for item in client.list_sessions_iter():	async for item in client.list_sessions_iter():
Event loop	Creates its own (works in Jupyter)	Uses the current event loop

The sync client works correctly in environments with existing event loops (e.g., Jupyter notebooks), unlike raw asyncio.run().

All methods have identical signatures and return types. The only difference is that async methods return coroutines (use await), and async iterators use async for.