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, as a keyword argument:

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

Or call aos.shutdown() (alias aos.close()) yourself: it flushes queued events, then closes the HTTP connection pool.

---

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, a question, or a child run.

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)
elif result.status == "awaiting_child":
    # A script it called as a tool paused; the run continues on its own once the child ends.
    print("Waiting on child runs:", result.child_run_ids)

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
timeout_seconds	Client timeout. Default 310.0, just over the server's 300 s limit.

The result is an InvokeResult with run_id, status, output, duration_ms, session_id, approval_id, input_request and child_run_ids. input_request is an InputRequest with request_id, question, context and options (the last two may be None). child_run_ids lists the child runs that paused when status is awaiting_child.

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']}")
    elif event["type"] == "done":
        print(f"\nFinished in {event.get('duration_ms')} ms")

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

done carries no output, and an agent with an output schema sends no chunk events. Use invoke() when you need the agent's output.

aos.reply(run_id, response, *, request_id=None, timeout_seconds=310.0)

Answers a hosted run that paused with awaiting_input, and waits for it to continue. Returns an InvokeResult. Needs a workspace key.

Without request_id it answers whatever question the run has pending. 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. All arguments are keyword-only. 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. aos.shutdown() also flushes, then closes the client.

---

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

emit_llm_request(model, prompt, options=None) adds the keys of options to the event payload. emit_llm_response takes the same token usage as run.complete().

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

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

Option	Description
kind	"agent", "step", "llm", "tool" or "retrieval"
parent	A span to nest under. span.child(name, *, kind=None, payload=None) does the same.
payload	Fields recorded on the span. Fields passed to end() are merged over them.

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, payload=None) to record it at error level with the error's message

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"}. context is stored with the request; a context["options"] list of strings shows as reply buttons. Waits up to timeout_seconds (from 30 to 86400), reconnecting as needed.

reply = run.request_human_input(
    "Refund the full amount or offer a credit?",
    context={"options": ["Full refund", "Offer a credit"]},
    timeout_seconds=1800,
)
# reply["input"]: the reply someone typed in the dashboard, e.g. "Offer a credit"

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})

usage also takes cache_read_tokens and cache_creation_tokens. Cost is estimated from token usage, priced by the agent's configured model; an agent with no model set is priced at a default rate.

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. The namespaces are agents, alerts, api_keys, approvals, conversations, evals, fleet, folders, integrations, marketplace, members, notifications, reports, resources, runs, schedules, tools and workspaces. 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 or network error that outlasted the retries, or REQUEST_TIMEOUT when no response arrived within the client timeout
AgentOSTimeoutError	TIMEOUT	Nobody answered request_human_input() in time. status is always 408.
AgentOSError	varies	Everything else, such as NO_PLAN (402) when the workspace has no plan

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