Python SDK

The Python SDK decides which model each step of your agent uses, and reports metadata about what happened. Your provider SDK and your provider keys stay yours: the SDK never calls a model itself, it calls the function you hand it.

Install

Open an account at https://app.ergoda.com/onboarding first; that is the only place one is made. Then, in your project:

pip install ergoda
ergoda login
ergoda init --frontier anthropic/claude-opus-5 --step resolve_ticket
source ergoda.env

ergoda login opens your browser to approve this terminal and writes ERGODA_URL, ERGODA_KEY and ERGODA_SITE to ergoda.env. ergoda init writes a routing policy that routes nothing, so nothing changes until your team approves a step on the dashboard, and adds ERGODA_POLICY and ERGODA_DEFAULT_MODEL to the same file. Without the hosted service, `ergoda onboard TRACES --step NAME --serve CHEAP --escalate-to FRONTIER --approved-by "Your Name" approves it in routes.yaml` instead; approve a step in one place (see the quickstart).

Configure

Client.from_env() reads its settings from the environment, so your code names no secret:

Keyword arguments to from_env override the environment, environment= included.

Missing or unreadable configuration does not raise, because the client is usually built at import time. It prints one line per problem on stderr and runs as a pass-through: every call runs on the model it names, and nothing is sent. The exception is an environment outside the four, which raises: read as production it would mix staging rows into your evidence, and read as anything else it would withhold production's.

Wrap a call

Wrap the provider call your agent already makes. The arguments and the response stay exactly as they were:

import anthropic
from ergoda.sdk import Client

client = anthropic.Anthropic()
ergoda = Client.from_env()
create = ergoda.wrap({"anthropic": client.messages.create}, step="resolve_ticket")

# unchanged from here: same arguments, same response object
reply = create(model="claude-opus-5", max_tokens=1024, messages=messages)

With OpenAI, or any API that speaks its format:

import openai
from ergoda.sdk import Client

client = openai.OpenAI()  # or OpenAI(base_url=...) for any compatible API
ergoda = Client.from_env()
create = ergoda.wrap({"openai": client.chat.completions.create}, step="resolve_ticket")

# unchanged from here: same arguments, same response object
reply = create(model="gpt-5.6-sol", messages=messages)

• step names the step, and it is what the dashboard shows. Leave it out and the step is found from the shape of the messages.
• model= is your frontier model: what the call uses whenever no approved route says otherwise. Only model= is ever replaced; every other argument is passed through untouched.
• An async client (AsyncAnthropic, AsyncOpenAI) gets an async wrapped call, awaited like the one it wraps.
• A Responses call that continues a stored conversation (previous_response_id or conversation; the Agents SDK sends one under auto_previous_response_id) carries only its newest turns, so the checks could not see the conversation they would judge. It goes to the provider as written, on your model, and is not recorded. The framework adapters below do the same.

One wrap per call site, each with the client that call already uses:

import anthropic
import openai
from ergoda.sdk import Client

ergoda = Client.from_env()
# one wrap per call site, each with the client that call already uses
claude = ergoda.wrap({"anthropic": anthropic.Anthropic().messages.create}, step="resolve_ticket")
gpt = ergoda.wrap({"openai": openai.OpenAI().chat.completions.create}, step="draft_reply")

# unchanged from here: same arguments, same response objects
reply = claude(model="claude-opus-5", max_tokens=1024, messages=messages)
draft = gpt(model="gpt-5.6-sol", messages=messages)

Frameworks

An agent built on a framework never calls the provider SDK itself, so there is no create to wrap. Route the framework's model object instead: the adapter returns the same object, configured, so the change is one line. Every model call the framework makes then goes through wrap, with its routing, its checks, and the frontier's redo of anything they refuse. Streaming, tool calls, structured output and async work as before, and the model you configured is your frontier, as model= is under wrap. Ergoda does not install the frameworks; an adapter uses the one you have.

LangChain and LangGraph, for ChatOpenAI and ChatAnthropic:

from langchain_anthropic import ChatAnthropic
from ergoda.sdk import Client
from ergoda.sdk.integrations import langchain

ergoda = Client.from_env()
llm = langchain.route(ChatAnthropic(model="claude-opus-5"), ergoda)

Inside a LangGraph graph each node is its own step, named after the node. Outside one, the step is found from the shape of the messages. Pass step= to name it yourself.

The OpenAI Agents SDK, with its Responses or Chat Completions model:

from agents import Agent
from ergoda.sdk import Client
from ergoda.sdk.integrations import openai_agents

ergoda = Client.from_env()
model = openai_agents.route("gpt-5.6-sol", ergoda, step="triage")
agent = Agent(name="triage", instructions=instructions, model=model)

A model name is resolved the way the Agents SDK resolves it. To route every agent at once, route the client instead: set_default_openai_client(openai_agents.route(AsyncOpenAI(), ergoda)).

Pydantic AI, with its OpenAI and Anthropic models:

from pydantic_ai import Agent
from ergoda.sdk import Client
from ergoda.sdk.integrations import pydantic_ai

