synapse-runtime-sdk

The Python SDK for the Synapse runtime: publish and subscribe to topics, serve and call services, keep shared values current with the store, and move H.264 video — all from asyncio.

pip install synapse-runtime-sdk              # the SDK
pip install "synapse-runtime-sdk[video]"     # + the H.264 encoder/decoder (CPU or NVIDIA)

Linux x86_64, Python ≥ 3.11. A Synapse runtime to connect to: curl -fsSL <release host>/install.sh | bash.

The surface in one screen

from pydantic import BaseModel
from synapse_sdk import Synapse, Topic, Service, Api, ServiceError, SynapseError

class Imu(BaseModel):
    accel: tuple[float, float, float]

imu = Topic("imu", schema=Imu)                                   # datagram (default): one datagram per message, newest wins
log = Topic("log", delivery="stream")                             # ordered, nothing dropped
api = Api(arm=dict(get_pose=Service(input=dict, output=dict)))    # api.arm.get_pose.name == "arm/get_pose"

async with Synapse("https://localhost:4470", "agent", "sim-1") as node:   # type agent|observer|web|hmi; the name is minted when absent
    pub = await node.publisher(imu)                               # resolves once the runtime admitted it on the current session
    pub.publish(Imu(accel=(0, 0, 9.8)))                           # synchronous; validated by the schema; dropped+counted while unbound
    async for batch in node.subscribe(imu):                       # one read = the newest message of each publisher (queue_depth=0: every message)
        for msg in batch:
            print(msg.publisher, msg.data.accel, msg.t_us)
    view = node.view(imu, stale_time_ms=200); await view.ready(); print(view.get().data)
    reg = await node.serve(api.arm.get_pose, lambda inp, ctx: {"frame": inp["frame"]})   # (input, ctx) -> result; sync runs in a thread, async on the loop
    pose = await node.call(api.arm.get_pose, {"frame": "tool0"}, deadline_ms=1000)       # await the Run for the result
    run = node.call("arm/move_to", {"target": 1}); async for u in run.updates(): ...; await run.cancel()
    board = await node.store("mission/board", access="public"); await board.set("phase", "loading"); board.value["phase"]   # a value kept current
    @node.every(hz=50)                                            # a fixed grid while the node lives (polling bridges)
    def poll(): pub.publish(read_imu())

