Cyberwave Python SDK

Making the physical world programmable.
Connect, control, and simulate any robot. The same code runs in simulation and on real hardware.

Cyberwave is an all-in-one platform for building and deploying intelligent physical AI agents. Connect a physical robot or sensor, test it in simulation, and run AI models; all through one Python SDK. This package is the official client.

Installation

pip install cyberwave

Optional features install via extras, e.g. cyberwave[camera] (video streaming), cyberwave[ml] (vision models), or cyberwave[zenoh] (edge data bus). See the installation docs for the full list.

Quick Start

Get an API key from your Cyberwave instance (Profile → API Tokens) and export it:

export CYBERWAVE_API_KEY="your_api_key_here"

A token scoped to more than one workspace also needs to know which one you mean. Set it once and every request carries it:

export CYBERWAVE_WORKSPACE_ID="your_workspace_uuid"   # or Cyberwave(workspace_id=...)

You can skip this for a token scoped to a single workspace — there is nothing to disambiguate. Without it, a multi-workspace token gets back 400 workspace_required, listing the workspaces the token holds.

Then create and control your first digital twin:

from cyberwave import Cyberwave

cw = Cyberwave()  # reads CYBERWAVE_API_KEY from the environment

# Create a digital twin from a catalog asset.
# Pin it with canonical slugs. UUID values remain accepted for compatibility.
# Omit both and Cyberwave creates a "Quickstart Environment" for you automatically.
arm = cw.twin(
    "the-robot-studio/so101",
    twin_id="acme/twins/arm-station-1",
    environment_id="acme/envs/production-floor",
)

# Place it in the scene (editor layout)
arm.edit_position(x=1.0, y=0.0, z=0.5)
arm.edit_rotation(yaw=90)  # degrees

# Move a joint by name
joint_names = arm.joints.list()
if joint_names:
    arm.set_joints({joint_names[0]: -0.2})  # radians
    print(arm.get_joints())

# Drive a locomotion twin in simulation
cw.affect("simulation")          # or cw.affect("live") for the real robot
rover = cw.twin("unitree/go2")
rover.move_forward(0.3)

cw.disconnect()

The same script targets real hardware by switching cw.affect("live"), no other changes.

Core Concepts

• Twins — virtual representations of robots and sensors. You develop and test against a twin, then deploy to hardware with identical code. Instantiate any catalog asset with cw.twin("vendor/slug").
• Environments — scenes your twins live in. Validate quickly in the browser-based Playground, or use MuJoCo for high-fidelity physics and RL.
• Simulation vs. live — cw.affect("simulation") and cw.affect("live") switch where commands and state go. The same code drives both.
• Edge & cloud — stream camera/sensor data and run AI models on the edge or in the cloud, without managing the infrastructure in between.
• Aerial twins — takeoff(), land(), and hover() record the flight state Cyberwave gates on, so a second takeoff() on an airborne aircraft is refused rather than sent. If you fly a drone without those helpers, record the intent with twin.set_hovering_status(hovering=True, hovering_altitude=2.0) (or client.twins.set_flight_request(twin_uuid, hovering=True)) — not with twins.update(metadata=...), which overwrites the state the aircraft itself reports. See the Drone commands docs.
• GPS positioning — a twin with a GNSS receiver can be placed in a georeferenced environment from its own fixes. Publish them with client.mqtt.update_twin_gps(...) (or publish_gps() from an edge node) and read them with twin.gps.get_fix(). Live only: get_fix() raises in simulation, because no simulation backend produces GPS. Set the environment's geo reference first — see the Geo reference docs.
• Drivers — subclass BaseDriver (or ship cw-driver.yml) so the dashboard Keyboard (Driver) controller lists this twin's commands. Discrete commands fire once; mark locomotion continuous if a held key should repeat. See the Keyboard (Driver) docs.
• Calibration — when a camera is mounted on an arm, cyberwave.calibration measures where it actually sits instead of you entering the offset by eye. See the hand-eye calibration guide.

Demos

Watch the SDK in action with our demos.

Build a natural language voice agent on SO101

Controlling a DJI Mini 4 Pro with the Cyberwave Python SDK

Examples

Runnable scripts live in examples/ and see the examples index for the full list.

Documentation

Full guides and the complete API reference are at docs.cyberwave.com (overview · API reference).

Listing recordings

recordings.list() is paged. With no start/end it lists the most recent day that has recordings instead of the environment's whole history, and it returns at most limit rows (default 200, fetched 50 per request). Pass limit=0 to follow every page.

Like the replay picker, listing excludes materializing or failed recordings by default. Pass include_unready=True only when a caller needs those rows — it requires elevated access, and the server rejects the call with HTTP 403 rather than silently returning ready rows.

items = cw.environments.recordings.list(environment_id="acme/envs/floor")
items = cw.environments.recordings.list(
    environment_id="acme/envs/floor",
    start="2026-07-01",
    end="2026-07-05",
    limit=0,
)

Whenever the result is partial — scoped to one day, or cut short by limit while the server still had pages — list() logs a warning naming the window and telling you which argument widens it, so a truncated list never looks complete.

