pyagentos-sdk 0.2.1

AgentOS Python SDK — instrument agents with runs, events, and tasks

pyagentos-sdk

Official Python SDK for AgentOS — instrument your AI agents with runs, events, and tasks.

The PyPI package is pyagentos-sdk (the agentos-* namespace was already claimed on PyPI by an unrelated project). The Python import name stays agentos for clean code.

The SDK is a thin client over the AgentOS API. Your agent code runs wherever you want; AgentOS observes it.

Install

pip install pyagentos-sdk

For LangChain integration:

pip install "pyagentos-sdk[langchain]"

Quick start

Generate an API key from your agent's page in the AgentOS dashboard, then pick a path:

Calling a hosted agent

For agents you've defined in AgentOS — invoke() runs the agent server-side and returns its output:

import os
from agentos import AgentOS

aos = AgentOS(
    api_key=os.environ["AGENTOS_API_KEY"],
    agent_id=os.environ["AGENTOS_AGENT_ID"],
)

result = aos.invoke({"prompt": "Hello"})
print(result["output"])

Instrumenting your own agent

For agents that run in your own code — start_run / emit / complete reports them to AgentOS for observability:

import os
from agentos import AgentOS

aos = AgentOS(
    api_key=os.environ["AGENTOS_API_KEY"],
    agent_id=os.environ["AGENTOS_AGENT_ID"],
)

run = aos.start_run(input={"prompt": "Hello"})

try:
    run.emit("step.completed", {"step": 1})

    # ... do work ...

    run.complete(output={"result": "Done"})
except Exception as e:
    run.fail(error_message=str(e))
    raise

emit(type, payload, level) is non-blocking — events are buffered and flushed on a timer (default 500 ms) or when the buffer fills (50 events). Call run.flush() if you need to be sure pending events are sent before the process exits. run.complete() and run.fail() flush automatically.

LangChain integration

Drop in AgentOSCallbackHandler to mirror LangChain runs into AgentOS automatically:

from agentos import AgentOS
from agentos.langchain import AgentOSCallbackHandler
from langchain_anthropic import ChatAnthropic

aos = AgentOS(api_key=..., agent_id="...")
run = aos.start_run()

model = ChatAnthropic(callbacks=[AgentOSCallbackHandler(run)])
model.invoke("What's the weather in SF?")

run.complete()

langchain-core is an optional dependency — only install it if you use the LangChain handler (pip install "pyagentos-sdk[langchain]").

Configuration

AgentOS(
    api_key="aos_...",
    agent_id="...",
    base_url="https://agentos-ai.dev",  # optional — production default; override for self-hosted / local dev
    batch_interval_ms=500,              # flush events every 500 ms
    batch_size=50,                      # max events per batch
    disabled=False,                     # set True in tests for a no-op client
    timeout_seconds=10.0,               # httpx request timeout
)

Errors

from agentos import (
    AgentOSError,            # base class
    AgentOSAuthError,        # 401 — invalid/revoked key
    AgentOSValidationError,  # 400 — server-side validation failed
    AgentOSRateLimitError,   # 429 — ingest rate limit exceeded
)

Requirements

• Python 3.10+
• An AgentOS workspace with an agent + API key (sign up)

License

Apache-2.0