ergoda = Client.from_env()
agent = Agent(pydantic_ai.route("anthropic:claude-opus-5", ergoda, step="triage"))

Where an adapter differs from wrap:

• A caller past Client(caller_escalations_per_hour=) gets a 429, which the provider SDK raises as its own rate-limit error without retrying, where wrap raises EscalationCeilingExceeded.
• Azure OpenAI, Bedrock and Vertex clients name the model in the URL rather than the request, so they are refused, as is the Agents SDK's websocket transport.
• An OpenAI-compatible API that your routes name as another provider takes provider= with that name.
• Tracing nests the other way. The adapter runs inside the provider SDK's HTTP transport, below the method your provider instrumentation wraps, so that instrumentation's one span for the call is the parent of ergoda.step, and the cheap answer, a check and the frontier redo inside it are HTTP sends with no inference span of their own. The call's span reports the answer you got: its usage leaves out a cheap answer the checks refused, which Ergoda still records.

What the Python SDK checks

On a routed step the cheaper model answers first, and the Python SDK checks its answer in your process before you get it: the format, the tool calls, loops, and what your assurance level allows. Anything the checks refuse is redone by your frontier model, and you get that answer instead. ERGODA_ASSURANCE sets the level; the default, decided, leaves every decision to your frontier until a step earns it (see concepts and Autopilot).

The Python SDK also keeps a cost ceiling per route: each routed step is charged to its route's bill, and a route that has cost more than your frontier alone would have (5% more by default, the route's max_cost_ratio) serves your frontier until its bill recovers. The TypeScript SDK does the same from 0.1.1 (@ergoda/sdk 0.1.0 has no cost ceiling).

wrap assumes you act on the answer it hands you. If you don't, say so, so the evidence stays honest:

ergoda.last_step().commit(False)

Declaring a caller

Say which of your customers a request is for, and the dashboard's Customers page and GET /v1/sites/{site}/customers show cost per successful task for each of them. wrap reads it from the call's own field: OpenAI's user or safety_identifier, or Anthropic's metadata.user_id. A step made by hand takes it as caller:

with ergoda.step("resolve_ticket", messages=messages, caller=customer_id) as step:
    ...

Only a digest of the id is sent, salted with your site id and keyed with ERGODA_DIGEST_KEY when you set one. ergoda customers --ids customers.txt digests your ids the same way on your machine and prints the report with them in it; ergoda.commitments.caller_digest(site_id, customer_id) is that digest in code.

Outcomes

If something downstream went wrong with a step's answer, report it. It is joined to that request and counts as evidence:

ergoda.last_step().outcome(corrected=True)  # also: retried, escalated_to_human, tool_error

That is about one step. How the whole task ended is a separate report, sent once the task is over, however late:

# "success", "failure" or "unknown"
step.task_outcome("success", source="ticket_closed")
# by the task's messages
ergoda.task_outcome(messages, "failure")
# by the trajectory digest
ergoda.task_outcome(digest, "unknown", cohort="eu")

The task is named by the trajectory digest its steps already carried; the messages are digested on your machine, and your own task or ticket id is refused rather than sent. Send unknown when you could not tell: a task with no report reads as pending, then missing, and neither counts as a failure. A step's corrected=True is never read as a failed task. source and the optional cohort are short lowercase labels (user_feedback, tier-a), never a person. The cost per successful task report reads these.

Trajectories and recording

A trajectory is one run of your agent. Grouping calls into one lets Ergoda tell the steps of one run from independent ones:

from ergoda import record, trajectory

create = ergoda.wrap({"anthropic": record(client.messages.create)}, step="resolve_ticket")
with trajectory(ticket_id):  # one agent run
    reply = create(model="claude-opus-5", max_tokens=1024, messages=messages)

record appends each exchange to a local file (traces.jsonl by default) for the commands below. That file is yours and stays on your machine. wrap hands its step= to the record inside it, so each line carries the step's name and ergoda analyse lists the step by that name; pass record(..., step="name") to choose a different one. trajectory(id) takes any run id and writes it as a string.

Tracing

If your process traces with OpenTelemetry, each step adds one span, ergoda.step, to the trace it runs in, so Langfuse, Datadog, Honeycomb or LangSmith show what Ergoda decided beside your own spans:

Under wrap and with ergoda.step(...) the span is current while the step runs, so the spans your provider instrumentation writes for each call (the cheap answer, a check, the frontier redo) appear inside it. A step that escalated shows both calls under one ergoda.step. A bare ergoda.record() adds the span beside your call's span instead. Under a framework adapter it is the other way round: your call's span is the parent of ergoda.step.

The span carries no prompt, output, tool name or caller id, and no token counts: your instrumentation already puts usage on its own spans, and repeating it would count the step twice. Langfuse records it as a span, not a generation, with the ergoda.* attributes in its metadata; Datadog LLM Observability shows it as a workflow span.

There is nothing to configure. Ergoda writes the span only when a recording span is current, and never imports OpenTelemetry itself, so a process that does not trace pays one dictionary lookup per step. `pip install 'ergoda[otel]' pins a compatible opentelemetry-api`. Tracing comes after 0.1.5.