If Cloud Run rejects a catalog response at its payload-size boundary, the SDK raises RecordingPayloadTooLargeError with the affected window, cloud trace, and a concrete retry hint. Restrict start/end or lower limit and retry.

Migrating to 0.7.0

recordings.list() with no arguments returns the most recent day that has recordings, rather than the environment's whole history. An unbounded listing could exceed the API gateway's response ceiling on a busy environment and fail with an opaque HTTP 500. It affects twin.recordings.list() and cw.environments.recordings.list() equally.

The behavior itself shipped in 0.6.6. 0.7.0 adds no further change to it — the version is bumped to a minor purely to label the break, which 0.6.6 should have done. If you are pinned to 0.6.6 you already have the new behavior; if you are on <= 0.6.5, the table below is your upgrade.

To widen the window again, name it explicitly and lift the cap. limit=0 alone is not enough, because the implicit single-day window is applied first, and both bounds are required together:

items = cw.environments.recordings.list(
    environment_id="acme/envs/floor",
    start="2026-01-01",           # both bounds are required together
    end="2026-07-05",
    limit=0,                      # follow every page in that window
)

There is no argument that means "all history": pick a start early enough to cover the range you care about. Callers that only need "the latest recordings" need no change — that is the default.

include_unready=True requires elevated access. Without it the server rejects the call with HTTP 403 rather than quietly returning ready rows, so do not add it speculatively.

Recording readiness

Recording list items expose the server's playback assessment when it is available. Use is_playback_ready before downloading artifacts, or request unready entries explicitly while building a retry UI:

items = cw.environments.recordings.list(
    environment_id="acme/envs/floor",
    include_unready=True,
)

for item in items:
    print(item.uuid, item.readiness, item.is_playback_ready)

item.readiness is None when connected to a server that predates this feature. If get() receives a materializing response, it raises CyberwaveError with the server's suggested retry interval instead of returning an empty recording.

Timestamped virtual camera sources

VirtualCameraStreamer accepts RGB arrays as before. For an asynchronous renderer or camera, return a CapturedVideoFrame from get_frame so pixels retain their acquisition time when the stream sends the same image again:

from cyberwave.sensor import CapturedVideoFrame

# In your acquisition callback, with the source's measured clocks:
latest = CapturedVideoFrame(rgb, capture_wall_time, capture_monotonic, acquisition_id)

# Pass this callback as get_frame to VirtualCameraStreamer.
def get_frame():
    return latest

Create the sample once per acquisition and increase acquisition_id even when the scene is stationary. Use Unix seconds for capture_wall_time and monotonic seconds for capture_monotonic, measured when the source state is acquired, not when rendering or encoding finishes. The sample owns a read-only RGB copy. Return the same sample until a new acquisition, or None while unavailable. Repeated sends are not fresh captures; placeholders have no capture timestamp. This preserves source timing, but does not by itself verify recording/Replay synchronization.

Fetching a recording

get() downloads a recording's artifacts into a temp directory. A long recording is stored as many parts; they are fetched in parallel and handed back in timeline order whenever the server labels each part's position:

with cw.environments.recordings.get(items[0]) as rec:
    rec.local_paths          # downloaded parts per stream, in playback order

# Fetch serially, or fetch less
rec = cw.environments.recordings.get(items[0], max_workers=1)
rec = cw.environments.recordings.get(items[0], path=".parquet")

max_workers defaults to 8 and is capped at 32; pass 1 for a strictly serial fetch. The gain scales with how many parts a recording has — one stored as a single large file gains nothing from extra workers.

path keeps only the artifacts whose signed URL or filename contains the given text, so it selects by file extension rather than by stream: ".parquet" fetches the tables and skips .mp4 video, which is usually most of a camera recording's bytes. A reader whose artifact was filtered out raises rather than returning empty.

Compatibility notes for the next release

depth_to_colored_pointcloud now checks its declared encoded-depth input: with output_mode="normalized_uint16" (the default) or "metric_mm", the array must have NumPy uint16 storage. Implicit conversion from an integer list or another dtype is no longer accepted. Validate the encoded values and convert them to uint16 before calling; do not reinterpret metre-valued floats as encoded depth. depth_array_to_meters accepts metre-valued float arrays with output_mode=None. The encoding alone does not establish metric calibration or a world-space pose.

Host ZENOH_MODE settings are validated only when Zenoh is selected and enabled. An unused host value no longer prevents filesystem or MQTT-only drivers from starting. Active Zenoh sessions still require a supported peer or client mode.

Contributing

Contributions are welcome. Please open an issue or a pull request.

Support

• Documentation: docs.cyberwave.com
• Issues: GitHub Issues
• Community: Discord

License

Released under the MIT License.

Asset kits

Create reusable robot/accessory assemblies with client.assets.create_kit(name, components, workspace_uuid=...). Components specify key, asset_uuid, optional parent_key and attach_to_link, position in metres [x,y,z], and quaternion rotation [w,x,y,z]. List parents before children, with exactly one root and 2–50 individual assets. Use get_kit(uuid) or update_kit(...) to inspect or replace the definition. instantiate_kit(uuid, environment_uuid) atomically creates and returns all docked twins; edits do not affect existing deployments. When updating, omit description or visibility to preserve its current value. See Asset kits.