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-sdkOne 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.
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 actionsYou should see something like:
physical-intelligence/pi-0
0.03262
[[0.0614, -0.0042, ...], [0.0672, -0.0008, ...], ...]
| Argument | What to put there |
|---|---|
model |
"sidekick/auto:manipulation" asks for any live manipulation model. You can also name one directly, e.g. "physical-intelligence/pi-0". |
instruction |
Plain English. The policy was trained on language, so say what you would say to a person. |
action_space |
The shape of action your robot can execute. joint_pos_14 is 14 joint angles, 7 per arm. Get the full list with sk.taxonomy(). |
horizon |
How many future timesteps to plan. 4 is fine to start; 16 is normal for real control. |
observations |
What the robot can see and feel right now. See below. |
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
}framesis a list, because most policies want several camera views at once. Each needs acameraname and one ofb64(base64 image data) orurl.propriois the robot's current joint positions. It must have the same number of values that your action space declares.joint_pos_14means fourteen, which is why there are fourteen zeros above.
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_stepsact.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 ranAlways 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.
One argument changes how the router picks:
sk.act(..., preference="reliable")| Value | Picks by |
|---|---|
"balanced" |
a blend of the three below (the default) |
"fastest" |
lowest measured latency |
"cheapest" |
lowest price |
"reliable" |
highest success rate |
For hard limits rather than preferences, use provider:
sk.act(..., provider={"max_latency_ms": 900, "allow_simulated": False})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.
| Call | The question | You get back |
|---|---|---|
sk.act(...) |
what should I do next? | an action chunk |
sk.predict(...) |
what happens next? | predicted future frames |
sk.ground(...) |
where is the thing? | points and boxes, 0 to 1 |
sk.evaluate(...) |
is this policy any good? | a benchmark job |
g = sk.ground(observations=[obs], instruction="point at every graspable object")
print(g.points) # [{"label": "mug", "x": 0.42, "y": 0.61}, ...]Everything above works with a placeholder image and no robot. The rest of this is what you need once real hardware is involved.
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_wristandcam_right_wrist; a single frame namedprimaryalso works, and is what the first call above sends. For any other checkpoint, take the names from its model card. propriomust be the width the action space declares, in the order the policy was trained on.
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")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.Noneis 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
UnsafeResponseon a simulated route, a wrong action space, or a step that is notdofwide.
stream.calls, stream.failures and stream.spend_usd track what it has done.
A full runnable loop is in examples/control_loop.py.
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 doCall 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 nowThose 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.
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:
...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_availableCommon ones:
| Status | Meaning | What to do |
|---|---|---|
401 |
bad or missing key | check SIDEKICK_API_KEY |
402 |
out of credits | top up at the console |
404 |
unknown model | check the id against sk.models() |
409 |
no live model fits | loosen provider, or check sk.routes() |
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.
Issues and pull requests are welcome. See CONTRIBUTING.md.
Apache-2.0. See LICENSE.