Python SDK

The Python SDK runs hosted AgentOS agents from your code and reports agents that run in your own code, with the same history, alerts and reports. Two jobs in one package:

Agents that run in your own code: start a run, emit events and spans, end it. It appears in AgentOS with the same run history, alerts and reports as a hosted agent.
Hosted agents: run them from your code with invoke(), stream their output, continue conversations and answer their questions.

Built for production: telemetry is batched and delivered from a background thread with automatic retries, so instrumentation adds no latency to your agent and never interrupts its work.

Package: pyagentos-sdk (import as agentos) · Python 3.10+

Installation

pip install pyagentos-sdk

AgentOS

from agentos import AgentOS

aos = AgentOS()  # reads AGENTOS_API_KEY and AGENTOS_AGENT_ID

Every option can be passed explicitly instead:

aos = AgentOS(api_key="aos_agent_...", agent_id="your-agent-id")

Config options

Option	Default	Description
`api_key`	`AGENTOS_API_KEY`	An agent key (`aos_agent_…`) or a workspace key (`aos_ws_…`). Required.
`agent_id`	`AGENTOS_AGENT_ID`	Default agent. With a workspace key you can pass `agent_id` per call instead.
`base_url`	`AGENTOS_BASE_URL`, then `https://agentos-ai.dev`	For self-hosted instances
`workspace`	`AGENTOS_WORKSPACE`	Workspace slug or id for control-plane calls. Needed with a personal key.
`batch_interval_ms`	`500`	ms between event flushes
`batch_size`	`50`	Max events per request
`max_queue_size`	`1000`	Events held per run. When full, the oldest debug/info events are dropped first.
`max_retries`	`5`	Retries for transient failures, with backoff from 0.25 s to 4 s
`timeout_seconds`	`30.0`	Per-request timeout for reporting calls
`disabled`	`AGENTOS_DISABLED`	No network and no side effects, for tests
`on_error`	logs the first one	Called with background failures, such as a batch that could not be delivered
`http_client`	a new `httpx.Client`	A custom client, for example behind a proxy

Use the client as a context manager to flush and close it at the end:

with AgentOS() as aos:
    ...

Hosted agents

`aos.invoke(input=None, **options)`

