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, sdk.transferkit, sdk.harnesskit, and sdk.notifykit 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()
• sdk.modelkit.decision.create(...): typed-question decisions, see below
• sdk.modelkit.list_app_models() and the app-model registration methods, see below

Decision

Decision models (task "decision", served by llama.cpp /v1/systemone) do not generate text. They score the options of typed questions about a given state. POST /modelkit/decision needs no app token.

async with AgentOSSDK() as sdk:
    decision_models = await sdk.modelkit.list_models_by_task("decision")

    response = await sdk.modelkit.decision.create(
        {
            "model": decision_models[0]["name"],  # optional, the gateway picks a default
            "state": {"text": "My order is late and I am furious!"},
            "questions": {
                # Pick one option (option -> description).
                "route": {
                    "type": "choice",
                    "instructions": "Which team should handle this?",
                    "criteria": {
                        "billing": "Payment and invoices",
                        "support": "Product or delivery problems",
                    },
                },
                # Ordinal scale with 2-10 level descriptions.
                "urgency": {
                    "type": "score",
                    "instructions": "How urgent is this?",
                    "criteria": ["low", "medium", "high"],
                },
                # Yes/no judgement. "criteria" ({"true": ..., "false": ...}) is optional.
                "angry": {"type": "noul", "instructions": "Is the user angry?"},
            },
            # "images": ["data:image/png;base64,..."],  # for models with image input
        }
    )

    route = response["answers"]["route"]
    print(route["choice"], route["confidence"])
    print(response["answers"]["urgency"]["score"], response["answers"]["angry"]["noul"])
    print(response["usage"]["input_tokens"], response["usage"]["output_tokens"])

The response is returned as sent by the server (wire field names). Each answer carries the field matching its question type (choice + confidence, score, or noul); probabilities and legend are passed through untouched. agentos_sdk exports DecisionRequest, DecisionQuestion, DecisionResponse and related TypedDict definitions for annotations.

App models

Apps can register their own models into the gateway. Registered models show up in list_models() / list_models_by_task() with isAppModel: True and the owning appBundleId.

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

    # Register (or replace) a model. Only "id" and "name" are required.
    result = await sdk.modelkit.register_app_model(
        request={
            "id": "my-router",
            "name": "my-router",
            "task": "decision",
            "baseUrl": "http://127.0.0.1:9000",
            "apiKey": "upstream-secret",  # stored by the server, never returned
        }
    )
    print(result.get("updated"))  # True when an existing model of this app was replaced

    # Partial update: unset fields stay unchanged.
    await sdk.modelkit.update_app_model(
        model_id="my-router", request={"description": "Routes support tickets"}
    )

    # Enable/disable without deleting.
    await sdk.modelkit.toggle_app_model(model_id="my-router")

    # List every app-registered model (public, no token needed).
    for model in await sdk.modelkit.list_app_models():
        print(model["id"], model["appBundleId"], model["isDisabled"])

    await sdk.modelkit.unregister_app_model(model_id="my-router")

Write calls attach the app token automatically and refresh it once on 401. They raise ApiError carrying the server status: 401 invalid token, 403 not the owning app or appBundleId mismatch, 404 unknown id, 409 the id belongs to another app, 400 invalid body or unknown task.

Decision models are normalized by the server: input text (+ optional image), output decisions, and stream / tool_call / reasoning all False. The low-level ModelKitClient (sdk.modelKit) offers the same app-model methods with the bearer token as the first argument, for example sdk.modelKit.register_app_model(token, request). camelCase aliases such as registerAppModel and listAppModels are available on both.

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.

HarnessKit

HarnessKit forwards requests through the Gateway to the Harness runtime selected for the currently registered AppKit bundle. Register the app with sdk.agentos.register_bundle() first. An installable process App must declare harnessKit to receive a Harness runtime; the built-in Harness App uses the host's default runtime. All sdk.harnesskit methods use the active AppKit token and retry once after an authentication failure.

