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

Sign in without a Watcher client

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 sign-in 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 sign-ins, skipping unreadable or invalid files. Pass base_url to check one server; an invalid matching sign-in then raises an error. The caller must select a sign-in when more than one is available.

SDK sign-ins 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 sign-in holds, so a copied refresh token stops working. Then it removes the SDK's local sign-in. It leaves the Watcher client's saved sign-in file alone. The server may end more than the SDK's sign-in, though: the browser that approved the sign-in 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 sign-in, and the server session lasts until it expires. An expired or rejected sign-in 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 sign-in when another account owns it or can write to it. That account could replace the file with its own sign-in, and your sessions would upload to its account. Run login again to replace a refused sign-in.

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

Reuse a saved Watcher sign-in

Local integrations on macOS and Linux can reuse the installed Watcher client's sign-in. 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 sign-in for the exact server URL, including its port and path. It never opens a sign-in page or falls back to another sign-in or API key. If the sign-in 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 sign-in 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
• cost_usd: estimated request cost for role grading; null for grader-UUID grading

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.

grade_by_role calls POST /v1/monitors/{role}/grade. Its response includes cost_usd, the estimated USD cost across this request's trajectories and retries. Only the selected role runs. A supplied triage assessment adds no cost. The estimate uses the prices set on the server. It is null when usage or pricing is missing. For stored-session grading, it is also null if you cannot read that session's findings or content. Otherwise, it is 0 when no model runs. Older servers may omit the field. Provider retries whose usage is not reported cannot be included.

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:findings: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. Argument previews require content access and are null without it. 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 sessions:findings:read:any 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

Request a Deep Review

Deep Review runs on its own for sessions the blocking pipeline scored highly. You can also request one for any session. The review covers the session's latest transcript and runs in the background, which can take several minutes.

import time

with WatcherClient() as client:
    review = client.request_deep_review("your-session-uuid")
    deadline = time.monotonic() + 3600
    while review.status in ("queued", "running"):
        if time.monotonic() > deadline:
            raise SystemExit(f"Deep Review still {review.status} after an hour")
        time.sleep(30)
        review = client.get_deep_review_request("your-session-uuid", str(review.operation_id))
    if review.status == "failed":
        raise SystemExit(f"Deep Review failed; can_retry={review.can_retry}")

    trajectory = client.get_trajectory("your-session-uuid", review.trajectory_hash)
    report = [
        g for g in trajectory.grades
        if g.trajectory_hash == review.trajectory_hash
        and g.grades.additional_properties.get("type") == "report"
    ]

Requested reviews need an operation worker. The shipped Kubernetes Helm chart does not include one, so requested reviews stay queued. Scheduled Deep Reviews still run through the scheduler.

In Docker Compose, the operation-worker profile starts xylon-operation-worker, which requires PostgreSQL 17 or later. New installs enable this profile in .env.example. Without a running worker, requested reviews stay queued, so the example stops polling after a deadline.

Method	HTTP request	Permission
request_deep_review(session_id)	POST /v1/sessions/{session_id}/deep-review	The session's owner, or sessions:grade:any
get_deep_review_request(session_id, operation_id)	GET /v1/sessions/{session_id}/deep-review/{operation_id}	Findings access to the session
get_current_deep_review(session_id)	GET /v1/sessions/{session_id}/deep-review	Findings access to the session

Requests have no body. Admins hold sessions:grade:any. An organization API key has no owner, so it needs sessions:grade:any to request a review and sessions:findings:read:any to poll. For a user sign-in, findings access means your own sessions, or every session with sessions:findings:read:any. The example's get_trajectory call also requires content access. API keys need both content and findings grants to retrieve the report through that endpoint. A session you may not review or read returns 404 (WatcherNotFoundError), never 403. A scrubbed session also returns 404 to a request.

request_deep_review and get_deep_review_request return a DeepReviewRequestResponse:

• operation_id: the ID to poll.
• status: queued, running, succeeded or failed.
• trajectory_hash: the transcript the review covers.
• can_retry: for a failed review, whether the latest transcript's retry budget allows another attempt. It is also true for queued, running and succeeded reviews, which another request reuses.

When the status is succeeded, the report is the session's Deep Review grade for that trajectory_hash. Read that trajectory even if the session has gained messages, and select a grade with type: report rather than a specific grader ID. This includes reports written by older releases. A newer review of the same session can replace it, leaving the report list empty. Unavailable session content returns 404. Asking again while the transcript is unchanged returns the same review. Asking after it failed starts a new attempt, until the transcript has failed too many times. get_current_deep_review returns a wrapper whose current field holds a DeepReviewRequestResponse: any unfinished review of the session, whoever requested it, otherwise the latest review of its current transcript. current is None when no review was requested.

A request can fail with these WatcherAPIError status codes:

Status	code	Meaning
402		The organization is over its monthly grading limit.
409	nothing_to_review	The session has no transcript yet. Try again after it records messages.
409	review_unavailable	This transcript has failed too many times, or the session was scrubbed while the request was being handled.

A self-hosted deployment needs its background-job worker to run requested reviews; without it, they stay queued. See What drives the schedule when you self-host.

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.