• Identity: Synapse(url, type, name=None, user_id=None, credential=None, token=None). type bounds the role the runtime grants (agent → agent; web/hmi → observer until promoted on a project); a web/hmi node without user_id mints one. node.info() has name, prefix, role, robot, runtime_id, broker_version.
• Connecting: url is the runtime, https://host:4470; left out, $SYNAPSE_ENDPOINT (set by the runtime for every task it starts). Through a relay, give the relay with the runtime's path, https://relay:4443/runtimes/<runtime_id> (runtime_id from node.info() of a node connected directly, or the runtime's banner), and token= the relay token (a jwt= in the URL is taken as the token; neither is ever printed). connect() (or async with) resolves once the runtime answered. A name a live node already holds is refused admission_refused (retryable); no answer within connect_timeout_ms is not_connected (through a relay: check the token); a relay address without the runtime's path is invalid_input. The first connect tries a retryable failure again for up to 30 s, then raises it.
• Sessions come and go, the node stays: a lost session is replaced (reconnect=True, backoff reconnect_initial_ms → reconnect_max_ms), every publisher, registration, subscription and namespace replays its intent on the new session; in-flight calls fail connection_lost; nothing is resumed. node.status / async for s in node.watch_status() / node.on_status(cb). reconnect=False ends the node closed on a drop.
• The runtime decides: publisher(), serve(), store() resolve on admission; a refusal (already_registered, unauthorized, owner_taken, ...) is asked again, paced, with a warning each time (warn= callback, default logging).
• Time: t_us is the publisher's clock, microseconds. node.clock(), node.time_offsets(), node.to_local_us(t_us, dom), node.age_us(msg); deadline_ms= on a subscription drops a message older than that in our own time domain.
• Errors: one SynapseError(code, message, data, retryable); ServiceError(code, message, data) raised by a handler crosses verbatim (data.app for an application code); the same error codes in every language.
• Video: cam = await node.video_publisher("gs/video") then cam.publish(image); async for image in node.video_subscriber("gs/video"); see below. A video topic is Topic(name, content_type="video/h264"): datagrams with loss repair unless fec= says otherwise; pub.publish(access_unit) on it sends one H.264 access unit, and a subscriber that names the topic gets bytes with msg.key / msg.config / msg.recovery.
• Bodies follow the topic: a subscription decodes what the runtime says the topic carries — msgpack topics give values (validated by your schema when the Topic has one), any other content type gives bytes.
• The certificate pin (broker_cert_sha256=) is discovered when absent (SYNAPSE_BROKER_CERT_SHA256, then http://host:port/certificate.sha256). SIGINT/SIGTERM close every open node, then re-raise (signals=False opts out).
• Timeouts and periods are _ms integers everywhere, in every language.

Scene

The active project's world: one set of entities (frames, robots, objects, cameras …) the runtime keeps for the project, in two planes. The authored plane is what exists — the hierarchy and the components — and every write to it is one atomic batch the runtime checks. The live plane is what is happening now — poses, joint positions and state of entities, written many times a second, never saved. What you read is both merged: a live pose replaces xform.t/xform.q of the entity it names; removing it brings the authored value back.

scene = await node.scene()                     # the active project's scene; detached (empty) while no project is active
await scene.wait_attached(timeout=10)          # until a project's scene is in
world = scene.entities                         # {id: entity}, read once to build a simulation from it
scene.entity(id); scene.tree(); scene.query(lambda e: e["kind"] == "robot")   # the value now; tree() = [{id, children}]
scene.status                                   # state (attached | detached | closed), synced, pending, version, hash, detached_drops

# author: each resolves once scene.entities shows it; a refusal (locked, name_taken, invalid_tree, ...) raises
id = await scene.spawn("box", parent=None, name="crate", xform={"t": [1, 0, 0]}, visual=[{"kind": "box", "size": [1, 1, 1]}])
await scene.patch(id, props={"mass": 2}); await scene.rename(id, "crate-1"); await scene.reparent(id, other, before=sibling)
await scene.set_locked(id, True); await scene.despawn(id, cascade=True)      # cascade: with its subtree
async with scene.txn() as t:                   # one atomic batch when the block ends
    a = t.spawn("frame", name="a"); t.patch(a, props={"k": 1})

# live: synchronous, like publish — one call per tick
scene.live(poses=[{"id": arm, "t": [0, 0, 1], "q": [0, 0, 0, 1]}], joints=[{"id": arm, "j": {"shoulder": 0.3}}], lifetime_ms=500)
scene.live(state=[{"id": arm, "s": {"badge": "serving"}}]); scene.live(remove=[arm])   # remove: back to authored

async for ops in scene.watch_changes(): ...    # authored changes; also watch_live_removals(), watch_live_denied(), watch_state()

• A live row with lifetime_ms goes away that long after its last write (timeout on watch_live_removals()), and when its writer leaves; one without it stays (a parked robot). The first writer of an id owns its live rows; another writer's rows for it are dropped and reported on its watch_live_denied().
• While detached, author verbs raise project_inactive and live() drops its batch (counted in status.detached_drops).
• Helpers: ulid() (a fresh entity id), order_between/order_before/order_after (sibling order values), derived_frame_name(parent, frame).

Derived frames

A robot's robot.frames (names of links) and an entity's hardpoints (named mounting points with a pose) each stand for frame children of that entity, marked flags.derived. materialize_frames and materialize_hardpoints keep those children in step with the declaration: they spawn what is missing, rename the frames after the entity is renamed, set a changed pose, and remove a frame that is no longer declared (refused, as a veto, when something is parented under it). They only add ops to a txn and return a report of what they did; nothing is sent until the block ends.

from synapse_sdk import materialize_frames, materialize_hardpoints

async with scene.txn() as t:                   # when a scene is loaded: no poses, so nothing is spawned or moved
    report = materialize_frames(t, robot, scene.entities)
async with scene.txn() as t:                   # after an edit of the robot: the rest poses are your FK at its joints
    report = materialize_frames(t, robot, scene.entities, {"tool0": {"t": [0, 0, 1.2], "q": [0, 0, 0, 1]}})
    if not report.clean:                       # raising here sends nothing
        raise RuntimeError("; ".join(report.errors + report.vetoes))
async with scene.txn() as t:                   # hardpoint frames take their pose from the hardpoint itself
    materialize_hardpoints(t, entity, scene.entities)

The report: frames (each declared name → its frame's id), spawned, updated, removed, vetoes, errors, and clean (no vetoes, no errors). A declared frame that does not exist yet and has no pose given is an error, never a spawn.

Tasks

A task is a program the runtime starts and keeps running for you (a driver, a simulator, a planner). You declare which tasks to run; an operator runs and stops them. There are two kinds, and stopping one kind never touches the other:

• Robot tasks belong to the robot and keep running when the active project changes (an arm driver). The runtime keeps their declarations across restarts.
• Project tasks belong to the active project: the tasks component of its scene's entities (one component; it replaces the prototype's behaviors). Switching projects stops them.

A declaration is {ref, enabled?, config?, schema?} under an instance name: ref is the task's address (as the task tool lists it), enabled: False leaves it out of the next run, config is passed to the task, and schema pins the task's schema hash (a different one fails the run). Whether a task has a process to run is the task's own spec: one with nothing to run is passive, counted and never started (something else uses its config, a simulator say).

Everything is under node.tasks, the same object for the node's life:

t = node.tasks

# robot tasks (operator role)
await t.declarations.set("arm", {"ref": "python:ur-driver:driver", "config": {"host": "10.0.0.5"}})
await t.declarations.set("arm", {**t.declarations.get("arm"), "enabled": False})     # off for the next run

# project tasks: the `tasks` component of a scene entity
scene = await node.scene()
await scene.patch(sim_id, tasks={"sim": {"ref": "python:mujoco-sim:sim"}})

r = await t.run_project_tasks()        # {"launched": 2, "passive": 3}
await t.stop_project_tasks()           # None, also when nothing ran
await t.run_robot_tasks()
await t.stop_robot_tasks()

await t.status.ready()                                          # the value is in
t.status.value["project"]["state"]                              # "starting" | "running" | "stopping" | "stopped" | None (not said yet)
t.status.value["project"]["tasks"].get(entity_id, {}).get("sim")   # {"scope", "addr", "instance", "entity", "state": "up" | "down", "since", "restarts", "exit"?, "log"}
t.catalog.value["python:ur-driver:driver"]                      # {"addr", "workspace", "pkg"?, "name"?, "inputs"?, "schemaHash"?, ...}

async for s in t.status.watch():                                # the whole status after each change
    arm = s["robot"]["tasks"].get("arm")
    if arm and arm["state"] == "down":
        print(f"arm exited {arm.get('exit')}, {arm['restarts']} restarts; log {arm['log']}")

• declarations is a store handle (value, get, set, delete, clear, synced, status, watch, ready); a write waits for the value to be current, then resolves once value shows it. status and catalog are read-only (value, synced, ready(), watch()). Each opens the first time you touch it; the verbs open nothing.
• status.value is {"robot": {"state", "since"?, "reason"?, "tasks": {instance: task}}, "project": {..., "tasks": {entity_id: {instance: task}}}}. reason says why a run stopped or failed (project_switch, or an error code). exit is {"code", "at"}, {"signal", "at"} or {"at"} (Unix ms). A part that did not change keeps its object (is tells what changed). watch() is state, not history: a slow reader sees the latest status, and restarts is the count to trust.
• A run checks every enabled declaration first and starts nothing unless all pass. A failed check raises preflight_failed, a task that could not be started launch_failed (what had started is stopped); both carry e.data["report"], one entry per declaration at fault: {scope, entity?, entity_name?, instance, addr, reason, detail?} with reason one of missing (unknown address), schema_mismatch, spec_error, invalid_input (the config); is_tasks_refusal(e) tells them apart. A second run while one is running raises already_running; a project run with no active project project_inactive. A run's reply is the runtime's answer; what runs afterwards is status.
• A task that exits is started again after a delay that grows (by default from 1 s up to 10 s) while it keeps exiting. A crash loop never ends the run; restarts climbs.
• Running and stopping need the operator role, which today is a promotion on the active project: robot tasks too need a project to be active (or the runtime's robot_autostart).

Writing a task

A task owes the runtime nothing. A task is any program the task tool can describe. The runtime asks the tool for the task's spec at every run and starts the spec's command directly; your declaration's config reaches it as the spec's flags, positionals or environment variables. The program never has to connect, report or handshake. What the runtime does to it:

• Environment: cleared, then exactly the spec's env, then the stamps below. There is no PATH or HOME unless the spec sets them. Working directory: the spec's cwd. No stdin.
• Output: stdout and stderr are appended to a log file, with one header line per start; its path is log in the status. Python buffers stdout when it is a file: set PYTHONUNBUFFERED=1 in the spec's env, or print(..., flush=True).
• Stop: SIGTERM to the program's process group, then SIGKILL 5 s later. Handle SIGTERM and leave within 5 s.
• Exit: any exit is followed by a new start after a delay (1 s doubling to 10 s, back to 1 s once a run lasted longer than 10 s); restarts and exit in the status show it.
• Runtime gone: the program gets SIGTERM.

The stamps (optional to read):

A plain task (no SDK at all):

# serve.py; its spec: argv ["python3", "serve.py"], mapping port → {as: "flag", flag: "--port"}
import argparse, http.server, signal, sys
p = argparse.ArgumentParser(); p.add_argument("--port", type=int, default=8080); a = p.parse_args()
srv = http.server.ThreadingHTTPServer(("0.0.0.0", a.port), http.server.SimpleHTTPRequestHandler)
signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))          # the runtime stops us with SIGTERM
srv.serve_forever()

