A robot in Python
The Python SDK is for a robot whose own process already holds encoded frames. It is a thin cffi wrapper over libseyd, the C ABI (ADR 0004): the wire format, FEC, NAT traversal, QoS and signaling all live in the Rust core, so this SDK, the daemon and the C wrapper cannot drift apart. If your camera publishes RTSP or RTP on a socket, run the daemon instead and skip the code.
Install
Section titled “Install”No package is published yet. From a source checkout, build the library and point Python at the SDK:
cargo build -p seyd-ffi --releasepip install cffiPYTHONPATH=sdks/python python3 -c "import seyd; print(seyd.ABI_VERSION)"The SDK finds libseyd in this order: $SEYD_LIBRARY (a full path), then inside the installed package, then the workspace’s target/release and target/debug. From a checkout it needs no install step. When none of those exist it raises SeydLibraryNotFound, which names the fix.
The shape of an integration
Section titled “The shape of an integration”Create an Agent, add channels, set handlers, start(), then push frames and messages. Channels are numbered from 1 in the order you add them, and that numbering is what the pilot sees.
from seyd import Agent, ChannelKind
with Agent("my-robot", "wss://signal.seyd.io/ws", qos_profile="latency") as agent: video = agent.add_channel(ChannelKind.VIDEO, "main", codec="avc1.42001f", fps=25) telemetry = agent.add_channel(ChannelKind.SENSOR, "telemetry") agent.add_channel(ChannelKind.COMMAND, "drive")
agent.on_command = lambda channel, payload: motors.apply(payload) agent.on_session_ended = lambda sid, reason: motors.park()
agent.start() for au in my_encoder: agent.push_frame(video, au.bytes, keyframe=au.is_idr, capture_ts_us=au.pts_us)Seyd relays encoded bytes. It never decodes, re-encodes or inspects a payload: hand it the access units your encoder produced, one per push_frame, and the pilot decodes exactly those.
The example: a robot with its own encoder
Section titled “The example: a robot with its own encoder”sdks/python/examples/ffmpeg_robot.py is a complete robot. x264 runs in a pipe, access units go straight into Seyd, and there is a sensor channel and a command channel. It is rendered here from the repository, so what you read is what runs.
#!/usr/bin/env python3# Copyright 2026 Anton Gravestam# SPDX-License-Identifier: Apache-2.0"""A complete Seyd robot in Python, with its own encoder.
This is the shape of a real integration: the robot already has an H.264encoder, and Seyd is handed the encoded access units. Nothing is transcoded andnothing is decoded — Seyd relays the bytes the encoder produced.
python3 ffmpeg_robot.py --robot-id seyd-py python3 ffmpeg_robot.py --signal ws://localhost:8080/ws # against a local cloud
Or ../../../py-robot.sh, which checks the prerequisites first.
Contrast with `seydd`, which owns the RTP/RTSP plumbing for robots that publishvideo on a socket. Here the customer's process owns the pipeline and calls`push_frame` directly, which is one process and one copy fewer.
The encoder settings below live here, in the publisher, on purpose: Seyd statestransport-observable *targets* (a bitrate ceiling, a latency budget, a GOPbound) through `on_requested_config`, and the publisher decides how to meetthem. See CLAUDE.md, "Where QoS settings live"."""
import argparseimport jsonimport loggingimport osimport signalimport subprocessimport sysimport threadingimport timefrom pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from seyd import Agent, ChannelKind # noqa: E402
log = logging.getLogger("ffmpeg-robot")
#: Mirrors sim/video-source.sh so the two sources are comparable.PROFILES = { "latency": dict(w=960, h=540, fps=30, kbps=1500, gop=30, preset="ultrafast", vbv_ms=100, idr_s=10), "balanced": dict(w=1280, h=720, fps=30, kbps=3000, gop=30, preset="veryfast", vbv_ms=100, idr_s=10), "quality": dict(w=1280, h=720, fps=30, kbps=6000, gop=60, preset="veryfast", vbv_ms=200, idr_s=4),}
def ffmpeg_command(profile: dict, device: str) -> list[str]: """H.264 Annex-B on stdout, one access unit per AUD.
`aud=insert` is what makes the stream splittable without a full parser: an access-unit delimiter marks every AU boundary explicitly, so the reader below never has to interpret slice headers. """ if device == "lavfi": source = ["-re", "-f", "lavfi", "-i", f"testsrc2=size={profile['w']}x{profile['h']}:rate={profile['fps']}"] else: source = ["-f", "avfoundation", "-framerate", str(profile["fps"]), "-video_size", f"{profile['w']}x{profile['h']}", "-i", device] bufk = profile["kbps"] * profile["vbv_ms"] // 1000 return [ "ffmpeg", "-hide_banner", "-loglevel", "error", "-fflags", "nobuffer", *source, "-pix_fmt", "yuv420p", "-c:v", "libx264", "-tune", "zerolatency", "-preset", profile["preset"], "-profile:v", "baseline", "-b:v", f"{profile['kbps']}k", "-maxrate", f"{profile['kbps']}k", "-bufsize", f"{bufk}k", # Periodic intra refresh instead of periodic IDRs (ADR 0009): `gop` is # the refresh sweep period, and an IDR is forced only every `idr_s`, the # profile's maxGopMs — x264 in a pipe cannot answer a recovery-request, # so this bounds a joining pilot's wait for its first picture. "-g", str(profile["gop"]), "-keyint_min", str(profile["gop"]), "-bf", "0", "-x264-params", "scenecut=0:intra-refresh=1", "-force_key_frames", f"expr:gte(t,n_forced*{profile['idr_s']})", "-an", "-flush_packets", "1", "-max_delay", "0", "-bsf:v", "h264_metadata=aud=insert", "-f", "h264", "pipe:1", ]
def access_units(stream): """Yield `(bytes, is_keyframe)` per access unit from an Annex-B stream.
Splits at each access-unit delimiter (NAL type 9); an AU is a keyframe if it carries an IDR slice (type 5) or a parameter set (type 7). """ buf = bytearray() pending = bytearray() while True: chunk = stream.read(4096) if not chunk: break pending += chunk # Split on start codes, keeping the last (possibly partial) NAL back. parts = pending.split(b"\x00\x00\x01") pending = bytearray(parts.pop()) for part in parts: if not part and not buf: continue # `part` is the previous NAL's payload plus its trailing zeroes. nal_type = part[0] & 0x1F if part else 0 if nal_type == 9 and buf: au = bytes(buf) buf.clear() yield au, _is_keyframe(au) buf += b"\x00\x00\x01" + part if buf: yield bytes(buf), _is_keyframe(bytes(buf))
def _is_keyframe(au: bytes) -> bool: for i in range(len(au) - 3): if au[i] == 0 and au[i + 1] == 0 and au[i + 2] == 1: if (au[i + 3] & 0x1F) in (5, 7): return True return False
def make_command_handler(udp_port: int | None): """Log commands, and optionally forward them to a UDP port.
Forwarding is how you bridge Seyd to robot software that is already listening on a socket — the same generic interface `seydd` offers. Sending is non-blocking, which matters: this runs on the Seyd callback thread. """ sock = None if udp_port: import socket as _socket
sock = _socket.socket(_socket.AF_INET, _socket.SOCK_DGRAM) sock.setblocking(False)
def on_command(channel: int, payload: bytes) -> None: log.info("command on channel %d: %s", channel, payload[:120]) if sock: try: sock.sendto(payload, ("127.0.0.1", udp_port)) except OSError as e: log.debug("forward failed: %s", e)
return on_command
def main() -> int: ap = argparse.ArgumentParser(description=__doc__) ap.add_argument("--robot-id", default="seyd-py") # The deployed cloud, so the example runs without a second terminal. ap.add_argument("--signal", default="wss://seyd-signal-flj7s44j4a-ew.a.run.app/ws") ap.add_argument("--profile", default="latency", choices=sorted(PROFILES)) ap.add_argument("--device", default="lavfi", help="avfoundation device index, or 'lavfi' for a synthetic source") ap.add_argument("--credential", default="./robot.key") ap.add_argument("--host-override", default=None, help="advertise this address only (LAN testing)") ap.add_argument("--command-udp", type=int, default=None, metavar="PORT", help="forward pilot commands to this UDP port on localhost, " "the way seydd does, for robot software already " "listening on a socket") args = ap.parse_args() logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
profile = PROFILES[args.profile] stopping = threading.Event()
with Agent( args.robot_id, args.signal, credential_path=args.credential, qos_profile=args.profile, host_override=args.host_override, ) as agent: video = agent.add_channel(ChannelKind.VIDEO, "main", codec="avc1.42001f", fps=profile["fps"]) telemetry = agent.add_channel(ChannelKind.SENSOR, "telemetry") # Named `ptz` because that is what the demo pilot's controls send on. agent.add_channel(ChannelKind.COMMAND, "ptz")
# Handlers run on one Seyd thread and must not block. agent.on_session_started = lambda sid, role, path: log.info( "session %s started (%s over %s)", sid, role.name.lower(), path) agent.on_session_ended = lambda sid, reason: log.info( "session %s ended (%s) — park actuators here", sid, reason) agent.on_command = make_command_handler(args.command_udp) # A real publisher reconfigures its encoder here; x264 in a pipe cannot # be re-rated live, so this example only reports the request. agent.on_requested_config = lambda cfg: log.info("requested config: %s", cfg) agent.on_recovery_request = lambda channel, kind, reason: log.info( "recovery request: %s (%s)", kind, reason) agent.on_signal_state = lambda state, detail: log.info( "signal %s%s", state.name.lower(), f": {detail}" if detail else "")
agent.start() log.info("robot %s online — open the pilot at ?robot=%s", args.robot_id, args.robot_id)
def sensor_loop(): seq = 0 while not stopping.wait(0.1): agent.push_message(telemetry, {"seq": seq, "t_us": int(time.time() * 1e6)}) seq += 1
threading.Thread(target=sensor_loop, daemon=True).start()
cmd = ffmpeg_command(profile, args.device) log.info("encoder: %s", " ".join(cmd)) proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, bufsize=0) signal.signal(signal.SIGINT, lambda *_: stopping.set())
frames = 0 try: for au, keyframe in access_units(proc.stdout): if stopping.is_set(): break # A real integration passes the encoder's own capture timestamp; # raw Annex-B on a pipe carries none, so this is the read clock. agent.push_frame(video, au, keyframe=keyframe, capture_ts_us=int(time.time() * 1e6)) frames += 1 if frames % (profile["fps"] * 10) == 0: log.info("%d frames pushed, %s", frames, json.dumps(agent.counters)) finally: stopping.set() proc.terminate() proc.wait(timeout=5) log.info("stopping after %d frames: %s", frames, json.dumps(agent.counters)) return 0
if __name__ == "__main__": sys.exit(main())Walking through it:
PROFILES and ffmpeg_command. The encoder settings live in the publisher, on purpose. Seyd states transport-observable targets through on_requested_config; resolution, preset and VBV are the publisher’s to choose. The command line follows the publisher contract: Baseline profile and -bf 0 (no B-frames), zerolatency, a VBV of about 100 ms, and periodic intra refresh (intra-refresh=1) instead of periodic IDRs. -g is the refresh sweep period, and force_key_frames forces an IDR only every idr_s seconds, the profile’s maxGopMs: x264 in a pipe cannot answer a recovery-request, so this bounds a joining pilot’s wait. aud=insert marks every access-unit boundary so the reader needs no slice parser.
access_units and _is_keyframe. Splits the Annex B stream at each access-unit delimiter (NAL type 9) and flags an access unit as a keyframe when it carries an IDR slice (type 5) or a parameter set (type 7). That flag is what push_frame needs; Seyd does not look inside the bytes to find it.
make_command_handler. Logs each command and, with --command-udp, forwards it to a local UDP port with a non-blocking socket. That is the same generic interface seydd offers, so robot software already listening on a socket needs no change. Non-blocking matters: the handler runs on the Seyd callback thread.
main. Builds the Agent with credential_path, qos_profile and an optional host_override for LAN testing. Adds the video channel with the codec string the pilot will configure its decoder from, a telemetry sensor channel and a ptz command channel (named so the demo pilot’s controls reach it). Registers the handlers, all of which only log: a real publisher reconfigures its encoder in on_requested_config and answers on_recovery_request. After start() a thread pushes a counter on the sensor channel at 10 Hz, and the main thread reads access units from ffmpeg and calls push_frame with the read clock as capture_ts_us, because raw Annex B on a pipe carries no timestamp. Every ten seconds it prints agent.counters. The with block stops the agent on the way out.
The rules
Section titled “The rules”- Handlers run on one Seyd thread and must not block. A slow handler delays every later event. Queue the work and return; do no I/O in a handler.
push_frameandpush_messagenever block. A frame arriving while the previous one is still going out is dropped by admission control and counted inagent.counters.frames_dropped_backlog. Keyframes are never dropped. This is whole frame or nothing: a torn frame costs a GOP, a skipped frame costs a frame.- Pass the encoder’s own
capture_ts_uswhenever you have it. The pilot paces presentation on the source’s capture clock (ADR 0005); passing “now” adds jitter the pilot cannot remove. - Park actuators in
on_session_ended. It fires whether the pilot said goodbye or simply vanished. - Answer
on_requested_configandon_recovery_request.on_requested_configis the one place Seyd talks to your encoder: a bitrate ceiling, a latency budget, a GOP bound. Ignore it and Seyd’s rate control cannot act on a congested link.on_recovery_requestcarriesltr,intra_refreshoridr; answer with what your encoder can do, and treatmaxGopMsas a long ceiling, never a one-second IDR cadence (ADR 0009).
Simulcast layers (ADR 0008) are not exposed by the Python SDK yet; they are on the C ABI and in the daemon.
Running it
Section titled “Running it”./py-robot.sh # robot id seyd-py, synthetic sourceVIDEO_DEVICE=0 ./py-robot.sh # the Mac webcam insteadDARC_QOS_PROFILE=latency ROBOT_ID=my-py ./py-robot.shSIGNAL_URL=ws://localhost:8080/ws ./py-robot.sh # against a local cloudENROL_TOKEN=seyd_enr_… ./py-robot.sh # first run on a cloud with real accountsThe script builds libseyd, checks for ffmpeg and cffi, and runs the example against the deployed signal server, printing the pilot URL to open in Chrome. ENROL_TOKEN redeems an enrolment token into the credential file the robot then runs with. The C ABI has no enrolment call yet, so the script goes through seydd enrol; the key file format is shared. To run the example directly:
python3 sdks/python/examples/ffmpeg_robot.py --robot-id seyd-py --device lavfiThe Python reference lists every class, handler and method, generated from the package’s docstrings.