Menlo Robot SDK

uv add "menlo-robot-sdk[livekit]"      # uv
pip install "menlo-robot-sdk[livekit]"  # pip

python -c "import menlo_robot_sdk; print(menlo_robot_sdk.__version__)"

Your code → menlo-robot-sdk ── HTTP ──→ api.menlo.ai/rcs   (control plane: robots, sessions, tokens)
                     │
                     └────── LiveKit SFU (runtime.* RPC) ──→ robot runtime   (drive, state, vision)

export MENLO_API_KEY=sk_live_...
export MENLO_RCS_URL=https://api.menlo.ai/rcs

from menlo_robot_sdk import AsyncClient

client = AsyncClient()                          # reads MENLO_RCS_URL + MENLO_API_KEY
client = AsyncClient(rcs_url=..., api_key=...)  # or pass explicitly

from menlo_robot_sdk import AsyncClient, ConnectCallbacks, connect

async with AsyncClient() as client:
    session = await connect(
        client,
        "rb_myrobot",
        worker_names=[],                   # or ["sim-worker"] to dispatch a server-side worker
        rcw_identity_prefix="simplesim",   # how to find the browser runtime in the room
        join_livekit=True,
        callbacks=ConnectCallbacks(on_action_result=lambda e: print(e.phase, e.action_id)),
    )

    skills = await session.discover_skills()
    status = await session.state.get("robot_status")

    result = await session.invoke("set_velocity", {"wz": 0.8, "duration_s": 2.0})
    # invoke() blocks until the action resolves and returns the terminal ActionResult:
    #   result.status          "done" | "failed"
    #   result.error           code/message when failed (e.g. NAVIGATION_STUCK, CANCELLED)
    #   result.meta.action_id  the action id

    await session.cancel_action("cancel-active")   # cancel the active action
    await session.disconnect()                     # closes the room + deletes the session

result = await session.invoke("set_upper_body", {
    "r_arm": [0.35, 0.18, 0.0, -0.87, 0.0],
    "l_arm": [-0.35, -0.18, 0.0, 0.87, 0.0],
})

R_STAND = [0.35, 0.18, 0.0, -0.87, 0.0]
L_STAND = [-0.35, -0.18, 0.0, 0.87, 0.0]

import asyncio
import contextlib
import logging

R_STAND = [0.35, 0.18, 0.0, -0.87, 0.0]
L_STAND = [-0.35, -0.18, 0.0, 0.87, 0.0]

HEARTBEAT_S = 0.1     # 10 Hz — must stay under the 200 ms trajectory timeout


class ArmHold:
    """Re-sends the latest arm pose at 10 Hz so the edge watchdog stays fed."""

    def __init__(self, session, r_arm, l_arm):
        self._session = session
        self._pose = (list(r_arm), list(l_arm))
        self._task = None
        self.failure = None      # set if the heartbeat dies; check it in your loop

    def set(self, r_arm, l_arm):
        self._pose = (list(r_arm), list(l_arm))

    async def _run(self):
        while True:
            r_arm, l_arm = self._pose
            try:
                await self._session.invoke(
                    "set_upper_body", {"r_arm": r_arm, "l_arm": l_arm}
                )
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                # A dead heartbeat means the arms go stale, so make it loud rather
                # than letting the task die silently. Retry — a single dropped RPC
                # shouldn't end the session.
                self.failure = exc
                logging.exception("arm heartbeat failed")
            await asyncio.sleep(HEARTBEAT_S)

    async def __aenter__(self):
        self._task = asyncio.create_task(self._run())
        return self

    async def __aexit__(self, *_):
        # Await the cancellation before the final command, or the two invokes race
        # and the arms may not end up standing.
        self._task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await self._task
        await self._session.invoke(
            "set_upper_body", {"r_arm": R_STAND, "l_arm": L_STAND}
        )


async def reach_and_return(session):
    async with ArmHold(session, R_STAND, L_STAND) as arms:
        arms.set([0.9, 0.15, 0.2, -0.3, 0.0], [-0.9, -0.15, 0.2, 0.3, 0.0])   # reach forward
        await asyncio.sleep(3)
        if arms.failure:
            raise RuntimeError("lost the arm heartbeat") from arms.failure

pov = await session.get_vision("pov")       # default 1280×720, quality 0.7
top = await session.get_vision("topdown", width=640, height=480)
assert pov[:3] == b"\xff\xd8\xff"

from menlo_robot_sdk.experimental import generate_room_key, generate_room_key_url

key = await generate_room_key(client, robot_id)                   # the compact key
url = await generate_room_key_url(                                 # ready-to-open link
    client, robot_id, viewer_base_url="https://sim.menlo.ai",      # or MENLO_ROBOT_VIEWER_URL
)