Python SDK

Python SDK

The Python SDK (packages/sdk/) instruments your existing agent code and sends lifecycle events to the agent-os API. It supports LangGraph and CrewAI out of the box.

Installation

pip install agent-os-sdk

With framework-specific extras:

pip install "agent-os-sdk[langgraph]"
pip install "agent-os-sdk[crewai]"

Quick Start

from agent_os_sdk import AgentOS

aos = AgentOS(api_key="your-api-key")
graph = aos.instrument(compiled_graph, agent_name="my-agent")
result = graph.invoke({"messages": ["Hello!"]})

AgentOS Client

Constructor

AgentOS(
    api_key: str,           # Required – your agent-os API key
    base_url: str = "https://app.agent-os.dev",
    timeout: float = 10.0,
)

instrument(graph, agent_name)

Wraps a compiled LangGraph StateGraph with automatic event sending.

instrumented = aos.instrument(compiled_graph, agent_name="my-agent")
result = instrumented.invoke({"messages": [...]})

Events sent automatically:

• run_start — when .invoke() or .ainvoke() is called
• run_end — when execution completes successfully
• error — when an exception occurs (exception is re-raised)

send_event(event_type, sdk_run_id, payload, agent_name=None)

Send a custom event manually:

from agent_os_sdk.events import EventType

aos.send_event(
    EventType.STEP,
    sdk_run_id="run-123",
    payload={"node": "classifier", "phase": "start"},
)

Context Manager

with AgentOS(api_key="...") as aos:
    graph = aos.instrument(compiled_graph, agent_name="my-agent")
    result = graph.invoke(input)
# HTTP client is closed automatically

Callback Handler

For per-node and per-tool events, attach the callback handler:

from agent_os_sdk.callbacks import AgentOSCallbackHandler

handler = AgentOSCallbackHandler(aos, sdk_run_id="run-123")
config = {"callbacks": [handler]}
graph.invoke(input, config=config)

This sends step events (with duration) and tool_call events automatically for every node and tool execution.

LangGraph Integration

Step-by-step

from typing import TypedDict
from langgraph.graph import StateGraph, END
from agent_os_sdk import AgentOS

# 1. Define your state and graph
class State(TypedDict):
    query: str
    result: str

def search(state: State) -> State:
    return {"result": f"Answer to: {state['query']}"}

builder = StateGraph(State)
builder.add_node("search", search)
builder.set_entry_point("search")
builder.add_edge("search", END)
graph = builder.compile()

# 2. Instrument with agent-os
aos = AgentOS(api_key="your-api-key")
graph = aos.instrument(graph, agent_name="search-agent")

# 3. Run as usual
result = graph.invoke({"query": "What is the capital of France?"})

Async support

The SDK supports both synchronous and asynchronous LangGraph execution:

result = await graph.ainvoke({"query": "..."})

Multi-agent graphs

For graphs that call sub-graphs, instrument each compiled graph independently:

sub_graph = aos.instrument(sub_graph, agent_name="sub-agent")
main_graph = aos.instrument(main_graph, agent_name="main-agent")

CrewAI Integration

The SDK includes a CrewAI adapter at agent_os_sdk.crewai:

from agent_os_sdk import AgentOS
from agent_os_sdk.crewai import instrument_crew

aos = AgentOS(api_key="your-api-key")
instrumented_crew = instrument_crew(aos, crew, agent_name="my-crew")
result = instrumented_crew.kickoff()

Environment Variables

export AGENT_OS_API_KEY="your-api-key"

import os
from agent_os_sdk import AgentOS

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

Event Types

Event Type	Value	Description
RUN_START	run_start	Agent execution begins
RUN_END	run_end	Agent completed successfully
STEP	step	Node/step execution
TOOL_CALL	tool_call	Tool invocation
ERROR	error	Exception occurred
HUMAN_INPUT_REQUESTED	human_input_requested	Agent waiting for human input
HUMAN_INPUT_RECEIVED	human_input_received	Human input received

Reliability

• Retries: Failed requests are retried up to 2 times with backoff (0.5s, 1.0s)
• Non-blocking: SDK failures never crash your agent — errors are logged and swallowed
• Thread-safe: Internal lock for sync operations
• Async-safe: Lazy-initialized async HTTP client
• Lightweight: Only httpx as external dependency