AgentOS SDK for Python

Async Python SDK for AgentOS APIs. The package is a thin client over AgentOS HTTP and SSE endpoints with small compatibility helpers for bearer tokens, stream callbacks, Pythonic method names, and request defaults.

Install

pip install agentos-sdk-python

For local development:

python3 -m venv .venv
.venv/bin/python -m pip install -e ".[dev]"

Quickstart

from agentos_sdk import AgentOSSDK


async def main():
    async with AgentOSSDK(base_url="http://localhost:8888") as sdk:
        version = await sdk.agentos.get_version()
        print(version)

        registration = await sdk.agentos.register_bundle(
            {
                "bundleId": "com.demo.app",
                "appGroupId": "com.demo.group",
                "opentoolServers": [
                    {"ref": "websearch", "name": "opentool-server-websearch"}
                ],
            }
        )
        print(registration.get("opentoolServers", []))

        tasks = await sdk.cronkit.list_tasks(limit=10)
        print(tasks)

AgentOSSDK also exposes TS-style camelCase aliases for common migration paths, such as registerBundle, listTasks, chatSimple, listModels, loadTool, getPeers, and listTransfers.

Transport

The SDK uses httpx.AsyncClient and defaults to http://localhost:8888.

from agentos_sdk import AgentOSSDK

sdk = AgentOSSDK(
    base_url="http://localhost:8888",
    connect_timeout=5.0,
    receive_timeout=10.0,
)

Use async with AgentOSSDK(...) or call await sdk.aclose() when done.

Token Handling

sdk.agentos.register_bundle(...) stores the active app token. Protected facade methods in sdk.agentkit, sdk.cronkit, sdk.authkit, sdk.discoverykit, and sdk.transferkit use that token automatically.

If a protected call fails with HTTP 401, HTTP 403, or recognizable invalid-token text, the SDK re-runs registration for the same bundle once and retries the call. The cached bundle is a defensive copy and retains its opentoolServers declarations during refresh.

AgentOS Events

from agentos_sdk import AgentOSEventHandler, AgentOSSDK


class Handler(AgentOSEventHandler):
    async def on_welcome(self, welcome):
        print("welcome", welcome)

    async def on_app_event(self, event):
        if event["type"] == "toolkit.provision.succeeded":
            print("tool ready", event["data"]["ref"], event["data"]["id"])
        elif event["type"] == "toolkit.provision.failed":
            print("tool failed", event["data"]["error"])

    async def on_done(self):
        print("subscription ended")

    async def on_error(self, error):
        raise error


async with AgentOSSDK() as sdk:
    await sdk.agentos.register_bundle(
        {
            "bundleId": "com.demo.app",
            "opentoolServers": [
                {"ref": "websearch", "name": "opentool-server-websearch"}
            ],
        }
    )
    await sdk.agentos.subscribe(Handler())

Registration statuses are running, starting, and notFound; only running includes a callable tool id. A starting declaration becomes callable after toolkit.provision.succeeded. Stop using a tool while it is toolkit.tool.unavailable and resume after toolkit.tool.available.

on_call_app / onCallApp were removed in 0.2.0. Use on_app_event / onAppEvent.

AgentKit

from agentos_sdk import AgentMessageHandler, AgentOSSDK, ToolReturn


class ChatHandler(AgentMessageHandler):
    async def on_message(self, session_id, message):
        print("message", session_id, message)

    async def on_chunk(self, session_id, chunk):
        print("chunk", session_id, chunk)

    async def on_function_call(self, session_id, function_call):
        return ToolReturn(function_call["id"], {"ok": True})

    async def on_done(self):
        print("done")

    async def on_error(self, error):
        raise error


async with AgentOSSDK() as sdk:
    await sdk.agentos.register_bundle({"bundleId": "com.demo.app"})
    session = await sdk.agentkit.init_session(
        capability={"systemPrompt": "You are a helpful assistant."}
    )
    await sdk.agentkit.chat(
        session,
        {"content": [{"type": "text", "message": "Hello"}]},
        ChatHandler(),
    )

