Python SDK

Package: pyagentos-sdk (import as agentos)

Installation

pip install pyagentos-sdk

AgentOS

from agentos import AgentOS

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

Config options

Option	Type	Default	Description
api_key	str	required	Agent key (aos_agent_…)
agent_id	str	required	Agent ID from the dashboard
base_url	str	https://agentos-ai.dev	Override for self-hosted instances
batch_interval_ms	int	500	ms between event flushes
batch_size	int	50	Max events per batch
disabled	bool	False	No-op mode for tests
timeout_seconds	float	10.0	HTTP request timeout

Use as a context manager to ensure the HTTP client is closed:

with AgentOS(api_key="...", agent_id="...") as client:
    run = client.start_run()
    await run.complete()

client.start_run(input?, trigger_type?, external_run_id?)

run = client.start_run(
    input={"invoice_id": "INV-001"},
    trigger_type="sdk",           # "manual" | "scheduled" | "webhook" | "sdk"
    external_run_id="my-job-123", # optional correlation ID
)

client.invoke(input?, agent_id?)

Call a hosted agent and get its output synchronously.

result = client.invoke({"query": "Summarise the Q1 report"})
print(result["output"]["summary"])

---

Run

run.emit(type, payload?, level?)

run.emit("step.completed", {"step": "fetch-data", "rows": 42})
run.emit("warning.detected", {"message": "Low confidence"}, level="warn")

Built-in emit helpers

run.emit_llm_request("claude-sonnet-4-6", "Summarise this...")
run.emit_llm_response("claude-sonnet-4-6", "The document...", token_usage={"input_tokens": 512, "output_tokens": 128})
run.emit_tool_call("search_web", {"query": "AI news"})
run.emit_tool_return("search_web", {"results": [...]})
run.emit_tool_error("search_web", "Timeout after 5s")

run.start_task(name, input?)

task = run.start_task("parse-pdf", input={"file": "report.pdf"})
pages = parse_pdf("report.pdf")
task.complete(output={"pages": len(pages)})

run.complete(output?)

run.complete(output={"summary": "..."})

run.fail(error_message, error_detail?)

run.fail(
    error_message="API timeout after 3 retries",
    error_detail={"endpoint": "/payments", "attempts": 3},
)

run.flush()

Force-flush the event buffer. Called automatically by complete() and fail().

---

Task

task = run.start_task("validate", input={"rows": 100})

task.complete(output={"valid": 98, "invalid": 2})
task.fail(error_message="Schema mismatch on row 14")
task.skip()

---

Full example

from agentos import AgentOS

with AgentOS(api_key="aos_agent_...", agent_id="...") as client:
    run = client.start_run(input={"date": "2026-05-11"})

    fetch = run.start_task("fetch-data")
    data = fetch_daily_data()
    fetch.complete(output={"rows": len(data)})

    run.emit_llm_request("claude-sonnet-4-6", "Analyse this data...")
    summary = call_llm(data)
    run.emit_llm_response("claude-sonnet-4-6", summary)

    run.complete(output={"summary": summary})

---

Error types

Class	When raised
AgentOSAuthError	Invalid or revoked API key
AgentOSValidationError	Bad request payload
AgentOSRateLimitError	Too many requests
AgentOSError	All other errors