Python SDK

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

The integration examples show how to connect agents, frameworks, gateways and observability tools.

For a Copilot integration, use the Copilot quickstart. Its installer provides the SDK CLI and adapter without a source checkout or local Watcher app. Library users can install the optional cli extra for watcher-sdk auth login|status|logout.

Installation

The SDK is available as a Python wheel attached to each Watcher client release on GitHub. Releases with a watcher_sdk-<version>-source.tar.gz asset also include MIT-licensed SDK source under sdk/ in the release's Git tag. The source tree also includes sdk/openapi.json, the OpenAPI spec the generated client was built from. Wheels remain available. Both formats require Python 3.12 or later and provide the same SDK code. Both include py.typed markers, so type checkers such as mypy and Pyright check your calls against the SDK's type annotations.

Choose the release matching your server version. Replace <version> in the commands below with that version, without the leading v.

Install the wheel

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 digest.

Install source

Check that the selected release includes watcher_sdk-<version>-source.tar.gz. If it does not, install its wheel instead. For a release with source, install from the version tag with Git available:

pip install 'watcher-sdk @ git+https://github.com/ApolloResearch/watcher-bin.git@v<version>#subdirectory=sdk'

For a uv-managed project:

uv add 'watcher-sdk @ git+https://github.com/ApolloResearch/watcher-bin.git@v<version>#subdirectory=sdk'

To install without Git, download the source archive instead:

curl -fsSLO https://github.com/ApolloResearch/watcher-bin/releases/download/v<version>/watcher_sdk-<version>-source.tar.gz
pip install ./watcher_sdk-<version>-source.tar.gz

The source archive's SHA-256 digest is in the release's checksums.txt. The source export also contains integration examples under sdk/examples/. Extract the archive with tar -xzf watcher_sdk-<version>-source.tar.gz and keep the whole sdk/ directory together to use them. Each example has its own installation instructions and tests; installing the SDK does not install the examples. Older source releases may not contain examples.

Source installation builds the Python package locally, so an offline install also needs cached build and runtime dependencies. The generated client is included; no API schema generation is needed.

Prerelease source is available by tag before production promotion. When a release includes source, production promotion moves the repository's main branch to that release's tag. Promoting a release without source leaves main unchanged, so it can differ from the current production version. Pin a version tag or commit for reproducible installations. Third-party dependencies retain their own licenses; the SDK includes their license notices.

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.

The constructors accept two explicit credential types:

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...",
)

Local integrations on macOS and Linux can sign in through the browser. This works with Watcher Cloud and self-hosted WorkOS SSO. No local Watcher client is required.

import asyncio
import webbrowser
from pathlib import Path
from watcher_sdk import DeviceLoginPrompt, login

def open_login(prompt: DeviceLoginPrompt) -> None:
    webbrowser.open(prompt.url)

async def main():
    auth = await login("https://watcher.example.com/api", open_login)
    profile = Path.home() / ".config/watcher-sdk/integration.json"
    profile.parent.mkdir(parents=True, exist_ok=True)
    profile.write_text(auth.model_dump_json(), encoding="utf-8")
    async with auth.create_client() as client:
        sessions = await client.list_sessions(limit=10)

asyncio.run(main())

login checks the approval link and code before it calls your function. It raises ValueError if the link is not HTTPS, has no host, contains a user name or password, has a # fragment or is longer than 2048 characters. It also raises ValueError if the code is empty or longer than 32 characters.

Call login only from interactive setup. Save the returned IntegrationAuth reference for later use; its JSON contains the server and credential location, not tokens. create_client reads and refreshes the saved login without a browser. Later processes load the same reference:

from pathlib import Path
from watcher_sdk import IntegrationAuth

profile = Path.home() / ".config/watcher-sdk/integration.json"
auth = IntegrationAuth.model_validate_json(profile.read_text(encoding="utf-8"))

Use auth.create_client() in an async context, as in the setup example.

discover_logins lists usable saved SDK and Watcher logins, skipping unreadable or invalid files. Pass base_url to check one server; an invalid matching login then raises an error. The caller must select a login when more than one is available.

SDK logins live under ~/.config/watcher-sdk, or $XDG_CONFIG_HOME/watcher-sdk. WATCHER_SDK_HOME or an explicit state_dir selects another SDK store. Each server has a separate credential file. logout asks the server to end the session that the SDK login holds, so a copied refresh token stops working. Then it removes the SDK's local login. It leaves the Watcher client's saved login file alone. The server may end more than the SDK's sign-in, though: the browser that approved the login may be signed out of Watcher too, and so may a Watcher client that signed in through that browser. If the server cannot be reached, logout still removes the local login, and the server session lasts until it expires. An expired or rejected login never falls back to another identity.