Summary-first history is available without loading every process message:

summaries = await sdk.agentkit.history_summary(session_id=session["sessionId"])
messages = await sdk.agentkit.history_process(
    session_id=session["sessionId"],
    original_task_id=summaries[0]["originalTaskId"],
)

history_summary defaults to page 1 with 30 items; history_process defaults to page 1 with 50 items. Page sizes must be between 1 and 200. CamelCase callers can use historySummary(sessionId=...) and historyProcess(sessionId=..., originalTaskId=...).

Message, chunk, and function-call handlers receive a child sessionId when the event supplies one, otherwise the root session ID. Function-call failures passed to on_error use FunctionCallLifecycleError with stage equal to function_call_parse_failed, function_call_handler_failed, or function_call_callback_failed.

AgentKit 1.0.0 follows the session-only gateway contract. Agent CRUD methods and the init_session(agent_id=...) form have been removed. Define sub-agents inline with Capability.subAgentList:

session = await sdk.agentkit.init_session(
    capability={
        "systemPrompt": "Coordinate the task.",
        "subAgentList": [
            {
                "name": "researcher",
                "capability": {"systemPrompt": "Research the requested topic."},
            }
        ],
    }
)

Direct AgentKit methods include init_session, init_simple, chat, chat_simple, history, history_summary, history_process, stop, clear, callback, and stream_callback.

CronKit

async with AgentOSSDK() as sdk:
    await sdk.agentos.register_bundle({"bundleId": "com.demo.app"})
    task = await sdk.cronkit.create_task(
        request={
            "name": "Daily summary",
            "schedule": "0 9 * * *",
            "spec": {
                "systemPrompt": "You are a scheduled agent.",
                "content": [{"type": "text", "message": "Summarize today."}],
            },
        }
    )
    await sdk.cronkit.run_task_now(cron_id=task["cronId"])

CronKit create requests default concurrency to skipIfRunning, enabled to True, and spec.timeoutSeconds to 3600.

ModelKit

async with AgentOSSDK() as sdk:
    models = await sdk.modelkit.list_models()
    chat_models = await sdk.modelkit.list_models_by_task("chat")
    print(models[0]["capabilities"]["reasoning"])

    completion = await sdk.modelkit.chat.create(
        {
            "model": "qwen3-1.7b",
            "messages": [{"role": "user", "content": "Hello"}],
            "maxTokens": 128,
        }
    )

    async for chunk in sdk.modelkit.chat.create_stream(
        {"model": "qwen3-1.7b", "messages": [{"role": "user", "content": "Stream"}]}
    ):
        print(chunk)

Normalized model capabilities include vision, tool_call, stream, and reasoning booleans. reasoning reflects only the server-provided capabilities.reasoning value and defaults to False when omitted.

ModelKit also supports:

• sdk.modelkit.embedding.create(...)
• sdk.modelkit.asr.transcribe(...)
• sdk.modelkit.tts.synthesize(...)
• sdk.modelkit.tts.list_voices()

ToolKit

async with AgentOSSDK() as sdk:
    registration = await sdk.agentos.register_bundle(
        {
            "bundleId": "com.example.app",
            "opentoolServers": [
                {
                    "ref": "websearch",
                    "name": "opentool-server-websearch",
                }
            ],
        }
    )

    tools = await sdk.toolkit.list_tool()
    websearch_tool = next(
        (
            tool
            for tool in tools
            if tool.get("status") == "running"
            and (tool.get("server") or {}).get("name")
            == "opentool-server-websearch"
        ),
        None,
    )
    if websearch_tool is None:
        raise RuntimeError("The registered websearch tool is not running")

    tool_id = websearch_tool["id"]
    definition = await sdk.toolkit.load_tool(tool_id)
    raw_json = definition.to_json() if definition else None

    result = await sdk.toolkit.call(
        tool_id,
        {
            "id": "call-1",
            "name": "websearch",
            "arguments": {
                "query": "AgentOS",
                "max_results": 5,
            },
        },
    )
    print(result)