Runs a hosted agent and waits for it (up to the server's 300 s limit). Always check status: a run can finish, or pause for an approval or a question.

result = aos.invoke({"ticket_id": 4521})

if result.status == "completed":
    print(result.output)  # e.g. {"category": "billing", "reply": "..."}
elif result.status == "awaiting_approval":
    # A tool that needs sign-off, such as sending an email, is waiting on the Approvals page.
    print("Waiting for approval:", result.approval_id)
elif result.status == "awaiting_input" and result.input_request:
    # The agent asked a question. Answer it with reply(), shown below.
    print("The agent asks:", result.input_request.question)

Option	Description
`agent_id`	Overrides the client's agent
`prompt_variables`	Values for `{{KEY}}` placeholders in the system prompt
`start_session` / `session_id`	Start, then continue, a conversation AgentOS remembers. The result carries `session_id`.
`history`	Earlier turns (`{"role", "content"}`, up to 50), if you keep the conversation yourself
`external_user_id`	Your end user's id, recorded on a new session
`parent_run_id`	Links the run as a child of another run
`test`	A sandboxed test run: tools that would change something are stubbed, and the run is kept out of metrics

The result is an InvokeResult with run_id, status, output, duration_ms, session_id, approval_id and input_request.

An invoke is never retried after a network error, because the agent may already be running. It is retried when the server answers 429.

`aos.invoke_stream(input=None, **options)`

Same options, but yields events as the agent works. They are typed dicts (InvokeStreamEvent), so type-checkers narrow on event["type"]:

for event in aos.invoke_stream({"ticket_id": 4521}):
    if event["type"] == "chunk":
        print(event["text"], end="", flush=True)
    elif event["type"] == "tool":
        print(f"\n[{event['status']}] {event['name']}")

Event types: start, chunk, tool, then one of done, awaiting_approval or awaiting_input. Types a newer server adds are skipped. If the run fails, the loop raises an AgentOSError.

`aos.reply(run_id, response)`

Answers a hosted run that paused with awaiting_input, and waits for it to continue. Returns an InvokeResult. Needs a workspace key. Pass request_id=result.input_request.request_id to answer that exact question; the call then fails with 409 if it was already answered.

if result.status == "awaiting_input":
    # result.input_request.question: "Refund the full amount or offer a credit?"
    resumed = aos.reply(result.run_id, "Offer a credit")
    print(resumed.status, resumed.output)

Agents that run in your own code

`aos.start_run(input=None, agent_id=None, trigger_type="sdk", external_run_id=None, parent_run_id=None)`

Starts a run. Use it as a context manager: the run is completed when the block ends, or failed with the exception's message and traceback if it raises (the exception still propagates).

with aos.start_run(input={"ticket_id": 4521}) as run:
    ticket = load_ticket(4521)  # your code
    run.emit("ticket.loaded", {"ticket_id": ticket.id, "words": len(ticket.body)})
    run.complete({"category": classify(ticket)})  # your code

Or end it yourself with complete(), fail() or cancel(). Starting again with the same external_run_id returns the same run; parent_run_id shows it as a child of another run.

`aos.flush()`

Sends the queued events of every open run. In serverless functions, call it (or end your runs) before returning. In long-running processes, queued events are also sent automatically at interpreter exit.

Run

`run.emit(type, payload=None, level="info")`

run.emit("ticket.classified", {"category": "billing", "confidence": 0.92})
run.emit("classification.uncertain", {"confidence": 0.41}, level="warn")

Built-in emit helpers

run.emit_llm_request("claude-sonnet-4-6", prompt)
run.emit_llm_response(
    "claude-sonnet-4-6", answer, usage={"input_tokens": 512, "output_tokens": 128}
)

run.emit_tool_call("lookup_order", {"order_id": 991})
run.emit_tool_return("lookup_order", {"status": "shipped"})
run.emit_tool_error("lookup_order", "Order service timed out after 5s")

`run.start_span(name, kind=None, parent=None, payload=None)`

Records timed, nested steps as a trace tree on the run page. See Tracing.

step = run.start_span("classify ticket", kind="step")

llm = step.child("claude-sonnet-4-6", kind="llm")
answer = call_model(prompt)  # your code
llm.end({"input_tokens": 1240, "output_tokens": 90})  # shown on the span

step.end()  # or step.fail(error) to record it at error level

`run.request_human_input(prompt, context=None, timeout_seconds=3600)`

Pauses and asks a person in the dashboard, then returns their reply as {"input", "request_id"}. Raises AgentOSTimeoutError if nobody replies in time.

`run.complete(output=None, usage=None)`

Sends queued events, then marks the run completed. Token usage is recorded on the run, so cost dashboards include it.

run.complete({"category": "billing"}, usage={"input_tokens": 1240, "output_tokens": 90})

`run.fail(error_message, error_detail=None)` and `run.cancel(reason=None)`

run.fail("Order service unavailable", {"attempts": 3})
run.cancel("Ticket closed by the customer")

complete, fail and cancel are safe to retry.

`run.flush()`, `run.dropped_events`, `run.is_ended`

flush() sends queued events now; it never raises. dropped_events counts events discarded because the queue was full or the server rejected them.

`run.start_task(name, input=None)`

Records a discrete step with its own status. Spans are the richer option for new code.

task = run.start_task("fetch order history", input={"customer_id": "cus_812"})
task.complete(output={"orders": 3})  # or task.fail("reason"), task.skip()

Control plane

Manage the whole workspace from code: agents, runs and their events, approvals, evals, schedules, resources, alerts, reports, members, keys. Use a workspace key (aos_ws_…), or a personal key with the workspace option.

aos = AgentOS(api_key=os.environ["AGENTOS_WORKSPACE_KEY"])

# Create an agent and run it every morning at 07:00 Lisbon time.
agent = aos.agents.create(
    name="Ticket triage",
    system_prompt="Classify each new support ticket and draft a first reply.",
    tools=["read_resource"],
)
aos.schedules.upsert(
    agent_id=agent["agent_id"], frequency="daily", hour_of_day=7, timezone="Europe/Lisbon"
)

# See what's waiting for sign-off, and why a run failed.
pending = aos.approvals.list()
timeline = aos.runs.events(run_id="11111111-1111-4111-8111-111111111111")

Every method takes keyword arguments plus workspace= and timeout_seconds=, and returns the server's JSON. Pass None to clear a field; leave an argument out to leave it alone. Reads are retried after network errors; writes only after a 429. The full list, generated from the server's schemas, is in the control-plane reference.

LangChain

pip install "pyagentos-sdk[langchain]"

from agentos.langchain import AgentOSCallbackHandler

with aos.start_run() as run:
    chain.invoke(inputs, config={"callbacks": [AgentOSCallbackHandler(run)]})

Chains, models, tools and retrievers are recorded as a span tree with real model and tool names and token counts.

Error types

Class	Code	When raised
`AgentOSConfigError`	`CONFIG_ERROR`	No API key, or no agent id where one is needed
`AgentOSAuthError`	`UNAUTHENTICATED`	Invalid or revoked API key
`AgentOSValidationError`	`VALIDATION_ERROR`	Bad request payload
`AgentOSRateLimitError`	`RATE_LIMITED`	Too many requests; `retry_after` says how long to wait
`AgentOSTransientError`	varies	A 5xx, network error or timeout that outlasted the retries
`AgentOSTimeoutError`	`TIMEOUT`	Nobody answered `request_human_input()` in time
`AgentOSError`	varies	Everything else, such as `PAYMENT_REQUIRED` (402) when the workspace has no plan

Every error has code and status (the HTTP status, or None).

Python SDK

Installation

AgentOS

Config options

Hosted agents

aos.invoke(input=None, **options)

aos.invoke_stream(input=None, **options)

aos.reply(run_id, response)

Agents that run in your own code

aos.start_run(input=None, agent_id=None, trigger_type="sdk", external_run_id=None, parent_run_id=None)

aos.flush()

Run

run.emit(type, payload=None, level="info")