AgentOS Python SDK

The official Python client for AgentOS, the operating system for agent-run companies: one place to build, run, supervise and improve the AI agents that do work for your business.

Documentation · Control-plane reference · HTTP API · Release history · TypeScript SDK

What it does

• Run hosted agents. Invoke agents that AgentOS runs for you, stream their output as they work, keep conversations going, and answer an agent that stops to ask a question.
• Report agents that run in your own code. Send runs, events, trace spans and token usage from any agent, so it gets the same run history, alerts, reports and cost tracking as a hosted one.
• Manage the platform. Create and update agents, read runs and their events, decide approvals, run evals, and manage schedules, resources, alerts, keys and members through a typed control plane.

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.

Contents

• Install
• Quick start
• Run hosted agents
• Report your own agents
• Manage the platform
• LangChain
• Configuration
• Keys and permissions
• Reliability
• Errors

Install

pip install pyagentos-sdk
# or: uv add pyagentos-sdk · poetry add pyagentos-sdk

The distribution is pyagentos-sdk; the module is agentos. Python 3.10 or later. The only dependency is httpx. The package ships type information (PEP 561), so mypy and Pyright check your calls.

Quick start

Create a key on your agent's Keys tab in the AgentOS dashboard, then set it and the agent's id in your environment:

export AGENTOS_API_KEY=aos_agent_...
export AGENTOS_AGENT_ID=00000000-0000-0000-0000-000000000000

from agentos import AgentOS

aos = AgentOS()  # reads AGENTOS_API_KEY and AGENTOS_AGENT_ID

result = aos.invoke({"task": "Summarise yesterday's support tickets"})
print(result.status, result.output)

Run hosted agents

invoke() runs a hosted agent and waits for it, up to the server's 300-second limit. A run can finish, or pause for a person, so always check status:

result = aos.invoke({"ticket_id": 4521}, prompt_variables={"TONE": "formal"})

if result.status == "completed":
    print(result.output)
elif result.status == "awaiting_approval":  # a gated tool call is waiting on the Approvals page
    print("Needs approval:", result.approval_id)
elif result.status == "awaiting_input":  # the agent asked a question; answer it with reply()
    aos.reply(result.run_id, "Refund the full amount")

Stream the agent's text and tool calls as they happen:

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

Keep a conversation going. AgentOS stores the history:

first = aos.invoke(
    {"message": "I need to change my plan"}, start_session=True, external_user_id="user_812"
)
following = aos.invoke({"message": "The annual one"}, session_id=first.session_id)