registration["opentoolServers"] also contains the resolved id when its status is running; the example uses list_tool() so it can select the running tool by server.name.

subscribe_tool_events() is retained for legacy direct-daemon deployments. The deprecated /toolkit/events endpoint is not exposed by the AgentOS API gateway.

stream_call() is only appropriate for functions whose tool server explicitly supports streamCall. For such a function, retain the stream and close it when consumer work ends early:

async def consume_streaming_function(sdk, streaming_tool_id):
    stream = sdk.toolkit.stream_call(
        streaming_tool_id,
        {"id": "call-2", "name": "streaming-function", "arguments": {}},
    )
    try:
        async for event in stream:
            print(event["event"], event["data"])
    finally:
        await stream.aclose()

The high-level ToolKit list, load, call, and stream-call methods use the bearer token returned by register_bundle() and refresh it once after a 401/403 response. load_tool() returns an OpenToolDefinition wrapper that preserves the raw JSON definition.

AuthKit

async with AgentOSSDK() as sdk:
    await sdk.agentos.register_bundle({"bundleId": "com.demo.app"})

    auth = await sdk.authkit.authorize({"scopes": ["profile"], "appName": "Demo"})
    print(auth["requestId"], auth["status"])

    status = await sdk.authkit.get_status(request_id=auth["requestId"])
    if status.get("authCode"):
        token_resp = await sdk.authkit.exchange_token(
            request={"authCode": status["authCode"]}
        )
        user = await sdk.authkit.get_user_info(access_token=token_resp["accessToken"])
        print(user["name"])

    await sdk.authkit.revoke({"bundleId": "com.other.app"})

DiscoveryKit

async with AgentOSSDK() as sdk:
    await sdk.agentos.register_bundle({"bundleId": "com.demo.app"})

    await sdk.discoverykit.register({"receivePolicy": "autoAccept"})

    peers = await sdk.discoverykit.get_peers()
    print(peers)

    self_info = await sdk.discoverykit.get_self()
    print(self_info["deviceName"], self_info["registered"])

    async for event in sdk.discoverykit.stream_peers():
        print(event["type"], event["device"]["name"])

    await sdk.discoverykit.unregister()

TransferKit

async with AgentOSSDK() as sdk:
    await sdk.agentos.register_bundle({"bundleId": "com.demo.app"})

    session = await sdk.transferkit.send(
        request={
            "targetDeviceId": "device-abc",
            "files": [
                {"fileName": "photo.jpg", "size": 1024, "filePath": "/tmp/photo.jpg"}
            ],
        }
    )
    print(session["transferId"], session["state"])

    transfers = await sdk.transferkit.list_transfers()

    async for progress in sdk.transferkit.stream_progress(
        transfer_id=session["transferId"]
    ):
        print(progress["progress"])

    await sdk.transferkit.set_policy(request={"policy": "autoAccept"})
    policy = await sdk.transferkit.get_policy()

    history = await sdk.transferkit.get_history()
    await sdk.transferkit.clear_history()

list_tool() is typed as list[ToolDto] and exposes daemon 0.5 metadata. Its optional server identity uses namespace, name, and version; materializedArtifactDigest identifies installed content when available, and contentState reports whether it is current.

Direct Clients

The low-level clients are available for explicit-token or custom transport usage:

from agentos_sdk.clients import AgentKitClient, CronKitClient, ModelKitClient
from agentos_sdk.transport import ApiClient

api = ApiClient(base_url="http://localhost:8888")
agentkit = AgentKitClient(api)
session = await agentkit.init_session(
    "bearer-token",
    capability={"systemPrompt": "You are a helpful assistant."},
)

Development

.venv/bin/python -m compileall agentos_sdk
.venv/bin/python -m pytest -q
.venv/bin/ruff check .