login creates the credential folder so that only your account can open it. If that folder already exists, login restricts it the same way. It refuses a credential folder that is a symbolic link, so it never changes the permissions of another file or folder. The SDK refuses a saved login when another account owns it or can write to it. That account could replace the file with its own login, and your sessions would upload to its account. Run login again to replace a refused login.

verify_integration_auth checks ingestion and session read access without uploading a test session. User logins need access to their own sessions; API keys also need sessions:read:any. Successful setup does not prove a coding agent has loaded its hooks or delivered a session.

Local integrations on macOS and Linux can reuse the installed Watcher client's login. Sign in to Watcher for the target server, then run the integration as the same OS user:

import asyncio
from watcher_sdk import async_client_from_watcher

async def main():
    async with async_client_from_watcher("https://watcher.example.com/api") as client:
        profile = await client.auth_me()
        sessions = await client.list_sessions(limit=10)

asyncio.run(main())

The helper reads current tokens from Watcher's credential store on each request. It shares Watcher's file lock when refreshing a rejected token, then retries once. It requires a saved login for the exact server URL, including its port and path. It never opens a sign-in page or falls back to another login or API key. If the login is missing or its refresh token is rejected, sign in to Watcher again. Also sign in again if Watcher answers a refresh with a redirect: the SDK never follows it, and saves nothing.

The default store is ~/.apollo_monitor. Set APOLLO_MONITOR_HOME or pass state_dir=Path("/path/to/watcher-state") if Watcher uses another location. The helper works with the packaged app and needs neither a running Watcher daemon nor its Python package. Install the SDK in the integration's environment. Use explicit service credentials for remote gateways and unattended services.

For a custom HTTP client, use WatcherLoginAuth(base_url) as the auth on an httpx.AsyncClient, then pass that HTTP client through httpx_client to either SDK client. Token refreshes never go through your client. The SDK sends each one on its own connection, which never follows a redirect or keeps a cookie. Your client's cookie jar, hooks and redirect setting never see the refresh token. Supplying access_token alone does not enable saved-login discovery or refresh.

import httpx
from watcher_sdk import AsyncWatcherClient, WatcherLoginAuth

url = "https://watcher.example.com/api"
http_client = httpx.AsyncClient(base_url=url, auth=WatcherLoginAuth(url))
client = AsyncWatcherClient(base_url=url, httpx_client=http_client)

Your client's connection settings do not apply to refreshes either. Refreshes ignore the verify, cert, proxy, trust_env and transport options you set on it. By default, the refresh connection reads proxy and certificate settings from the environment, such as HTTPS_PROXY, HTTP_PROXY and SSL_CERT_FILE. It reads them even when your client sets trust_env=False.

If you set up TLS, client certificates or a proxy in code, build one httpx.AsyncHTTPTransport. Pass it to your client and to WatcherLoginAuth(base_url, transport=...), and refreshes go through it too. The SDK hands each refresh to that transport directly, so redirects and cookie jars still never apply. The SDK never closes the transport.

import ssl
import httpx
from watcher_sdk import AsyncWatcherClient, WatcherLoginAuth

url = "https://watcher.example.com/api"
tls = ssl.create_default_context(cafile="/etc/ssl/certs/watcher-ca.pem")
transport = httpx.AsyncHTTPTransport(verify=tls)
http_client = httpx.AsyncClient(
    base_url=url, auth=WatcherLoginAuth(url, transport=transport), transport=transport
)
client = AsyncWatcherClient(base_url=url, httpx_client=http_client)

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 stored leaf trajectories
duplicate_count — number of already-stored leaf trajectories echoed back
skipped_count — number of uploads skipped (e.g., empty entries)

One upload can produce several trajectory ids and counts: a Claude Code session in which the developer rewound the conversation stores one trajectory per branch. trajectory_ids lists them in order with the current (live) branch last — the last id is the one to use as the anchor for a subsequent append.

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.

To include or exclude session owners, pass a UserFilter to search_sessions or search_sessions_iter. Both WatcherClient and AsyncWatcherClient accept the same filter:

from uuid import UUID
from watcher_sdk import UserFilter, UserFilterMode, WatcherClient

users = UserFilter(
    ids=[
        UUID("aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"),
        UUID("bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb"),
    ],
    mode=UserFilterMode.EXCLUDE,
)

with WatcherClient() as client:
    response = client.search_sessions(users=users, limit=100)
    for session in client.search_sessions_iter(users=users, page_size=100):
        print(session.session_id)

ids must contain at least one user UUID, and mode is required. UserFilterMode.INCLUDE matches any listed owner; UserFilterMode.EXCLUDE removes all listed owners while keeping unattributed sessions. Omit users to search all sessions you are allowed to read. Naming another user requires sessions:read:any in either mode; otherwise the server returns HTTP 403.

Filter by coding agent