Other options: history (manage the turns yourself), parent_run_id (show the run as a child of another), test (a sandboxed run whose tools don't change anything), and agent_id (target another agent with a workspace key). See Invoking hosted agents, Memory and Prompt variables.

Report your own agents

For agents that run in your code, use a run as a context manager. It completes when the block ends, and fails, with the traceback, if the block raises:

with aos.start_run(input={"ticket_id": 4521}) as run:
    step = run.start_span("classify ticket", kind="step")

    llm = step.child("claude-sonnet-4-6", kind="llm")
    answer = call_model(prompt)
    llm.end({"input_tokens": 1240, "output_tokens": 90})

    run.emit_tool_call("lookup_order", {"order_id": 991})
    step.end()
    run.complete({"category": "billing"}, usage={"input_tokens": 1240, "output_tokens": 90})

Spans become a trace tree on the run page (Tracing). You can also emit custom events at debug, info, warn or error level (Events). Token usage on complete() feeds the cost dashboards.

Prefer to end the run yourself? Call complete(), fail() or cancel(). Starting again with the same external_run_id returns the same run, so retrying a job never creates duplicates:

run = aos.start_run(input={"batch": "2026-09-22"}, external_run_id="nightly-2026-09-22")
try:
    ...  # do the work
    run.complete({"rows": 1200})
except Exception as exc:
    run.fail(str(exc))
    raise

Need a person to weigh in mid-run? run.request_human_input("Send the refund?") pauses, shows the question in the dashboard, and returns the answer.

Manage the platform

With a workspace key, the control plane gives you everything the dashboard can do, as typed methods with keyword arguments:

import os

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

created = admin.agents.create(
    name="Invoice checker",
    system_prompt="Check each invoice against its purchase order and flag mismatches.",
    tools=["read_resource", "http_request"],
)
admin.schedules.upsert(
    agent_id=created["agent_id"], frequency="daily", hour_of_day=7, timezone="Europe/Lisbon"
)

for failed in admin.runs.list(agent_id=created["agent_id"], status="failed"):
    events = admin.runs.events(
        run_id=failed["id"]
    )  # the full timeline, tool calls and errors included

Namespace	Methods
agents	list get create update set_status delete
runs	list get events invoke cancel
approvals	list decide
evals	list_cases create_case delete_case run_case get_run
schedules	get upsert delete
resources	list create delete
alerts	list create update delete
reports	list get generate delete
tools	list_builtin list_workspace list_composio create delete
folders · conversations · members · api_keys · notifications · integrations · marketplace · fleet · workspaces	see the reference

Methods return the server's JSON. Leave an argument out to leave a field alone; pass None to clear it. The control-plane reference lists every method and parameter. It's generated from the server's schemas, like the methods themselves, so the two can't disagree.

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 the real model and tool names and token counts. Already emitting OpenTelemetry? AgentOS ingests OTLP traces directly.

Configuration

Argument	Environment	Default
api_key	AGENTOS_API_KEY	required	Agent, workspace or personal key
agent_id	AGENTOS_AGENT_ID	none	Default agent for runs and invokes
workspace	AGENTOS_WORKSPACE	none	Workspace slug or id for control-plane calls with a personal key
base_url	AGENTOS_BASE_URL	https://agentos-ai.dev	For self-hosted AgentOS
disabled	AGENTOS_DISABLED	False	No network, no side effects: for tests
batch_interval_ms		500	ms between event flushes
batch_size		50	Events per request
max_queue_size		1000	Events held per run before debug/info ones are dropped
max_retries		5	Retries for transient failures
timeout_seconds		30.0	Per-request timeout for reporting calls
on_error		logs once	Receives background delivery failures
http_client		new httpx.Client	Custom client, e.g. behind a proxy

Use the client itself as a context manager (with AgentOS() as aos:) to flush queued events and close its connections at the end.

Keys and permissions

Key	Create it on	Can
Agent key aos_agent_…	the agent's Keys tab	Report runs for, and invoke, that one agent
Workspace key aos_ws_…	Backoffice → API Keys (owners and admins)	Report and invoke for any agent, answer paused runs, and use the control plane
Personal key aos_user_…	Backoffice → API Keys	Use the control plane with your own role; set workspace

Control-plane writes follow the key holder's role, exactly as in the dashboard. See Authentication & keys.

Reliability

• Never blocks, never raises. emit() only queues. Events are sent in batches from a background thread, and a delivery failure goes to on_error, not into your code.
• Retries that can't double-apply. Reporting calls and control-plane reads retry network errors, 5xx and 429 with exponential backoff (0.25 s to 4 s, honouring Retry-After). invoke(), reply() and control-plane writes retry only a 429, because repeating them could run an agent twice.
• Bounded memory. Each run holds at most max_queue_size events; when it's full, the oldest debug and info events go first, so warnings and errors survive. run.dropped_events counts anything dropped.
• Nothing left behind. Queued events are sent when the interpreter exits. On serverless platforms, which freeze rather than exit, call aos.flush() before returning.

Errors

Every error extends AgentOSError and carries a stable code and the HTTP status:

Error	When
AgentOSConfigError	No API key, or no agent id where one is needed
AgentOSAuthError	The key is invalid or revoked
AgentOSValidationError	The request was rejected as invalid
AgentOSRateLimitError	Rate limited after retries; retry_after says how long to wait
AgentOSTransientError	A 5xx, network error or timeout outlasted the retries
AgentOSTimeoutError	Nobody answered request_human_input() in time

import time

from agentos import AgentOSRateLimitError

try:
    aos.invoke({"task": "..."})
except AgentOSRateLimitError as exc:
    time.sleep(exc.retry_after or 5)

Versioning

The SDK follows semantic versioning. Until 1.0, a minor release can include breaking changes; each one is listed in CHANGELOG.md in the source distribution and summarised in the release history. The Python and TypeScript SDKs are released together with the same version.

Support

Questions or problems: agentos-ai.dev/contact or bernabranco@agentos-ai.dev.

License

MIT