komputer-ai-sdk 0.16.3

Python SDK for the komputer.ai platform

komputer-ai Python SDK

Python client for the komputer.ai platform.

Installation

pip install komputer-ai-sdk

Or install directly from the repository:

pip install git+https://github.com/komputer-ai/komputer-ai.git#subdirectory=komputer-sdk/python

Quick Start

from komputer_ai.client import KomputerClient

client = KomputerClient("http://localhost:8080")

# Create an agent
agent = client.create_agent(
    name="my-agent",
    instructions="Summarize the latest Kubernetes release notes",
    model="claude-sonnet-4-6",
)

# Stream events as the agent works
for event in client.watch_agent("my-agent"):
    if event.type == "text":
        print(event.payload.content)
    elif event.type == "task_completed":
        print(f"Done — cost: ${event.payload.cost_usd}")
        break

Usage

Agents

# Create
client.create_agent(name="researcher", instructions="Research AI trends", model="claude-sonnet-4-6")

# List
agents = client.list_agents()

# Get
agent = client.get_agent("researcher")

# Update
client.patch_agent("researcher", model="claude-haiku-4-5-20251001", lifecycle="Sleep")

# Send a follow-up task to an idle agent
client.create_agent(name="researcher", instructions="Now focus on LLM benchmarks")

# Cancel a running task
client.cancel_agent_task("researcher")

# Delete
client.delete_agent("researcher")

Memories

client.create_memory(name="company-context", content="We are a B2B SaaS company.", description="Background info")
client.patch_agent("my-agent", memories=["company-context"])

memories = client.list_memories()
client.patch_memory("company-context", content="Updated context.")
client.delete_memory("company-context")

Skills

client.create_skill(name="healthcheck", description="Check service health", content="curl -s http://api/healthz")
client.patch_agent("my-agent", skills=["healthcheck"])

skills = client.list_skills()
client.delete_skill("healthcheck")

Schedules

client.create_schedule(
    name="daily-report",
    schedule="0 9 * * *",
    instructions="Generate a daily status report",
    timezone="America/New_York",
)

schedules = client.list_schedules()
client.patch_schedule("daily-report", schedule="0 10 * * *")
client.delete_schedule("daily-report")

Secrets

client.create_secret(name="api-keys", data={"GITHUB_TOKEN": "ghp_xxx", "SLACK_TOKEN": "xoxb-xxx"})
client.patch_agent("my-agent", secret_refs=["api-keys"])

secrets = client.list_secrets()
client.update_secret("api-keys", data={"GITHUB_TOKEN": "ghp_new"})
client.delete_secret("api-keys")

Connectors

client.create_connector(name="slack", service="slack", url="https://mcp.slack.com", auth_type="token")
client.patch_agent("my-agent", connectors=["slack"])

connectors = client.list_connectors()
client.delete_connector("slack")

Offices

offices = client.list_offices()
office = client.get_office("my-office")

Streaming Events

for event in client.watch_agent("my-agent"):
    match event.type:
        case "task_started":
            print("Agent started working...")
        case "text":
            print(event.payload.content)
        case "tool_call":
            print(f"Using tool: {event.payload.tool}")
        case "task_completed":
            print(f"Done — cost: ${event.payload.cost_usd}")
            break
        case "error":
            print(f"Error: {event.payload.error}")
            break

Event types: task_started, thinking, tool_use, tool_result, text, task_completed, task_cancelled, error.

Distributed consumers — group=

By default, watch_agent opens a broadcast subscription: every connected client receives every event. If you run multiple instances of your service (e.g. 3 replicas of a Slack bot) and they all call client.watch_agent("my-agent") without further options, each instance will process every event — duplicate work.

To get queue-style delivery (each event handled by exactly one instance across your fleet), pass group=:

for event in client.watch_agent("my-agent", group="my-bot"):
    ...

The API uses Redis-coordinated routing to deliver each event to exactly one client per group, regardless of how many replicas connect or which API replica they hit. Pick any string for the group name (my-bot, audit-pipeline, prod-webhook-fwd).

Use broadcast for: dashboards, debugging, single-instance workers, anywhere "see everything" is the goal. Use group= for: distributed services, webhook forwarders, anywhere you'd otherwise dedupe events yourself.

On write failure, the API retries delivery to other group members on the same replica before giving up — an event is only lost when all members on the routing replica fail simultaneously. Use client.get_agent_events(name, limit=...) to backfill on reconnect if you need strict exactly-once guarantees.

Context Manager

with KomputerClient("http://localhost:8080") as client:
    agents = client.list_agents()