Pins and deploys

A route can be pinned to the system prompt and tool catalogue it was measured on (ergoda sdk pin). After a deploy that changes either, every request on that step goes to your frontier, which is correct: the evidence was about the old prompt. It also stops the step saving anything. The dashboard and your alerts say so when it happens; ergoda pins check says so before you ship.

Capture the requests your test suite builds, from the call your code makes (outside wrap, or with no Ergoda client at all), and check them in CI:

from ergoda import record

create = record(client.messages.create, "pin-samples.jsonl",
                step="resolve_ticket", requests_only=True)

ergoda pins check pin-samples.jsonl --policy routes.yaml   # exits 1 on a broken pin

requests_only writes each request before the call and nothing of the response, so a fake client works. Without --policy the check reads your live policy with the key ergoda login wrote; --saving adds what each broken step saved over the last 30 days, with a read key. If the change is intended, measure the step on it, then run the ergoda sdk pin line the check prints and publish the policy. pins check and requests_only come after 0.1.5.

Delivery

Events are batched and sent in the background. Sending never raises and never blocks your call. ergoda.flush() sends what is queued now, ergoda.close() flushes and stops, and the client also flushes when your process exits.

close() retries each batch up to four times, about two seconds against a control plane that is down. Events still undelivered then are written to the disk spool (below), and one line on stderr says how many and where; with no spool they are dropped with the client, and the line says how many and why. While the control plane is away the client keeps at most 1,000 events in memory (20 batches of 50) and drops the oldest beyond that, counting them in the next batch it sends; Client(max_pending_batches=) changes the limit.

The disk spool. Events the client could not deliver go to disk in two cases: when the control plane answers with a billing hold (HTTP 402, see billing), at once and for the whole queue, and when the process ends or close() returns with events left. The next client for the same site and control plane, in this process or a later one, sends them first, oldest first, and deletes each file once the control plane has taken it. While a hold stands a client sends one batch a minute to find out whether it has lifted, not the backlog. In between it writes each batch to disk as it fills and keeps only the partly filled one in memory, so a lifted hold is noticed up to a minute late.

• Where. $XDG_STATE_HOME/ergoda/spool, else ~/.local/state/ergoda/spool, one directory per site and control plane. ERGODA_SPOOL_DIR=/some/dir moves it; ERGODA_SPOOL_DIR=off turns it off. The directory is created mode 0700 and each file 0600; a directory that another user owns or could write turns the spool off, since every file in it is sent with your key.
• What. Each file is one event batch exactly as it would have been sent. Nothing reaches the disk that would not have reached the control plane.
• How much. At most 100,000 events, 64 MB and 30 days. Past any of the three the oldest files are deleted first and counted as dropped, like any other drop (ergoda.dataplane.holdspool: MAX_EVENTS, MAX_BYTES, MAX_AGE_S).
• Sent twice. Ingest keeps one copy of a batch id and of each event id, so a file two processes both send, or one sent again after a lost reply, counts once.
• Where it stops. A process with no writable home directory, which includes most serverless platforms, has no spool: the first failed write turns it off for that process and the client falls back to memory. On a platform that keeps /tmp between warm invocations you can point ERGODA_SPOOL_DIR there, but a cold start loses it. A worker that ends with os._exit skips the exit write as it skips the exit flush.

Forking servers. Build the client once at module scope. A client built in a parent that then forks workers (gunicorn --preload, Celery's prefork pool) works in each worker: the child starts its own sender and its own buffer, and what the parent had queued stays the parent's to send. A worker that ends with os._exit, which skips the exit flush, should call ergoda.close() in its shutdown hook (Celery's worker_process_shutdown signal, gunicorn's worker_exit).

When Ergoda can't be reached

Your calls keep running. The SDK keeps the last routing policy it fetched, and before its first successful fetch every step uses your own model: an unreachable control plane never sends a step to a cheaper model. The policy is fetched again every five minutes.

The last good policy is not served for ever. After 24 hours with no answer from the control plane (a refused key, an error or a hang all count as none), every step uses your own model and the SDK says so once on stderr, until a fetch succeeds. That is how a revocation or an expiry reaches a process that has lost its connection. Client(policy_max_age_s=) or ERGODA_POLICY_MAX_AGE_S changes the 24 hours; 0 turns it off. The proxy under --policy-from-control-plane does the same by forwarding every request as the caller asked.

With ERGODA_URL unset, a client given a policy file routes from it, sends nothing and never expires it. That is how routing continues with no control plane at all; see concepts, "If Ergoda is unreachable, or goes away". What the SDK adds to a step's time is under "Latency" on the same page.

Find what can move, from traces you already have

ergoda onboard traces.jsonl --step resolve_ticket \
  --serve anthropic/claude-haiku-4-5 --escalate-to anthropic/claude-opus-5
ergoda analyse traces.jsonl --model anthropic/claude-haiku-4-5 --html report.html

For what leaves your process and what never does, see Security and data.