A task that talks to the runtime (a simulator's shape):

import asyncio, json, os, signal
from synapse_sdk import Synapse, Topic
from synapse_sim.util import JointStates                         # the simulator's own message model

config = json.loads(os.environ.get("SYNAPSE_CONFIG", "{}"))     # or flags from the spec's mapping

async def main():
    stop = asyncio.Event()
    for s in (signal.SIGINT, signal.SIGTERM):                    # our own graceful path; the SDK leaves an installed handler alone
        asyncio.get_running_loop().add_signal_handler(s, stop.set)
    async with Synapse(type="agent", name="sim-server-1") as node:              # no url: $SYNAPSE_ENDPOINT
        state = await node.store("mujoco/state", access="protected")             # ours alone; a second sim waits
        scene = await node.scene(); await scene.wait_attached(timeout=10)
        joints = await node.publisher(Topic("ur5e/joint_states", schema=JointStates))   # once; publish() is sync
        async def control(req, ctx):                                             # (input, ctx); ctx.update() is sync
            ...; return {"result": "stopped"}
        await node.serve("mujoco/control", control)
        await state.set({"simulation_status": "stopped", "sim_time": 0.0})       # one atomic write, resolves when visible
        # loop: joints.publish(...); scene.live(joints=[...], poses=[...], lifetime_ms=2000)
        await stop.wait()

asyncio.run(main())

A name is not taken from SYNAPSE_INSTANCE for you: two entities with an instance sim would collide. Give each task's node its own name (from its config, say).

Examples

Getting started, one verb per file: examples/getting-started/python/01_connect.py … 13_tasks.py in the repository.

Video

Needs the extra: pip install "synapse-runtime-sdk[video]" (without it the two calls below raise MediaError saying so).

cam = await node.video_publisher("gs/video")       # resolves once the topic is admitted
cam.publish(image)                                  # numpy uint8 (h, w, 3) RGB: convert + encode + publish in one call
cam.publish(image, t_us=capture_us)                 # the capture time, if you have one (default: now)
cam.close()

async for image in node.video_subscriber("gs/video"):   # decoded pictures
    print(image.width, image.height, image.t_us)
    frame = image_to_numpy(image)                       # (h*3/2, w) I420; np.asarray(image.y) is the luma plane, no copy
node.video_subscriber("gs/video", on_image=show)       # or a callback per picture (def or async def)

• publish is synchronous and queues nothing: the caller waits for one conversion + encode (about a millisecond at 640x360). The picture's format comes from its shape — (h, w, 3) RGB, (h, w, 4) RGBA, (h*3/2, w) I420, a tuple of planes (y, u, v) (I420) or (y, uv) (NV12); pixel_format="bgr24" for OpenCV pictures.
• Options of video_publisher, all optional: pixel_format, fps (nominal, for rate control; 30), bitrate_kbps (2000), backend ("auto": NVIDIA when present, else CPU).
• One publisher, one size: the encoder is made from the first picture; a picture of another size raises SynapseError("invalid_input").
• A subscriber that joins mid-stream starts at the next picture it can decode, and asks the publisher for one when it waits too long; close() stops it.
• The encoder and decoder are also there on their own (H264Encoder, H264Decoder, TestPattern in synapse_sdk.media) for a pipeline of your own.