menlo-robot-sdk 0.2.0

Python client for the Menlo Robot Control Service (RCS) control plane.

menlo-robot-sdk

Python client for the Menlo Robot Control Service (RCS): robot CRUD, session lifecycle, and runtime skill calls through RCS.

Alpha: this package is in active development and may introduce breaking API changes between minor versions.

Install

uv add menlo-robot-sdk

The base package covers the RCS HTTP control plane. Runtime skills, state, and camera capture use LiveKit SFU RPC (runtime.*) when join_livekit=True. Install the LiveKit extra for that path:

uv add "menlo-robot-sdk[livekit]"

Configure

Local development against RCS uses an unsigned development identity:

from menlo_robot_sdk import DevIdentity, MenloSettings

settings = MenloSettings(
    rcs_url="http://localhost:8002",
    identity=DevIdentity(
        sub="user_dev",
        organization_id="org_dev",
        team_id="team_dev",
    ).to_header(),
)

You can also load settings from environment variables:

export MENLO_RCS_URL=http://localhost:8002
export MENLO_IDENTITY='{"sub":"user_dev","organization_id":"org_dev","team_id":"team_dev","roles":[],"scopes":[]}'

Alternatively set MENLO_API_KEY (an sk_live_… key from the console) and point MENLO_RCS_URL at the platform-auth edge (https://platform-auth.menlo.ai/rcs) — no identity needed; the edge mints it.

Quick Example

import asyncio

from menlo_robot_sdk import AsyncClient, ConnectCallbacks, DevIdentity, MenloSettings, connect


async def main() -> None:
    settings = MenloSettings(
        rcs_url="http://localhost:8002",
        identity=DevIdentity(
            sub="user_dev",
            organization_id="org_dev",
            team_id="team_dev",
        ).to_header(),
    )

    events = []

    def on_action_result(event) -> None:
        events.append(event)

    async with AsyncClient(settings=settings) as client:
        created = await client.robots.create(name="sdk-demo", model="asimov-v0")
        session = None
        try:
            session = await connect(
                client,
                created.robot.id,
                worker_names=["sim-worker"],
                join_livekit=False,
                callbacks=ConnectCallbacks(on_action_result=on_action_result),
            )

            skills = await session.discover_skills()
            status = await session.state.get("robot_status")
            # On-demand JPEG from rcw (requires join_livekit=True and a viewer in the room):
            # jpeg = await session.get_vision("pov")
            result = await session.invoke(
                "set_velocity",
                {"wz": 0.8, "duration_s": 2.0},
            )

            print([skill.name for skill in skills], status.runtime.status, result.status)
        finally:
            if session is not None:
                await session.disconnect()
            await client.robots.delete(created.robot.id)


asyncio.run(main())

The runtime skill calls above require a running local RCS stack and browser viewer.