sidekick-sdk

Send a camera frame and an instruction. Get back robot actions.

One API for every robotics foundation model. You ask for a job ("fold the towel"), not a checkpoint, and Sidekick picks a model that can do it, calls it, and falls to another if the first one is down.

pip install sidekick-sdk

One dependency (httpx), so it installs on a robot control computer, or inside a ROS container, without dragging a web framework along.

You will need a key: get one at sidekickrobotics.ai/api.

1. Your first call

Copy this whole block and run it. It works before you have a camera or a robot attached: the image below is a 1x1 placeholder so the call goes through.

from sidekick_sdk import Sidekick

sk = Sidekick(api_key="sk-sidekick-...")

image = ("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADElEQVR42mNgYGAA"
         "AAAEAAHI6uv5AAAAAElFTkSuQmCC")

act = sk.act(
    model="sidekick/auto:manipulation",   # a job, not a checkpoint
    instruction="fold the towel",         # what you want done
    action_space="joint_pos_14",          # what shape of action you can execute
    horizon=4,                            # how many future steps to plan
    observations=[{
        "frames": [{"camera": "primary", "b64": image}],
        "proprio": [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
    }],
)

print(act.route["model"])      # which model actually answered
print(act.usage["cost_usd"])   # what this call cost you
print(act.steps)               # the actions

You should see something like:

physical-intelligence/pi-0
0.03262
[[0.0614, -0.0042, ...], [0.0672, -0.0008, ...], ...]

What each argument means

The observation

This is the part worth reading twice, because getting it wrong produces a confident answer rather than an error.

{
    "frames": [{"camera": "primary", "b64": image}],   # what it sees
    "proprio": [0] * 14,                               # where it is now
}

• frames is a list, because most policies want several camera views at once. Each needs a camera name and one of b64 (base64 image data) or url.
• proprio is the robot's current joint positions. It must have the same number of values that your action space declares. joint_pos_14 means fourteen, which is why there are fourteen zeros above.

2. What you get back

act.steps      # list of timesteps, each a list of joint targets
act.space      # the action space, e.g. "joint_pos_14"
act.route      # which model and provider answered, and whether it fell back
act.usage      # cost_usd, gpu_seconds, action_steps

act.steps is a chunk, not a single command. With horizon=4 you get four timesteps to be issued one after another, act.action["dt_ms"] apart (66 ms for pi0). They are close together because a robot moves smoothly: over 66 ms, no joint moves more than a fraction of a degree.

act.route is the audit trail. It names the model that answered and lists every attempt, including failures, so a fallback is visible rather than silent:

act.route["model"]          # "physical-intelligence/pi-0"
act.route["fallback_used"]  # True if the first choice did not answer
act.route["simulated"]      # True means placeholder output, no GPU ran

Always check route["simulated"] before moving a robot. A simulated response is structurally identical to a real one, and setting provider={"allow_simulated": False} on the call refuses it outright.

3. Say what you care about

One argument changes how the router picks:

sk.act(..., preference="reliable")

For hard limits rather than preferences, use provider:

sk.act(..., provider={"max_latency_ms": 900, "allow_simulated": False})

4. The four things you can ask

These are separate calls on purpose. A grounding model returns pixel coordinates and a policy returns joint targets, so the API will not let one quietly stand in for the other.

g = sk.ground(observations=[obs], instruction="point at every graspable object")
print(g.points)     # [{"label": "mug", "x": 0.42, "y": 0.61}, ...]

Going further

Everything above works with a placeholder image and no robot. The rest of this is what you need once real hardware is involved.

Sending real camera frames

A frame can be a URL, a file path, or base64. sk.observation() accepts any of the three and works out which it is:

obs = sk.observation(image_path="frame.jpg", proprio=robot.joints())
obs = sk.observation(image_url="https://cdn.example/frame.jpg", proprio=...)

For a live camera, encode to base64 yourself. Use JPEG, not PNG: the same 640x480 scene is about 24,000 base64 characters as JPEG against 135,000 as PNG, and you pay that on every frame of every call, so it shows up as latency long before it shows up as bandwidth.

import base64, cv2

cams = {"cam_high": cv2.VideoCapture(0)}       # one entry per view

def grab(name: str) -> str:
    ok, frame = cams[name].read()
    if not ok:
        raise RuntimeError(f"{name}: camera read failed")
    _, buf = cv2.imencode(".jpg", frame, [cv2.IMWRITE_JPEG_QUALITY, 90])
    return base64.b64encode(buf).decode()

obs = sk.observation(cameras={"cam_high": grab("cam_high")},
                     proprio=robot.joints())

Multi-camera rigs pass every view in one dict. Most manipulation policies want several; pi0.5 on ALOHA expects four:

cams = {"cam_high":        cv2.VideoCapture(0),
        "cam_low":         cv2.VideoCapture(1),
        "cam_left_wrist":  cv2.VideoCapture(2),
        "cam_right_wrist": cv2.VideoCapture(3)}

obs = sk.observation(cameras={name: grab(name) for name in cams},
                     proprio=robot.joints())

Two mistakes here fail silently rather than raising:

• Camera names must match what the checkpoint was trained with. Send the wrist view as the head view and you get a well-formed action chunk computed from the wrong scene. The live pi0 and pi0.5 deployments expect cam_high, cam_low, cam_left_wrist and cam_right_wrist; a single frame named primary also works, and is what the first call above sends. For any other checkpoint, take the names from its model card.
• proprio must be the width the action space declares, in the order the policy was trained on.

Connecting your robot

robot in these examples is your object. This SDK talks to the API; it ships no driver and knows nothing about your kinematics, limits or e-stop. It needs two things from you:

class Robot:
    """Wrap whatever you already use: ROS 2, Interbotix, LeRobot, serial."""

    def joints(self) -> list[float]:
        """Current joint positions, matching the action space width and order."""
        ...

    def set_joint_positions(self, targets: list[float]) -> None:
        """Command one timestep. Must return promptly, never block."""
        ...

Play the chunk out one step at a time rather than firing it in one go:

import time

dt = act.action["dt_ms"] / 1000.0        # 0.066 for pi0
for step in act.steps:
    robot.set_joint_positions(step)
    time.sleep(dt)

Check three things before any of it reaches an actuator. Each is a response that is perfectly well-formed and still wrong for your robot:

if act.route.get("simulated"):
    raise RuntimeError("placeholder output, no GPU ran")
if act.space != "joint_pos_14":
    raise RuntimeError(f"got action space {act.space}, this arm is joint_pos_14")
if any(len(s) != 14 for s in act.steps):
    raise RuntimeError("wrong number of joints for this robot")

Control loops that keep up

The sleep loop above is fine for one move and wrong for sustained control: it stops commanding while the next chunk is fetched. You cannot call a cloud API once per actuation. A control loop runs at 30 to 200 Hz and a policy answers in a few hundred milliseconds, so asking per tick leaves the robot standing still.

ActionStream solves this. It plays the current chunk from a local buffer while fetching the next one on a background thread:

stream = sk.stream_actions(
    model="sidekick/auto:manipulation",
    instruction="fold the towel",
    action_space="joint_pos_14", dof=14,
    preference="fastest",
    observe=lambda: sk.observation(cameras={n: grab(n) for n in cams},
                                   proprio=robot.joints()),
)

stream.warm_up()          # first call, while the arm is still braked
with stream:              # starts the background thread
    while running:
        step = stream.next_action()
        if step is None:
            robot.hold()  # policy fell behind: decelerate, never repeat
        else:
            robot.set_joint_positions(step)
        time.sleep(stream.dt_s)

Three things to know:

• next_action() never blocks and never raises. All network work is on the background thread, so your servo loop stays real-time.
• None is a real answer, not an error. The buffer is dry or the chunk has gone stale. Hold or decelerate. Repeating the last command forever is a robot acting on a world that has moved on.
• It runs the three safety checks for you and raises UnsafeResponse on a simulated route, a wrong action space, or a step that is not dof wide.

stream.calls, stream.failures and stream.spend_usd track what it has done. A full runnable loop is in examples/control_loop.py.

Finding models, and checking cost first

sk.routes()                        # aliases that are live right now
sk.models(domain="manipulation")   # the catalog
sk.model("physical-intelligence/pi-0")
sk.taxonomy()                      # every valid domain, family, action space
sk.action_space_dims("joint_pos_14")   # -> 14
sk.usage()                         # your spend so far
sk.key()                           # what this key is allowed to do

Call sk.taxonomy() rather than hard-coding action-space or domain strings.

preview_route() is a free dry run: it returns the ranked plan and the price without running any model.

plan = sk.preview_route(model="sidekick/auto:manipulation", contract="act")
print(plan[0]["estimated_list_usd"])   # the published rate
print(plan[0]["estimated_cost_usd"])   # what it would cost right now

Those two differ when the model is cold: waking a GPU costs real money and so does the minute it stays warm, and that is charged to the call that caused it. Run it twice and the second number drops.

Configuration

Sidekick(api_key=..., base_url=..., timeout=120.0, max_retries=2)

base_url defaults to $SIDEKICK_ORIGIN, then to the public gateway, so pointing at a private deployment needs no code change. The client is also a context manager, which closes the connection pool on exit:

with Sidekick(api_key="sk-sidekick-...") as sk:
    ...

When something goes wrong

SidekickError means the call did not succeed. It carries .status, .code and .body.

from sidekick_sdk import Sidekick, SidekickError, UnsafeResponse

try:
    act = sk.act(...)
except SidekickError as e:
    print(e.status, e.code)     # 409 no_route_available

Common ones:

UnsafeResponse is separate on purpose. It means the call succeeded and the answer is wrong for this robot, which is the more dangerous case because nothing about the response looks broken.

Contributing

Issues and pull requests are welcome. See CONTRIBUTING.md.

License

Apache-2.0. See LICENSE.