Method	Parameters	Description
health()	none	Checks Harness runtime health through the authenticated Gateway route.
get_runtime() / switch_engine(engine=...)	engine	Reads the runtime engine or requests an engine switch.
list_threads() / get_thread(thread_id=...)	thread_id	Lists durable threads or reads one thread snapshot.
rename_thread(thread_id=..., title=...) / delete_thread(thread_id=...)	thread_id, title	Updates a thread title or deletes an idle thread.
start_run(thread_id=..., request=...)	thread_id, run request	Starts a Harness run and returns its runId.
submit_command(thread_id=..., command=...)	thread_id, command	Submits work or control commands; rejected commands return their receipt.
events(thread_id=..., after_sequence=0, run_id=None)	thread_id, after_sequence, run_id	Opens a resumable SSE stream of typed events.
get_run_result(run_id=...)	run_id	Reads the current or terminal run result.
get_operational_snapshot()	none	Reads an operational snapshot when the runtime supports it.
request(path, ...)	Harness-relative path, method, query, data, headers	Calls other JSON Harness endpoints through /harnesskit/*.
request_stream(path, ...)	same as request	Opens a raw streaming response for other Harness endpoints.

import uuid

from agentos_sdk import AgentOSSDK


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

        runtime = await sdk.harnesskit.get_runtime()
        print("Harness engine:", runtime.get("activeEngine", runtime["engine"]))

        accepted = await sdk.harnesskit.start_run(
            thread_id="research-thread",
            request={
                "requestId": str(uuid.uuid4()),
                "goal": "Summarize the latest project status",
                "llmConfig": {
                    "provider": "openai-compatible",
                    "baseUrl": "http://127.0.0.1:1234/v1",
                    "model": "local-model",
                },
            },
        )

        async for event in sdk.harnesskit.events(
            thread_id="research-thread",
            run_id=accepted["runId"],
            after_sequence=0,
        ):
            print(event["type"], event["data"])

events() returns an async generator and opens the connection on first iteration. Pass the sequence of the last event you processed as after_sequence to resume a stream. Frames that are not JSON objects raise ValueError. To stop early, call aclose() on the generator so the connection is released:

events = sdk.harnesskit.events(thread_id="research-thread")
try:
    async for event in events:
        if event["type"] == "run.completed":
            break
finally:
    await events.aclose()

Request and response bodies are plain dicts, so newer Harness fields pass through untouched; the exported HarnessKit* TypedDicts (for example HarnessKitRunRequest and HarnessKitCommand) are only annotations. Notes:

• A command the Harness rejects comes back as HTTP 400 with a receipt body. submit_command() returns that receipt (status == "rejected") instead of raising; every other error raises ApiError.
• Runtime controls (get_runtime(), switch_engine()) and get_operational_snapshot() are reserved for the built-in Harness App; other Apps get an ApiError with status 403. get_operational_snapshot() also answers 501 when the configured engine does not expose the optional reader.
• Thread and run IDs are percent-encoded. Empty IDs and . / .. raise ValueError before any request is sent.
• request() and request_stream() accept paths relative to the Harness API root (the /harnesskit prefix is optional). Absolute URLs, . / .. path segments, and malformed URL encoding raise ValueError, and any caller-supplied authorization header is replaced by the AppKit token. A query string in path is kept; keys in query override the same keys there.

NotifyKit

NotifyKit lets an App submit user-facing events to the AgentOS notification center. Register the app with sdk.agentos.register_bundle() first; sdk.notifykit.publish() uses the active AppKit token and retries once after an authentication failure.

from agentos_sdk import AgentOSSDK, ApiError


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

        try:
            result = await sdk.notifykit.publish(
                request={
                    "eventType": "billing.invoice_ready",
                    "title": "Invoice ready",
                    "message": "Your October invoice is ready.",
                    "subjectId": "invoice-1",
                }
            )
        except ApiError as error:
            if error.status == 429:
                print("Retry in", error.data["retryAfterSeconds"], "seconds")
            else:
                raise
        else:
            print(result["accepted"], result["notification"]["id"])

An App only provides content: eventType (a lowercase dotted name such as task.completed), title, message, and an optional subjectId. The platform decides the level, category, grouping, and deduplication, so repeated events can collapse into one notification whose count grows. Invalid events raise ApiError with status 400. When the App exceeds the per-minute rate limit the call raises ApiError with status 429; error.data["retryAfterSeconds"] says how long to wait.

Direct Clients

The low-level clients are available for explicit-token or custom transport usage. sdk.harnessKit and sdk.notifyKit expose the HarnessKit and NotifyKit clients on an existing AgentOSSDK; they take the AppKit bearer token as the first argument:

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 .