menlo-rcs-types

Typed Pydantic models for the RCS → S1 (LiveKit dispatch worker) wire contract — the JSON payloads RCS attaches to LiveKit calls that any dispatched worker reads on connect.

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

Install

uv add menlo-rcs-types

Models

DispatchMetadata (menlo_rcs_types.dispatch)

The JSON RCS passes to livekit.create_agent_dispatch(metadata=...). Each worker receives this on its JobContext.job.metadata.

class DispatchMetadata(BaseModel):
    robot_id: str               # the robot this session is for
    user_id: str                # who initiated the session
    scene_id: str | None        # optional scene tag

RoomMetadata (menlo_rcs_types.room)

The JSON RCS writes onto the LiveKit room itself. Visible to every participant; carries the orchestrator's view of room state.

class RoomMetadata(BaseModel):
    robot_id: str
    status: str
    dispatched_agent_names: list[str]
    required_participants: dict[str, RequiredParticipantSlot]
    dispatches: dict[str, DispatchStatusEntry]
    scene_id: str | None
    created_at: datetime

Quick Example

# Worker side — parsing dispatch metadata
from menlo_rcs_types import DispatchMetadata

async def runtime_session(ctx: JobContext) -> None:
    md = DispatchMetadata.model_validate_json(ctx.job.metadata or "{}")
    print(f"Robot: {md.robot_id}, User: {md.user_id}")

Forward Compatibility

All models use model_config = ConfigDict(extra="allow"). New optional fields are safe to add; workers built against older versions will not break.