agent_platform is an open string in session responses. Clients must accept additional values. Watcher can recognize retained SDK adapter IDs, including on older sessions whose stored platform is other.

Use agent_platforms=["github_copilot_cli"] to find Copilot sessions. agent_platforms=["other"] excludes recognized agents. Multiple names select any of those agents; all other filters still apply. Counts and pagination use the same rules. Missing, unknown or conflicting IDs remain other.

The coding agent and its chosen model are separate. Recognition needs no adapter update, hook reinstall or new sign-in. It does not change recording, ownership or tool rules. See integration limits for the monitoring features each adapter provides.

Filter by ingestion source

Pass sources to search_sessions or search_sessions_iter to select how messages reached Watcher. Both sync and async clients accept this filter:

with WatcherClient() as client:
    for session in client.search_sessions_iter(
        sources=["watcher", "tailscale-aperture"],
        agent_platforms=["codex"],
    ):
        print(session.session_id, session.sources)

This selects Codex sessions with at least one retained message from either source. Selected sources use OR; other filters must also match. The source match covers the full session, even when a date filter selects its activity window. Source filtering also applies to the result count and every page.

Source key	Meaning
`watcher`	Messages uploaded by the Watcher client, shown as Watcher Live.
`tailscale-aperture`	Aperture imports and Aperture gateway-hook captures.
`watcher-gateway`	Gateway hook messages in a session whose conversation ID starts with a gateway example's prefix. Shown as Gateway.
`watcher-sdk`	Gateway hook messages in a session whose conversation ID starts with a registered SDK coding agent's prefix, such as `github-copilot-cli:`. Shown as Watcher SDK. It does not include your own agent's gateway hooks.
`watcher-hook`	Other gateway hook messages, including structured transcripts and raw provider exchanges, with any other conversation ID or none. Shown as Hooks API. This includes your own agent sending gateway hooks through the Watcher SDK, for example with `pre_tool_use_hook`, with its own conversation ID.

Hook captures whose metadata has no transport identifier are classified as Aperture. The last three keys come from the session's other external ID; stored message sources stay watcher-hook. The gateway prefixes are bifrost:, cloudflare:, helicone:, kong:, litellm:, openrouter: and portkey:. A prefix counts only when more text follows it, and matching is case-sensitive. To select every non-Aperture gateway hook session, pass all three keys: sources=["watcher-gateway", "watcher-sdk", "watcher-hook"]. Add tailscale-aperture to include Aperture. The ingestion source is separate from the detected agent platform and provider.

Custom source strings, such as the source you pass to ingest, are accepted and matched exactly. Omit sources or pass None to select all sources; an empty list returns HTTP 422.

Session list, search and detail responses return sources as a sorted list of distinct sources among retained messages. A mixed session can have several sources. This is not a complete ingestion history: duplicate messages keep the source first stored, and removed messages no longer contribute. Scrubbed sessions return an empty list and do not match a source filter.

Search Tool Calls

List individual tool calls rather than the sessions containing them:

with WatcherClient() as client:
    response = client.search_tool_calls(
        severities=["critical", "high"],
        decision_categories=["human_denied", "auto_denied"],
        limit=50,
    )

Each row is one tool call Watcher recorded a permission decision about. Check content_state before reading its details: available rows include the tool and argument preview, while scrubbed rows retain only the content-free verdict, owner, and timestamp. Calls Watcher never saw are not listed.

severities takes the severity bands critical, high, medium and low, plus not_scored for calls with no usable policy score. This can include ordinary unscored decisions and calls whose grading was bypassed, skipped, or failed. Read each row's coverage_state to distinguish those cases.

decision_categories takes the notable outcomes plus cleared, which selects the calls Watcher allowed without asking anyone.

sort_by takes severity, decision, occurred_at or tool_name. Sorting by decision orders by what Watcher did, with the calls it stopped first. Tool-name filters cannot match scrubbed rows because Watcher does not retain tool names after session content is removed.

Read scope matches the sessions endpoints: without organization-wide session read you see only your own calls, and asking for another person's is refused.

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)
sources: sorted, distinct ingestion sources among retained messages

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.

Found an issue with this page? Report it on GitHub.

Installation​

Install the wheel​

Install source​

Quick Start​

Synchronous​

Asynchronous​

Configuration​

Sign in without a Watcher client​

Reuse a saved Watcher login​

Proxy-mode deployments​

Ingest​

Grade​

Grading with Monitors​

Ingest and Grade (Convenience)​

List Graders​

Sessions​

List Sessions​

Search Sessions​

Filter by coding agent​

Filter by ingestion source​

Search Tool Calls​

Get Session Detail​

Trajectories​

List Trajectories in a Session​

Get Trajectory Detail​

Health and Version​

Pagination​

Error Handling​

Status Code Mapping​

Error Attributes​

Examples​

Retry Behavior​

Sync vs Async​