Rusty SDK — Python client

Zero-dependency, stdlib-only Python SDK for rusty-server. Threads, runs (background / blocking / SSE-streaming), checkpoint history, time travel (fork + replay), assistants, crons, the cross-thread KV store, and the R0.6 durable task queue's control plane — over plain HTTP + SSE with nothing but urllib.request and json. Python 3.8+, no pip install of anything else, ever.

Philosophy

This SDK is the "interop over HTTP" story: the Rust server owns orchestration, checkpoints, and streaming; any language that can speak HTTP and parse SSE can drive it. Python is the language most likely to already be on the machine, so this client deliberately uses only the standard library — no requests, no httpx, no sseclient. Drop the rusty_client/ package into any project (or any python3 -c one-liner) and it works. The trade-off is explicit: you get a hand-rolled SSE parser and blocking I/O instead of a fancy async stack — which is exactly what you want for scripts, CI, notebooks, and LangChain-adjacent glue code.

Install

pip install rusty-agent-runtime

From a local checkout (editable path install):

pip install /path/to/repo/sdks/python

Or just copy the package — it has no build step and no dependencies:

cp -r sdks/python/rusty_client /your/project/

Quickstart

Start the demo server (scripted model — no network, no API keys):

cargo run -p rusty-agent-server --example server_demo
# rusty-server demo on http://127.0.0.1:8100  (graphs: pipeline, react_agent)

Then, mirroring the curl quickstart from the server README:

from rusty_client import RustyClient

client = RustyClient("http://127.0.0.1:8100")   # api_key="..." when auth is on

# Liveness + what's registered
client.ok()      # True
client.info()    # {"service": "rusty-server", "graphs": [...], ...}

# Create a thread bound to a registered graph
thread = client.create_thread("pipeline")
tid = thread["thread_id"]

# Blocking run
result = client.run_wait(tid)
# {"status": "success", "output": {"log": ["first", "second"]}, ...}

# Streaming run (SSE) — frames arrive as the graph executes
for frame in client.run_stream(tid, stream_mode=["updates", "values"]):
    print(frame.event, frame.id, frame.data)
# metadata -:0:1 {"run_id": ..., "graph": "pipeline", ...}
# updates  -:0:2 {"step": 0, "updates": {"log": ["first"]}}
# values   <cp>:0:3 {"log": ["first"]}
# ...
# end      <cp>:1:6 {"status": "success"}

# Background run + polling
run = client.run(tid)
status = client.run_status(run["run_id"])   # terminal runs carry output/error

# Thread state + checkpoint history
client.get_state(tid)                       # {"values", "next", "checkpoint"}
client.history(tid, limit=10)               # newest first

# Time travel: fork at an earlier checkpoint, replay on the fork
mid = next(h for h in client.history(tid) if h["next"] == ["second"])
cp_id = mid["checkpoint"]["checkpoint_id"]
fork = client.fork(tid, checkpoint_id=cp_id)
client.run_wait(fork["thread_id"], checkpoint_id=cp_id)

# Human-in-the-loop: resume an interrupted run
client.run_wait(tid, command={"resume": {"approved": True}})

# Assistants, crons, KV store
assistant = client.create_assistant("support-bot", graph="react_agent",
                                    config={"recursion_limit": 25})
client.run_wait(tid, assistant_id=assistant["assistant_id"])

cron = client.create_cron(graph="react_agent", interval_secs=60,
                          input={"messages": [{"role": "user", "content": "hourly summary"}]})
client.list_crons()
client.delete_cron(cron["cron_id"])

client.kv_put("memories", "user-1", {"preference": "dark-mode"})
client.kv_list("memories")
client.kv_delete("memories", "user-1")

# Durable task queue (R0.6) — control plane: submit, observe, cancel
enqueued = client.tasks.enqueue(
    "send_email", {"to": "user@example.com"},
    idempotency_key="welcome-42",     # re-enqueueing dedupes on this key
    effect="idempotent",              # declares the work is safe to retry
    deadline="2026-08-11T00:00:00Z",  # RFC 3339, across attempts
)
enqueued  # {"task_id": "…", "deduplicated": False}

task = client.tasks.get(enqueued["task_id"])   # the full TaskRecord
client.tasks.list()                            # all tasks, oldest first
client.tasks.list(status="dead")               # the dead-letter queue
client.tasks.cancel(enqueued["task_id"])       # 409 if already terminal
client.tasks.cancel_run_tasks(run["run_id"])   # cancel a run's open tasks

With auth configured on the server (ServerConfig::with_api_key), pass RustyClient(url, api_key="...") — it is sent as the X-Api-Key header on every request.

API reference

Durable tasks (R0.6)

client.tasks is the control plane of the durable task queue: submit work, observe records, cancel. Task records carry the full envelope — kind, payload, pool, status (queued / leased / failed / completed / dead / cancelled), attempt / max_attempts, the live lease, error_class + last_error from the last failed attempt, idempotency_key, result / receipt when settled, run/thread linkage, cancel_requested, deadline, and timestamps.

Two submission paths: enqueue makes the task claimable immediately; enqueue_outbox writes through the transactional outbox (202 accepted — the relay publishes it into the queue within one poll interval, deduped on the idempotency key, so a crash neither loses nor doubles the task). Cancellation is a hint for promptness, not a force: a queued or retry-scheduled task goes terminal-cancelled immediately, while a leased task keeps its lease with cancel_requested set so its holder aborts cleanly on the next heartbeat.

Why there's no claim / heartbeat / complete / fail here: those endpoints are the queue's worker-machine half — lease-guarded by worker_id, they exist so a worker process holds, renews, and settles exactly one lease at a time. A control-plane client that claimed a lease would starve real workers until the visibility timeout, or race their settlement into 409s. That surface belongs to the worker SDK (rusty-worker's ActivityWorker); this client never holds leases.

Streaming details

• run_stream returns a generator of SSEEvent dataclasses: event (e.g. metadata, updates, values, messages, error, end), data (JSON-decoded when possible), and id ({checkpoint_id}:{step}:{seq}).
• stream_mode filters frame families ("updates", "values", "messages"); metadata/error/end are always emitted.
• Pass last_event_id=frame.id to resume a dropped connection — the server replays only frames after that id (sent as the Last-Event-ID header).

Errors

Every non-2xx response raises RustyError with .status (HTTP code, None for transport failures) and .body (raw response text):

from rusty_client import RustyError

try:
    client.create_thread("no_such_graph")
except RustyError as exc:
    print(exc.status, exc.body)   # 404 / 400, server's error JSON

Tests

The suite has two halves. test_client.py is a true end-to-end test: it builds (if needed) and launches the real server_demo binary as a subprocess, waits for /ok, exercises every endpoint family against it, and kills the process afterwards. test_sse_parser.py and test_tasks.py are no-I/O unit tests: the SSE parser and the tasks control plane are exercised against fake transports (mocked urllib.request.urlopen / _request), so they run anywhere in milliseconds.

python -m pytest sdks/python/tests -q

Note: server_demo registers no interrupting graph, so the interrupt/resume round trip is a documented skip in the suite; the client's resume path is run_wait(tid, command={"resume": value}).

License

Dual-licensed under MIT OR Apache-2.0, same as the rest of the repo.