SDKs

Python

$ pip install -e sdk/python

The client

AgentOS is a synchronous client built on httpx. Point it at a base URL and authenticate with an API key or a session token; it appends /api for you and attaches the right header on every request. It is also a context manager, so it closes its connection pool cleanly.

from agentos_sdk import AgentOS, AgentOSError

# api_key -> X-API-Key ; token -> Authorization: Bearer
client = AgentOS("http://localhost:8000", api_key="aos_...")

# or as a context manager
with AgentOS("http://localhost:8000", token="sess_...") as client:
    workflows = client.list_workflows()

Any non-2xx response raises AgentOSError(status, detail) — the detail is lifted from the API's { "detail": ... } envelope when present. A 204 or empty body returns None.

Running workflows

The everyday path. run() takes a workflow id or its exact name (it tries the id, then falls back to a name lookup) and either returns the freshly-created run immediately or, with wait=True, polls until the run settles:

from agentos_sdk import AgentOS

client = AgentOS("http://localhost:8000", api_key="aos_...")

run = client.run(
    "Support Triage Crew",
    {"ticket": "I was double charged"},
    wait=True,                      # poll until success / failed / paused
    timeout=120.0,                  # give up after N seconds -> TimeoutError
    poll_interval=0.5,              # seconds between polls
    idempotency_key="ticket-8841",  # safe to retry: repeats return the same run
)

if run["status"] == "paused":       # stopped at a human gate
    client.approve(run["id"], input={"reviewer": "lukas"})
    run = client.get_run(run["id"])

print(run["status"], run["output"])

• wait=True polls with get_run until the run reaches success, failed, or paused and returns it complete with steps; without it you get the run object right after the trigger and poll yourself.
• If wait=True exceeds timeout the client raises a plain TimeoutError (the run keeps going server-side).
• idempotency_key is sent as the Idempotency-Key header — a repeated key returns the original run instead of starting a duplicate, so triggers are safe to retry.

Full method reference

Every method is a one-line wrapper over a REST call. Column three is the endpoint each one hits.

register_tool carries the full registration in keyword arguments: kind ("http" default or "mcp"), label, description, method (default "POST"), headers, mcp_tool (the target tool name for MCP kinds), input_schema, timeout_seconds (default 60), and retry_count (default 1).

Connect your agent in ten lines

The @tool decorator plus serve() turn plain Python functions into first-class workspace tools without you writing any HTTP glue:

from agentos_sdk import AgentOS, tool, serve

@tool(description="Score a sales lead 0-100.")
def score_lead(company: str, employees: int = 0) -> dict:
    return {"score": min(100, employees // 10)}

client = AgentOS("http://localhost:8000", api_key="aos_...")
serve(client, [score_lead], port=8801)
# prints: tool custom:score-lead -> http://localhost:8801/tools/score-lead

• @tool works bare or with arguments (name, description, input_schema). The tool name defaults to the function name with underscores turned to hyphens (score_lead becomes score-lead), the description defaults to the docstring, and the input schema is inferred from the signature — parameter annotations map to JSON types and parameters without defaults become required.
• serve() stands up a threaded HTTP server that exposes each function at /tools/<name>and registers it with the workspace (deleting any prior tool of the same name first), so it appears in the builder palette immediately. Every invocation arrives as a POST whose JSON body becomes the function's keyword arguments; a non-dict return is wrapped as { "result": ... }.
• Bad or missing arguments return 400; an exception inside the function returns 500 and fails the node, which then retries per its retry_count. Pass block=False to run the server on a background thread, or client=None to serve without registering.

Build graphs in code

GraphBuilder is a fluent, contract-shaped way to assemble a workflow graph without hand-writing node and edge JSON. Each node method returns a NodeRef you pass to edge(), and build() emits the { nodes, edges } object the API expects.

from agentos_sdk import AgentOS, GraphBuilder

g = GraphBuilder()

classify = g.agent("classify",
                   system_prompt="Classify the ticket as billing/technical/other.",
                   model="claude-haiku-4-5", output_key="category")

route = g.condition("route", expression="category == 'billing'")
refund = g.tool("refund", tool_name="custom:score-lead",
                params={"company": "{category}"})
other = g.tool("other", tool_name="slack_message", params={"text": "Non-billing ticket"})

review = g.gate("review", prompt="Approve the drafted reply?")
done = g.merge("done")

g.edge(classify, route)
g.edge(route, refund, condition_value="true")   # billing branch
g.edge(route, other, condition_value="false")   # everything else
g.edge(refund, review)
g.edge(review, done)
g.edge(other, done)

client = AgentOS("http://localhost:8000", api_key="aos_...")
wf = client.create_workflow("Triage", g.build())
assert client.validate(wf["id"])["valid"]

Node positions are auto-laid-out if you do not pass x/y, duplicate ids raise immediately, and build() checks that every edge endpoint refers to a real node before returning — so a graph that builds is at least internally consistent (run client.validate() for the full structural check).

LangGraph and CrewAI adapters

from agentos_sdk import AgentOS
from agentos_sdk.adapters import register_langgraph, register_crewai

app = my_state_graph.compile()   # anything with .invoke(state) -> dict

client = AgentOS("http://localhost:8000", api_key="aos_...")
register_langgraph(client, "researcher", app, port=8802)
# workflows can now call custom:researcher

# for CrewAI, anything exposing .kickoff(inputs=...)
register_crewai(client, "planner", my_crew, port=8803)
# workflows can now call custom:planner

The adapters are deliberately thin and duck-typed — neither langgraph nor crewai is imported. register_langgraph expects a compiled app with an .invoke(state) method; register_crewai expects a crew with .kickoff(inputs=...). Each wraps that call in a @tool and hands it to serve(), so your agent keeps running in your process with its own framework while Ballast calls it over HTTP as a registered tool and wraps every invocation with checkpoints, retries, gates, and audit. No graph translation happens, by design. Both accept the same public_url, host, port, and block keywords as serve().

TypeScript

$ npm install ./sdk/typescript

The TypeScript client mirrors the Python one method-for-method (in camelCase) and is dependency-free — it uses the built-in fetch, so it needs Node 18+ or any modern runtime. Construction takes an options object rather than positional arguments:

import { AgentOS, AgentOSError } from "@agentos/sdk";

const client = new AgentOS({
  baseUrl: "http://localhost:8000",
  apiKey: "aos_...",           // or token: "sess_..."
});

const run = await client.run(
  "Support Triage Crew",
  { ticket: "Refund?" },
  { wait: true, timeoutMs: 120_000, pollMs: 500, idempotencyKey: "ticket-8841" },
);

if (run.status === "paused") {
  await client.approve(run.id, true, { reviewer: "lukas" });
}

await client.registerTool({
  name: "score-lead",
  endpoint_url: "https://my-agent.example.com/score",
  description: "Score a sales lead 0-100.",
});

const webhookUrl = await client.enableWebhook(run.workflow_id);

• run() resolves ids or exact names and waits on request, exactly like Python — but on timeout it throws AgentOSError(408, ...) rather than a native timeout type.
• registerTool() takes a single RegisterToolInput object and fills sensible defaults (kind: "http", method: "POST", empty headers, timeout_seconds: 60, retry_count: 1) for whatever you omit.
• The test_tool helper and the graph builder/serve/adapter stack are Python-only; from TypeScript, call the tool-test endpoint directly if you need it.

The package also exports the full type surface for building typed integrations: Workflow, Graph, WorkflowNode, WorkflowEdge, Run, RunStatus, Step, Check, CheckType, CheckResult, EvalSummary, CustomTool, RegisterToolInput, ValidationResult, WebhookInfo, AgentOSOptions, and AgentOSError.

Evaluation checks

Both SDKs added set_checks / setChecks for attaching automatic evaluations to a workflow. Each successful run is scored against the checks and the results ride back on the run under evals (a { passed, total, results } summary), so you can assert quality in CI or gate a deploy on a pass rate. Both accept a workflow id or exact name.

client.set_checks("Support Triage Crew", [
    {"name": "mentions refund", "type": "contains", "target": "draft_reply", "value": "refund"},
    {"name": "under budget",    "type": "max_cost_usd", "value": 0.05},
    {"name": "fast enough",     "type": "max_duration_ms", "value": 8000},
])

run = client.run("Support Triage Crew", {"ticket": "..."}, wait=True)
print(run["evals"])   # { "passed": 3, "total": 3, "results": [...] }

Check type is one of contains, not_contains, equals, regex, max_cost_usd, max_duration_ms, or expression. Text checks take a target (a state key) and a value, with an optional case_sensitive flag; the cost and duration checks take just a numeric value.

Authentication

• API keys (recommended for programs) — create one in Settings → API keys; its role (or the linked member's permissions) bounds what the SDK can do. Pass it as api_key= / apiKey.
• Session tokens — pass token= / token if you already hold a login session.
• In AUTH_MODE=open local development, neither is required — the SDK acts as the workspace owner.