kachaka-sdk-toolkit

A unified SDK wrapper for Kachaka robots, providing a shared core library, an MCP Server with 88 tools for AI-driven robot control, and a Skill reference document for development-time agents.

Overview

Kachaka robots expose a gRPC API through the kachaka-api SDK. This toolkit wraps that SDK with production-grade infrastructure -- thread-safe connection pooling, automatic retry with exponential backoff, gRPC timeout protection, connection health monitoring, structured error responses, background camera/transform streaming, and two-tier device-info caching -- so that every consumer (MCP Server, application scripts, patrol systems) shares the same battle-tested code path.

The project follows a layered architecture: a core library (kachaka_core) handles all robot communication, while the MCP Server and Skill document are thin layers on top.

Architecture

graph TD
    subgraph Consumers
        MCP["MCP Server<br/>(88 tools, stdio)"]
        SKILL["Skill .md"]
        APP["Your Script<br/>or App"]
    end

    subgraph kachaka_core["kachaka_core (shared)"]
        CONN["connection.py<br/>Pool mgmt · Auto health monitor<br/>Keepalive · Resolver · Two-tier cache"]
        INT["interceptors.py<br/>TimeoutInterceptor<br/>5s default · 300s long-poll watchdog"]
        CMD["commands.py — KachakaCommands<br/>Fire-and-accept wrappers<br/>Movement · Shelf · Speech · Torch · Map"]
        QRY["queries.py<br/>Status · Camera intrinsics<br/>ToF · Locations · Map"]
        ERR["error_handling.py<br/>@with_retry<br/>Exponential backoff"]
        CAM["camera.py<br/>CameraStreamer (daemon thread)<br/>Detection overlay · Stats"]
        DET["detection.py<br/>ObjectDetector (on-device)<br/>Bbox annotation (PIL)"]
        CTRL["controller.py — RobotController<br/>Production movement (non-blocking + deadline)<br/>Background state polling · Metrics"]
        TF["transform.py<br/>TransformStreamer<br/>Dynamic TF · Auto-reconnect"]
    end

    SDK["kachaka-api SDK (gRPC)<br/>KachakaApiClient → Robot :26400"]

    MCP --> CONN
    MCP --> CMD
    MCP --> QRY
    MCP --> DET
    MCP --> CTRL
    MCP --> TF
    SKILL --> kachaka_core
    APP --> kachaka_core
    CONN --> INT
    CONN --> ERR
    CMD --> ERR
    QRY --> ERR
    DET --> ERR
    CTRL --> CONN
    CAM --> DET
    INT --> SDK
    kachaka_core --> SDK

Features

Connection pooling -- KachakaConnection.get(ip) returns a cached, thread-safe connection. Same IP always yields the same instance.
gRPC timeout protection -- TimeoutInterceptor injects a 5-second default deadline on every unary gRPC call, plus a bounded 300-second watchdog on cursor-based long-polls (which the upstream SDK leaves unbounded). No call can hang forever (raw gRPC blocks 15--18 minutes without a deadline).
Connection monitoring (on by default) -- KachakaConnection.get() auto-starts a background health-check thread that detects disconnection within ~7 seconds and exposes a ConnectionState (UNKNOWN / CONNECTED / DISCONNECTED) with callbacks. The state never lies: it reads UNKNOWN until the first ping proves otherwise.
HTTP/2 keepalive -- silent network drops (WiFi vanishing, no TCP RST) are detected at the transport layer, so wedged in-flight calls fail fast (~20 s on Linux via TCP_USER_TIMEOUT, measured 60 s on macOS) instead of hanging for 15+ minutes.
Fire-and-accept commands -- KachakaCommands movement/shelf wrappers return as soon as the robot accepts the command, bypassing the SDK's unbounded blocking long-poll. Every accepted command returns its command_id; completion is driven by poll_until_complete() (command_id-verified) or RobotController — both timeout-protected.
Two-tier device-info cache -- Tier 1 (permanent): serial, version, error_definitions. Tier 2 (semi-static, manually refreshable): shortcuts, maps, map images. Reduces redundant gRPC round-trips for data that rarely changes.
Disconnect-aware components -- Both RobotController and CameraStreamer skip gRPC calls while ConnectionState.DISCONNECTED, avoiding wasted timeout cycles during network outages.
Automatic retry -- Transient gRPC errors (UNAVAILABLE, DEADLINE_EXCEEDED, RESOURCE_EXHAUSTED) are retried with exponential backoff. Non-retryable errors fail immediately.
Unified response format -- Every method returns {"ok": True, ...} or {"ok": False, "error": "...", "retryable": ...}.
Name + ID resolver -- Commands accept location/shelf names or IDs interchangeably. The resolver patches the upstream SDK to support both.
Background camera streaming -- CameraStreamer runs a daemon thread for continuous JPEG capture without blocking the main loop. Optional detection overlay draws bounding boxes on frames.
Camera intrinsics + ToF -- Query calibration parameters (focal length, principal point, distortion) for front/back/ToF cameras. Retrieve 16-bit ToF depth images.
Object detection -- ObjectDetector wraps the on-device detector (person, shelf, charger, door) and can annotate frames with bounding boxes via PIL.
Transform streaming -- TransformStreamer consumes the GetDynamicTransform server-streaming RPC with auto-reconnect and backoff, storing latest transforms for thread-safe retrieval.
Enriched error messages -- Failed commands include human-readable error descriptions fetched from the robot firmware.
Map management -- Export, import, switch, and create maps from ROS-style PNG occupancy grids. import_image_as_map uses gRPC stream_unary directly for chunked image upload.
Torch control -- Set front/back LED torch intensity (0--255) for illumination.
Laser scan -- Activate on-demand LiDAR scans for a configurable duration.
MCP Server -- 88 tools exposing the full API surface to Claude Desktop, Claude Code, or any MCP client. Installed via the [mcp] extra so host projects that only need kachaka_core stay free of the MCP/starlette dependency stack.
Skill document -- A self-contained reference (skills/kachaka-sdk/SKILL.md) for development-time LLM agents.

Tech Stack

Component	Version
Python	>= 3.10, < 3.13
kachaka-api	>= 3.17 (3.18.1 for `depart_from_charger` / `switch_map` inherit method)
grpcio	>= 1.66
mcp[cli]	>= 2, < 3 (optional `[mcp]` extra — MCP server only; ported to the mcp 2.x `MCPServer` API)
Pillow	>= 10.0
pytest	>= 9.0 (dev)
pytest-mock	>= 3.15 (dev)
Build system	setuptools >= 68 + setuptools-scm >= 8
Package manager	uv

kachaka-api >= 3.17 is the floor for the whole toolkit (the Sound API needs it). depart_from_charger and switch_map's docking_state_inherit_method need kachaka-api 3.18.1; on 3.17 they return ok: False with an explanatory error. Individual features have their own robot-firmware requirements — the authoritative per-feature table is the Feature ↔ Version Matrix in skills/kachaka-sdk/SKILL.md, which also explains why the package version and the firmware version are not the same number. Per-release firmware history (behaviour changes, bugfixes, Pro-only features, app settings from 1.0 onward) is in skills/kachaka-sdk/references/firmware-release-notes.md.

Getting Started

Prerequisites

Python 3.10--3.12
uv (recommended) or pip
A Kachaka robot accessible on the network (default gRPC port: 26400)

Installation

Option A: Claude Code Plugin (recommended)

/plugin marketplace add sigmarobotics/kachaka-sdk-toolkit
/plugin install kachaka

This installs the MCP server and skill automatically via uvx -- no local clone needed.

Option B: Local Development

git clone https://github.com/Sigma-Snaken/kachaka-sdk-toolkit.git
cd kachaka-sdk-toolkit
pip install -e ".[mcp]"
kachaka-setup

Host projects that only use kachaka_core (FastAPI apps, patrol scripts) should install plain kachaka-sdk-toolkit — without the [mcp] extra it does not pull in the MCP/sse-starlette/starlette stack, so it cannot conflict with your web framework's pins.

kachaka-setup registers the MCP Server and Skill with Claude Code automatically. To remove:

kachaka-setup uninstall

For development dependencies: pip install -e ".[dev]"

Quick Start

from kachaka_core import KachakaConnection, KachakaCommands, KachakaQueries

# 1. Connect (port 26400 is appended automatically)
conn = KachakaConnection.get("192.168.1.100")
print(conn.ping())
# {"ok": True, "serial": "KCK-XXXX", "pose": {"x": 1.2, "y": 0.5, "theta": 0.0}}

# 2. Query robot state
queries = KachakaQueries(conn)
status = queries.get_status()
print(f"Battery: {status['battery']['percentage']}%")

locations = queries.list_locations()
for loc in locations["locations"]:
    print(f"  {loc['name']} ({loc['id']})")

# 3. Send commands
cmds = KachakaCommands(conn)
cmds.move_to_location("Kitchen")
result = cmds.poll_until_complete(timeout=60.0)
print(f"Arrived in {result['elapsed']}s")

# 4. Speak
cmds.speak("Hello, I have arrived")

Core API Reference

KachakaConnection

Thread-safe, pooled gRPC connection manager with built-in timeout protection, connection monitoring, and two-tier device-info caching.

from kachaka_core import KachakaConnection, ConnectionState

conn = KachakaConnection.get("192.168.1.100")     # Get or create pooled connection
conn = KachakaConnection.get("192.168.1.100:26400")  # Explicit port (same instance)

conn.ping()             # -> {"ok": True, "serial": "...", "pose": {...}}
conn.ensure_resolver()  # Initialize name-to-ID resolver (idempotent)
conn.client             # Raw KachakaApiClient for direct SDK access
conn.state              # Real-time ConnectionState (monitoring auto-starts)
conn.connection_info()  # {"state": ..., "monitoring": ..., "last_ok_ping_ago_s": ...}

KachakaConnection.remove("192.168.1.100")  # Remove from pool
KachakaConnection.clear_pool()             # Drop all connections (for tests)

Timeout Protection

Every gRPC call is automatically wrapped by TimeoutInterceptor. Detection is cursor-aware: a GetRequest with metadata.cursor != 0 asks the server to hold the call until new data arrives (long-poll), so it gets a bounded 300-second watchdog instead of the 5-second default. Everything else — immediate reads, StartCommand, all other RPCs — gets the 5-second deadline. Without this, a call to an unreachable robot blocks for 15--18 minutes (TCP retransmission timeout), and a long-poll whose completion event is lost hangs forever (observed in production: 82 minutes). Both timeouts are configurable:

conn = KachakaConnection.get("192.168.1.100", timeout=10.0, long_poll_timeout=120.0)

The channel also carries HTTP/2 keepalive options (keepalive_time_ms=15000, keepalive_timeout_ms=5000, pings permitted without data and without active calls), so a silently dead transport is declared dead and re-dialled instead of wedging every subsequent call.

Connection Monitoring

Monitoring starts automatically in KachakaConnection.get() — the health-check thread pings the robot every 5 seconds (first ping immediately) and maintains conn.state. Calling start_monitoring() again is safe: it registers a callback in place and retunes the cadence without losing state.

conn = KachakaConnection.get("192.168.1.100")          # monitoring already on
conn.wait_until_known(timeout=10.0)                     # block until first ping verdict
print(conn.state)                                       # CONNECTED / DISCONNECTED

def on_state_change(new_state: ConnectionState):
    print(f"Connection: {new_state.name}")

conn.start_monitoring(interval=3.0, on_state_change=on_state_change)  # retune + callback
conn.wait_for_state(ConnectionState.CONNECTED, timeout=30.0)
conn.connection_info()  # {"state": "connected", "monitoring": True,
                        #  "state_changed_ago_s": 42.0, "last_ok_ping_ago_s": 1.2}

Opt out with KachakaConnection.get(ip, monitor=False) — conn.state then stays UNKNOWN.

Both RobotController and CameraStreamer skip gRPC calls entirely while disconnected instead of wasting 5 seconds per call on the interceptor timeout.

Disconnect Behaviour (measured on real hardware)

Event after a silent network drop	Latency	Mechanism
`conn.state` flips to DISCONNECTED	~7 s (interval 5 s + ping timeout)	health-check ping fails
Immediate-read RPC fails	≤ 5 s	per-call deadline
In-flight long-poll released	≤ 60 s on macOS, ~20 s on Linux	HTTP/2 keepalive kills dead transport
Worst-case long-poll release	≤ 300 s	`long_poll_timeout` watchdog
Channel re-dials after network returns	≤ ~60 s after restore	keepalive declares zombie transport dead, gRPC reconnects

Robot firmware note (3.16.x, measured): the server cancels a held long-poll after ~35--40 s with CANCELLED — independent of keepalive. The toolkit's retry/poll layers absorb this; raw SDK users will see the exception.

Two-Tier Device-Info Cache

Reduces redundant gRPC round-trips for data that rarely changes:

Tier	Data	Lifetime	Refresh
1 (permanent)	serial, version, error_definitions	Never expires	Fetched once on first access
2 (semi-static)	shortcuts, map_list, current_map_id, map_image	Manual	Invalidated by `switch_map()` or explicit refresh

KachakaCommands

Robot action commands. All methods return dict with an ok key. All are decorated with @with_retry except the stop family (stop_manual_drive / deprecated stop / set_emergency_stop), which must act immediately rather than retry.

Fire-and-accept contract (since 0.6.0): movement and shelf commands return as soon as the robot accepts the command — {"ok": True} means accepted, not completed. This bypasses the SDK's blocking long-poll, which has no client deadline and hangs indefinitely if the completion event is lost (observed in production: 82 minutes). Drive completion with poll_until_complete(timeout=...), or use RobotController for supervised execution — it polls with a deadline and always returns within timeout= seconds. (speak() still blocks until the utterance finishes, bounded by long_poll_timeout.) KachakaCommands remains the home of actions RobotController does not expose (TTS, volume, torch, map switch, shortcuts, etc.).

command_id verification (since 0.7.0): every accepted command includes its command_id in the result dict and tracks it on the instance. poll_until_complete() verifies completion against that id. This matters because the robot's idle state (PENDING + empty command_id) is indistinguishable from the registration window right after StartCommand — an unverified poll can report completion in ~0 s while the command is still queuing, and a follow-up command then cancels the queued one (2026-07-07 field incident).

Method	Description
`move_to_location(name, source_location_name="")`	Move to a registered location by name or ID. Never target a location a shelf is parked on (incl. its home) -- shelves are invisible to the LiDAR on final approach; the robot collides silently. Optional `source_location_name` (3.16.1+) forces the planner to treat the named location as the route's starting point
`move_to_pose(x, y, yaw)`	Move to absolute map coordinates
`move_forward(distance_meter, speed=0.1, mute_sensors=False)`	Move forward (positive) or backward (negative). Default `speed` is `0.1` m/s -- 3.16 firmware rejects `speed=0.0` with error 15508 (the vendor fixed the no-speed error in 3.17.5; what 0.0 does there is unverified), so calls without an explicit speed succeed on every firmware. Firmware before 3.18.1 may rotate at the end of the move. Set `mute_sensors=True` (3.16.1+) to bypass safety sensors for rescue/recovery when the robot is wedged; collision detection is suppressed -- use with care
`move_by_velocity_muted(signed_velocity, duration_sec)`	Drive at the given m/s for the given seconds with safety sensors muted the entire time (3.16.1+). First-class rescue command. Velocity clamped to [-0.3, 0.3] m/s, duration to [0, 30] s
`rotate_in_place(angle_radian)`	Rotate in place (positive = counter-clockwise)
`return_home()`	Return to charger
`depart_from_charger()`	Drive slightly forward off the charging dock; no-op when not docked (firmware 3.18.1+)
`move_shelf(shelf, location)`	Pick up shelf and deliver to location
`return_shelf(shelf_name="")`	Return shelf to its home location
`dock_shelf()`	Engage the shelf directly in front of the robot (does not travel to it; position the robot first)
`undock_shelf()`	Put the carried shelf down here and drive out from under it
`dock_any_shelf_with_registration(location, dock_forward)`	Move to location, dock any shelf there (auto-registers new shelves)
`reset_shelf_pose(shelf_name)`	Reset recorded pose of a shelf
`start_shortcut(shortcut_id)`	Execute a registered shortcut by ID
`switch_map(map_id, *, pose_x, pose_y, pose_theta, inherit_docking_state, docking_state_inherit_method)`	Switch active map (invalidates Tier 2 cache). `inherit_docking_state` keeps a docked shelf (Pro); `docking_state_inherit_method` is `"unspecified"` / `"shelf_id_based"` / `"fiducial_id_based"`
`export_map(map_id, output_path)`	Export map to binary file (Kachaka proprietary format). Can fail with 12117 while moving -- dock and retry
`import_map(file_path)`	Import a previously exported map backup
`import_image_as_map(image_path, resolution, ...)`	Import ROS-style PNG occupancy grid as a new map (firmware 3.14.4+; the vendor describes the output as a Kachaka Pro map)
`speak(text)`	Text-to-speech
`set_speaker_volume(volume)`	Set volume 0--10 (clamped)
`set_front_torch(intensity)`	Set front LED torch intensity (0--255)
`set_back_torch(intensity)`	Set back LED torch intensity (0--255)
`activate_laser_scan(duration_sec)`	Activate on-demand LiDAR scan
`set_auto_homing(enabled)`	Enable/disable automatic return-to-charger
`cancel_command()`	Cancel running command
`proceed()`	Resume a command waiting for confirmation
`set_manual_control(enabled)`	Enable/disable velocity control mode
`set_velocity(linear, angular)`	Send velocity (max 0.3 m/s, 1.57 rad/s)
`stop_manual_drive()`	Stop manual velocity driving. Does not stop autonomous navigation -- use `cancel_command()` for that. (`stop()` is a deprecated alias)
`set_emergency_stop()`	Latched hardware pause (error 21051). Released only by the physical power button -- no gRPC path exists. Not exposed as an MCP tool
`poll_until_complete(timeout, command_id="")`	Block until the tracked command finishes. Completion is verified against the `command_id` of the most recently accepted command on this instance (or an explicit one) — never misled by the idle/registration-window state

KachakaQueries

Read-only status queries. All methods return dict with an ok key. All are decorated with @with_retry.

Method	Description
`get_status()`	Full snapshot: pose, battery, command state, errors, shelf
`get_pose()`	Current position (x, y, theta)
`get_battery()`	Battery percentage and power status
`list_locations()`	All registered locations with pose, `type_name` (charger / shelf home / SLAM marker), and per-location undock flags (`undock_aligning_to_wall`, `undock_avoiding_obstacles`)
`list_locations_digest()`	Lightweight variant (3.16.1+): `[{id, name, type}]` only, no pose. ~58% smaller payload -- intended for picker UIs
`list_shelves()`	All registered shelves with home location, last known `pose` (None if unreported), and `speed_mode`
`list_shelves_digest()`	Lightweight variant (3.16.1+): `[{id, name}]` only, no home location. Intended for picker UIs
`get_moving_shelf()`	ID of currently carried shelf (or null)
`get_command_state()`	Current command execution state
`get_last_command_result()`	Result of most recently completed command
`get_front_camera_image(fresh=True, timeout=5.0)`	Front camera JPEG as base64. `fresh=True` waits for a frame captured after this call — the buffer may still hold a frame from before the robot's last move. `fresh=False` returns the buffered frame immediately
`get_back_camera_image(fresh=True, timeout=5.0)`	Back camera JPEG as base64 (same `fresh` contract)
`get_camera_intrinsics(camera)`	Calibration parameters for front/back/tof camera
`get_tof_image()`	16-bit ToF depth image (16UC1 encoding)
`get_map()`	Current map as base64 PNG with metadata
`list_maps()`	All available maps + current map ID
`get_errors()`	Active error codes
`get_error_definitions()`	All error code definitions from firmware (title, description, `error_type` severity, ref_url)
`get_serial_number()`	Robot serial number
`get_version()`	Firmware version
`get_speaker_volume()`	Current speaker volume (0--10)
`get_auto_homing_enabled()`	Whether auto-homing is enabled
`get_manual_control_enabled()`	Whether manual velocity control is active
`is_ready()`	Non-blocking readiness check
`get_static_transform()`	Static TF frames with quaternion-to-yaw conversion
`list_shortcuts()`	All registered shortcuts
`get_history()`	Command execution history

CameraStreamer

Background daemon thread for continuous camera capture. Proven optimal in connection testing (30--40% lower RTT vs. inline capture, lowest drop rates).

Basic Usage

from kachaka_core import KachakaConnection, CameraStreamer

conn = KachakaConnection.get("192.168.1.100")
streamer = CameraStreamer(conn, interval=1.0, camera="front")
streamer.start()

# Non-blocking frame retrieval in your main loop
frame = streamer.latest_frame  # Thread-safe, returns latest captured dict
if frame and frame["ok"]:
    process(frame["image_base64"])  # base64-encoded JPEG
    print(f"Captured at {frame['timestamp']}")

streamer.stop()
print(streamer.stats)
# {"total_frames": 120, "dropped": 3, "drop_rate_pct": 2.5}

With Callback

def on_new_frame(frame: dict):
    save_to_disk(frame["image_base64"])

streamer = CameraStreamer(conn, interval=0.5, on_frame=on_new_frame)
streamer.start()

Back Camera

streamer = CameraStreamer(conn, camera="back")

With Detection Overlay

streamer = CameraStreamer(conn, interval=1.0, detect=True, annotate=True)
streamer.start()

# latest_frame now includes "objects" key + bbox drawn on image
frame = streamer.latest_frame
# {"ok": True, "image_base64": "...", "objects": [...], "timestamp": ...}

# Detection results separately
detections = streamer.latest_detections
# [{"label": "person", "label_id": 1, "roi": {...}, "score": 0.95, "distance": 2.3}, ...]

detect=True, annotate=False -- raw frame + detection results (no bbox)
detect=True, annotate=True -- annotated frame + detection results
Default detect=False, annotate=False -- unchanged behavior

Properties

Property	Type	Description
`latest_frame`	`dict \| None`	Most recent frame (thread-safe read)
`latest_frame_bytes`	`bytes \| None`	Most recent raw JPEG bytes (no base64 encoding overhead)
`latest_detections`	`list \| None`	Most recent detection results (requires `detect=True`)
`is_running`	`bool`	Whether the capture thread is alive
`stats`	`dict`	`{total_frames, dropped, drop_rate_pct, longest_gap_s, recovery_latency_ms}`

Design Notes

Thread is a daemon -- it will not prevent interpreter exit
stop() signals the thread and waits up to interval * 3 seconds
Errors in capture increment the dropped counter but never crash the thread
Detection failure never blocks frame capture (logged and skipped)
Callback exceptions are caught and logged, never propagated
start() on an already-running streamer is a no-op
stop() on a non-running streamer is a no-op
Disconnect-aware: When ConnectionState.DISCONNECTED, the capture loop sleeps instead of attempting gRPC calls that would each waste 5 seconds on the interceptor timeout
Recovery metrics: Wire streamer.notify_state_change to conn.start_monitoring() to track longest_gap_s and recovery_latency_ms in stats

TransformStreamer

Background daemon thread for streaming dynamic TF (coordinate frame) transforms via the GetDynamicTransform server-streaming RPC. Auto-reconnects with backoff on stream errors.

from kachaka_core import KachakaConnection
from kachaka_core.transform import TransformStreamer

conn = KachakaConnection.get("192.168.1.100")
streamer = TransformStreamer(conn)
streamer.start()

# Thread-safe snapshot of latest transforms
transforms = streamer.latest_transforms
# {"base_link": {"frame_id": "map", "child_frame_id": "base_link",
#   "translation": {"x": 1.2, "y": 0.5, "z": 0.0},
#   "rotation": {"x": 0, "y": 0, "z": 0.1, "w": 0.99},
#   "theta": 0.2, "stamp_nsec": 1234567890}, ...}

print(streamer.stats)
# {"total_updates": 150, "errors": 0, "last_update_time": 1711800000.0}

streamer.stop()

Properties

Property	Type	Description
`latest_transforms`	`dict`	Latest transform per child frame (thread-safe)
`is_running`	`bool`	Whether the streaming thread is alive
`stats`	`dict`	`{total_updates, errors, last_update_time}`

Design Notes

Consumes server-streaming RPC (long-lived connection, push-based)
Quaternion rotation is automatically converted to theta (yaw) for convenience
On stream error, reconnects with exponential backoff
Daemon thread -- auto-exits when the process ends

RobotController

Background state polling + non-blocking command execution with command_id verification and deadline-based timeout. This is the production-recommended path for any caller that cares about timeouts or robustness.

Why prefer this over `KachakaCommands`?

KachakaCommands wraps the upstream Python SDK's convenience methods directly. Those methods drive completion through GetLastCommandResult — a server-side long-poll RPC that is intentionally exempted from the default 10s deadline (kachaka_core/interceptors.py:43-48). If the result-delivery stream is ever silently dropped (publisher race, half-open HTTP/2 stream, cursor mismatch), the call hangs forever. A real incident in production logged an 82-minute hang inside this loop while the robot had actually completed the move — only docker restart freed the thread.

RobotController._execute_command was designed to avoid this entirely:

Behaviour	`KachakaCommands.*`	`RobotController.*`
Completion driver	SDK's blocking `while True: GetLastCommandResult(cursor)` (long-poll)	`GetCommandState` unary read (subject to interceptor's 10s timeout) + sleep
Timeout	None on the long-poll — relies on caller's threading	Deadline-based; always returns within `timeout=` seconds
Disconnect handling	Hangs until TCP RST	Returns `{"error": "DISCONNECTED"}` if `ConnectionState` drops
Return on stall	Caller's thread stays wedged	Returns `{"error": "TIMEOUT"}`
Background state	None — caller must poll separately	`state` snapshot kept fresh by `_state_loop`
Metrics	None	RTT, poll count, success/failure counters

Use KachakaCommands only for simple one-shot scripts where blocking-forever is acceptable (e.g. interactive REPL, quick utility). For production services, patrols, dashboards, or anything driven by an upstream caller that can't be left hanging, use RobotController.

Basic Usage

from kachaka_core import KachakaConnection, RobotController

conn = KachakaConnection.get("192.168.1.100")
ctrl = RobotController(conn)
ctrl.start()  # starts background state polling thread

# Thread-safe state snapshot (updated every fast_interval)
state = ctrl.state
print(state.battery_pct, state.pose_x, state.pose_y, state.is_command_running)

# Non-blocking command execution with polling + command_id verification
result = ctrl.move_to_location("Kitchen", timeout=120)
# {"ok": True, "action": "move_to_location", "target": "Kitchen", "elapsed": 45.2}

result = ctrl.return_home(timeout=60)
result = ctrl.move_shelf("Shelf A", "Meeting Room", timeout=120)
result = ctrl.return_shelf("Shelf A", timeout=60)
result = ctrl.dock_any_shelf_with_registration("Warehouse", timeout=120)

# Free-pose movement — no registered location needed
result = ctrl.move_to_pose(1.5, 2.0, 0.5, timeout=120)
result = ctrl.move_forward(0.3, speed=0.1, timeout=30)
result = ctrl.rotate_in_place(1.57, timeout=15)

# Metrics collected during command execution
m = ctrl.metrics
print(f"polls={m.poll_count}, avg_rtt={sum(m.poll_rtt_list)/len(m.poll_rtt_list):.1f}ms")
ctrl.reset_metrics()

ctrl.stop()

Movement API

All methods return a result dict of the form {"ok": bool, "action": str, "target": str, "elapsed": float, ...}. On failure, error_code and error (with description) are populated. On deadline expiry, {"ok": False, "error": "TIMEOUT", "timeout": <seconds>}.

Method	Notes
`move_to_location(name, *, timeout=120, ...)`	Move to a registered location
`return_home(*, timeout=60, ...)`	Return to charger
`move_shelf(shelf, location, *, timeout=120, ...)`	Pick + deliver a shelf (enables shelf-drop monitor)
`return_shelf(shelf="", *, timeout=60, ...)`	Return a shelf to its home
`dock_any_shelf_with_registration(location, dock_forward=False, *, timeout=120, ...)`	Dock + register an unknown shelf at a location
`move_to_pose(x, y, yaw, *, timeout=120, ...)`	Free-pose move — target is absolute map coords, not a named location. Use this for patrols whose points are stored as raw `(x, y, yaw)`
`move_forward(distance_meter, *, speed=0.1, mute_sensors=False, timeout=60, ...)`	Forward (positive) / backward (negative). `mute_sensors=True` bypasses safety sensors (3.16.1+)
`rotate_in_place(angle_radian, *, timeout=30, ...)`	Rotate (positive = counter-clockwise)

Constructor Parameters

ctrl = RobotController(
    conn,
    fast_interval=1.0,   # pose + command_state poll interval (seconds)
    slow_interval=30.0,   # battery poll interval (seconds)
    retry_delay=1.0,      # delay between retries on StartCommand failure
    poll_interval=1.0,    # delay between GetCommandState polls during execution
)

When to Use

Use Case	Recommended
Production service / patrol / dashboard	`RobotController`
Anything driven by an upstream API where a hung thread is unacceptable	`RobotController`
Free-pose movement (patrol points stored as raw `x, y, yaw`)	`RobotController.move_to_pose`
Background state monitoring (pose, battery, errors)	`RobotController.state`
Need timeout / disconnect handling on a command	`RobotController`
Interactive REPL / one-off utility script	`KachakaCommands`
Need a command that `RobotController` doesn't expose (e.g. `speak`, `set_volume`, `switch_map`, torch / scan / shortcut)	`KachakaCommands`

Command Execution Flow

sequenceDiagram
    participant Controller
    participant Robot

    rect rgb(230, 245, 255)
        Note over Controller, Robot: Phase 1 — StartCommand (with retry until deadline)
        loop Retry until deadline (retry_delay between attempts)
            Controller->>Robot: StartCommand(request)
            alt gRPC success
                Robot-->>Controller: command_id + result
            else gRPC failure
                Note right of Controller: Wait retry_delay, then retry
            end
        end
        alt result.success == false
            Controller->>Controller: get_robot_error_code(result)
            Note right of Controller: Return error immediately
        end
    end

    rect rgb(255, 245, 230)
        Note over Controller, Robot: Phase 2 — Registration Poll (max 5s, 0.2s interval)
        loop Every 0.2s for up to 5 seconds
            Controller->>Robot: GetCommandState()
            Robot-->>Controller: state (command_id, state)
            alt Our command_id with state RUNNING or PENDING
                Note right of Controller: Registration confirmed, proceed
            else Not yet registered
                Note right of Controller: Continue polling
            end
        end
    end

    rect rgb(230, 255, 230)
        Note over Controller, Robot: Phase 3 — Main Polling Loop (until deadline)
        loop Every poll_interval (default 1s) until deadline
            Controller->>Robot: GetCommandState()
            Note right of Controller: Measure RTT for metrics
            Robot-->>Controller: state (command_id, state)
            alt State leaves RUNNING/PENDING OR command_id changes
                Controller->>Robot: GetLastCommandResult()
                Robot-->>Controller: result with command_id
                alt result.command_id matches AND success
                    Note right of Controller: Return ok
                else result.command_id matches AND failure
                    Controller->>Controller: get_robot_error_code(result)
                    Note right of Controller: Return error with description
                end
            end
        end
    end

    rect rgb(255, 230, 230)
        Note over Controller, Robot: Phase 4 — Timeout
        Note over Controller: Deadline reached → return TIMEOUT error
    end

Concurrency notes:

Command preemption: If Command B is issued while Command A is still running, the robot cancels A. Command A receives error_code=10001 (interrupted).
Not thread-safe: _execute_command is not thread-safe. Callers must serialise command execution externally.
Disconnect-aware: The background _state_loop skips gRPC calls while ConnectionState.DISCONNECTED.

ObjectDetector

On-device object detection (person, shelf, charger, door) with optional bounding-box annotation via PIL.

Basic Usage

from kachaka_core import KachakaConnection, ObjectDetector

conn = KachakaConnection.get("192.168.1.100")
det = ObjectDetector(conn)

# Get current detections
result = det.get_detections()
# {"ok": True, "objects": [{"label": "person", "label_id": 1,
#   "roi": {"x": 100, "y": 50, "width": 200, "height": 300},
#   "score": 0.79, "distance": 2.3}, ...]}

# Capture image + detections together
result = det.capture_with_detections(camera="front")
# {"ok": True, "image_base64": "...", "format": "jpeg", "objects": [...]}

# Draw bounding boxes on raw JPEG bytes
import base64
raw = base64.b64decode(result["image_base64"])
annotated = det.annotate_frame(raw, result["objects"])
# Returns annotated JPEG bytes (not base64)

Labels

label_id	Label	Bbox Color
0	unknown	pink
1	person	green
2	shelf	blue
3	charger	cyan
4	door	red

Notes

distance is None when distance_median <= 0 (close range or sensor unavailable)
annotate_frame uses PIL ImageDraw -- does not depend on kachaka_api.util.vision

MCP Server

The MCP Server exposes 88 tools for controlling Kachaka robots through any MCP-compatible client (Claude Desktop, Claude Code, etc.). Each tool is a thin one-liner delegation to kachaka_core.

Running the Server

Requires the [mcp] extra: pip install 'kachaka-sdk-toolkit[mcp]' (or uvx --from 'kachaka-sdk-toolkit[mcp]' kachaka-mcp).

# Console entry point (preferred)
kachaka-mcp

# Or via module
python -m mcp_server.server

Claude Desktop Configuration

Add to your Claude Desktop config.json:

{
  "mcpServers": {
    "kachaka-robot": {
      "command": "python",
      "args": ["-m", "mcp_server.server"],
      "cwd": "/path/to/kachaka-sdk-toolkit"
    }
  }
}

Tool Reference

All tools require an ip parameter (e.g., "192.168.1.100"). Port 26400 is appended automatically.

Connection (3 tools)

Tool	Description
`ping_robot`	Test connectivity, return serial + pose
`get_connection_state`	Real-time connection health: state (connected/disconnected/unknown), monitoring status, last-ping age, time since last state change
`disconnect_robot`	Remove from connection pool

Status Queries (6 tools)

Tool	Description
`get_robot_status`	Full snapshot (pose, battery, command, errors, shelf)
`get_robot_pose`	Current position (x, y, theta)
`get_battery`	Battery percentage + charging status
`get_errors`	Active error codes
`get_error_definitions`	What each error code means, incl. `error_type` severity (live from firmware, ~860 entries)
`get_robot_info`	Serial number + firmware version

Locations and Shelves (5 tools)

Tool	Description
`list_locations`	All locations (name, id, type, pose)
`list_locations_digest`	Lightweight locations list (id, name, type) -- ~58% smaller payload
`list_shelves`	All shelves (name, id, home location)
`list_shelves_digest`	Lightweight shelves list (id, name)
`get_moving_shelf`	Currently carried shelf ID

Movement (7 tools)

All movement tools are fire-and-accept: they return as soon as the robot accepts the command, not on arrival. Use the controller_* tools for blocking moves that return on completion.

Tool	Description
`move_to_location`	Move to a named location. Accepts optional `source_location_name` to force the planner to treat that location as the route's starting point
`move_to_pose`	Move to (x, y, yaw) coordinates
`move_forward`	Move forward/backward by distance. Accepts optional `mute_sensors=True` to bypass safety sensors for rescue/recovery
`move_by_velocity_muted`	Drive at a signed velocity for a fixed duration with sensors muted (rescue command, clamped to ±0.3 m/s and 30 s)
`rotate`	Rotate in place
`return_home`	Return to charger
`depart_from_charger`	Drive slightly forward off the charging dock (firmware 3.18.1+)

Shelf Operations (6 tools)

Tool	Description
`move_shelf`	Pick up shelf, deliver to location
`return_shelf`	Return shelf to home
`dock_shelf`	Engage the shelf in front of the robot (no travel)
`undock_shelf`	Put the carried shelf down here
`dock_any_shelf_with_registration`	Move to location, dock any shelf (auto-registers new)
`reset_shelf_pose`	Reset recorded pose of a shelf

Controller (8 tools)

The controller tools expose RobotController through the MCP server, providing background state polling and command-ID-verified execution for multi-step patrols. start_controller must be called before any other controller tool.

Tool	Description
`start_controller`	Start background state polling (idempotent)
`stop_controller`	Stop and remove controller
`get_controller_state`	Full state snapshot (pose, battery, shelf, command, connection_state, snapshot age, disconnect duration)
`controller_move_to_location`	Move to location via controller (blocks until arrival)
`controller_rotate`	Rotate in place via controller (blocks until done)
`controller_move_shelf`	Move shelf via controller (auto-starts shelf monitor)
`controller_return_shelf`	Return shelf via controller (auto-stops shelf monitor)
`controller_dock_any_shelf`	Dock any shelf at location via controller (auto-registers new)

Speech (3 tools)

Tool	Description
`speak`	Text-to-speech
`set_volume`	Set speaker volume (0--10)
`get_volume`	Get current volume

Custom Sounds (5 tools)

Tool	Description
`list_sounds`	List custom sound clips registered on the robot (id + name)
`add_sound`	Upload a custom sound clip (WAV) — local path or base64 data
`play_sound`	Play an uploaded sound by ID (optional loop)
`stop_sound`	Stop the custom sound currently playing
`delete_sound`	Delete a custom sound clip by ID

Command Control (4 tools)

Tool	Description
`cancel_command`	Cancel running command -- the way to stop a moving robot
`proceed`	Resume a command that is waiting for confirmation
`get_command_state`	Whether a command is running
`get_last_result`	Result of last completed command

Camera (8 tools)

Tool	Description
`capture_front_camera`	Single JPEG from front camera. `fresh=True` (default) waits for a frame captured after the call — never returns a pre-move buffered frame
`capture_back_camera`	Single JPEG from back camera (same `fresh` contract)
`start_camera_stream`	Start background capture (options: `detect`, `annotate`)
`get_camera_frame`	Get latest frame from stream
`stop_camera_stream`	Stop background capture
`get_camera_stats`	Stream statistics (frames, drops)
`get_camera_intrinsics`	Camera calibration parameters (front/back/tof)
`get_tof_image`	16-bit ToF depth image

Object Detection (2 tools)

Tool	Description
`get_object_detection`	Detect objects (person, shelf, charger, door) with scores + distances
`capture_with_detection`	Camera capture with detection overlay (bounding boxes)

Map (6 tools)

Tool	Description
`get_map`	Current map as base64 PNG
`list_maps`	All maps + current map ID
`switch_map`	Switch active map (invalidates Tier 2 cache)
`export_map`	Export map to binary file (Kachaka proprietary format)
`import_map`	Import previously exported map backup
`import_image_as_map`	Import ROS-style PNG occupancy grid as a new map (firmware 3.14.4+)

Shortcuts and History (3 tools)

Tool	Description
`list_shortcuts`	All registered shortcuts
`start_shortcut`	Execute a registered shortcut by ID
`get_history`	Command execution history

Manual Control (4 tools)

Tool	Description
`enable_manual_control`	Enable/disable velocity mode
`get_manual_control`	Read whether velocity mode is currently enabled
`set_velocity`	Set linear + angular velocity
`stop_manual_drive`	Stop manual driving. Does not stop autonomous navigation -- use `cancel_command`

Torch / Lighting (2 tools)

Tool	Description
`set_front_torch`	Set front LED torch intensity (0--255)
`set_back_torch`	Set back LED torch intensity (0--255)

Laser Scan (1 tool)

Tool	Description
`activate_laser_scan`	Activate on-demand LiDAR scan for a configurable duration

Auto Homing (2 tools)

Tool	Description
`set_auto_homing`	Enable/disable automatic return-to-charger
`get_auto_homing`	Query auto-homing status

Readiness (1 tool)

Tool	Description
`is_ready`	Non-blocking readiness check

Localization (2 tools)

Tool	Description
`localize`	Force re-localization on the current map — recover from localization jumps or after the robot was manually moved
`set_robot_pose`	Override the pose estimate with known map coordinates (immediate RPC); seed before `localize` when the true position is known

Recovery (1 tool)

Tool	Description
`restart_robot`	Reboot the robot to clear hardware-fatal errors (e.g. 21004 LiDAR). Use only when `is_ready` reports `recovery_hint="restart_robot"`

Transforms (4 tools)

Tool	Description
`get_static_transform`	Static TF frames (one-shot query)
`start_transform_stream`	Start background dynamic TF streaming
`get_dynamic_transform`	Get latest dynamic transforms from stream
`stop_transform_stream`	Stop transform streaming

Playground (5 tools)

Manage scripts running inside the robot's on-board Docker container (offline mode / 離線模式) — see the Playground Offline Execution section below. All five tools go through JupyterLab :26501 by default (factory password kachaka, no setup) and take a transport parameter: "auto" (default — JupyterLab first, SSH :26500 only when :26501 is unreachable), "jupyter", or "ssh".

Tool	Description
`playground_upload`	Upload a Python script to `/home/kachaka/<filename>` in the container (JupyterLab by default, no SSH key required)
`playground_run`	Run an uploaded script in the background, capturing stdout/stderr to `log_path` (default `/tmp/script.log`)
`playground_stop`	Stop a running script (pass `filename=`)
`playground_log`	Tail the script's log file
`playground_status`	Check whether a script is still running (pass `filename=`)

Playground Offline Execution

俗稱「離線模式」／offline mode／Playground 執行——三者同義。程式經 JupyterLab :26501 傳入容器，在機器人上自主執行，移動全程不依賴 WiFi。

Run scripts directly on the robot's on-board Docker container (Playground) for autonomous operation without network dependency. Scripts use kachaka_api (the raw SDK) -- not kachaka_core -- because the container only has the pre-installed SDK.

Why Playground -- Offline-First Robot Control

In normal mode, your script runs on an external machine and sends gRPC commands to the robot over WiFi. If WiFi drops mid-route, the robot stops receiving commands. In Playground mode, the script runs inside the robot's Docker container and commands travel through a container-internal virtual network (100.94.1.1:26400) that never touches WiFi. The robot can walk into a zero-connectivity zone and keep executing autonomously.

WiFi is only needed for the deploy step: the script is uploaded through JupyterLab :26501 (REST contents API) and started from a Jupyter terminal — factory password kachaka, no SSH key or other setup. The playground_upload / playground_run MCP tools do this for you. SSH :26500 is an optional dev-time alternative for interactive debugging.

graph LR
    subgraph normal["Normal Mode (WiFi required)"]
        PC["Your PC"] -->|"WiFi → :26400"| R1["Robot gRPC"]
    end

    subgraph playground["Playground Mode (offline)"]
        CONT["Robot Container"] -->|"100.94.1.1:26400<br/>(internal bridge)"| R2["Robot gRPC"]
    end

    PC -->|"JupyterLab :26501 (deploy)"| CONT

    style normal fill:#fff3e0
    style playground fill:#e8f5e9

When to Use Playground vs. kachaka_core

Situation	Recommended	Why
Robot stays in WiFi range	`kachaka_core`	Real-time control, richer API, easier debugging
Route passes through WiFi dead zones	Playground	Script survives network loss -- runs on-board
Factory/warehouse with unreliable WiFi	Playground	Cannot guarantee connectivity during movement
Long-running autonomous task (>30 min)	Playground	Even brief WiFi drops can abort external commands
Need operator confirmation without network	Playground + IMU	Physical shake replaces network-based confirmation
Need real-time dashboard or camera stream	`kachaka_core`	Playground cannot push data out without WiFi

Decision rule: If the robot must travel to any location where WiFi may be unavailable -- even briefly during movement -- use Playground.

Container Environment

Constraint	Detail
gRPC address	`100.94.1.1:26400` (container-internal, always reachable)
Resolver	Must call `client.update_resolver()` before name-based commands
Libraries	stdlib only (`json`, `threading`, `time`, etc.) -- no pip
Ports exposed	26400 (gRPC), 26500 (SSH), 26501 (JupyterLab)
Script path	`/home/kachaka/<filename>`
Deploy channel	JupyterLab :26501 contents API + terminal (default) / SSH :26500 (dev-time)
Start command	`setsid nohup python3 -u /home/kachaka/<f> > /tmp/<f>.log 2>&1 < /dev/null &` -- `setsid` is mandatory: the Jupyter terminal kills its process group on close
Firmware updates	May wipe `/home/kachaka/` -- re-upload scripts on every deploy

On firmware < 3.17.8 JupyterLab may fail to start in Playground; curl -s -o /dev/null -w '%{http_code}\n' http://<robot-ip>:26501/login returning 200 means it is reachable. In a Jupyter terminal, write pgrep/pkill patterns as x[.]py -- the terminal echoes the command, so a plain x.py pattern matches itself.

MCP Tool Workflow

1. playground_upload(ip, script_content, filename)   → push to container
2. playground_run(ip, filename)                       → start in background
3. playground_log(ip)                                 → monitor output
4. playground_stop(ip, filename=filename)             → stop if needed
5. playground_status(ip, filename=filename)           → check if still running

Every tool takes transport (default "auto" = JupyterLab :26501, falling back to SSH only when :26501 is unreachable; "jupyter" / "ssh" force one). Always pass filename= to playground_stop / playground_status -- the default script.py targets the wrong process for any other file name. SSH key setup is only needed for transport="ssh" or dev-time shell access -- see "Dev-time SSH :26500 (optional)" in skills/kachaka-sdk/SKILL.md.

Example: Offline Multi-Stop Route with IMU Shake Detection

See skills/kachaka-sdk/examples/playground_offline_route.py for a production-verified script (tested on robot BKP40HD1T) that demonstrates:

Multi-stop shelf delivery -- configurable stop list with per-stop timeouts
IMU shake detection -- background thread with 2-of-3 sample voting, arm/disarm gating to avoid false positives during dock/undock (impacts reach 13+ m/s²)
Dual indicator thresholds -- accel > 11.0 m/s² OR gyro > 0.8 rad/s (gyro catches shakes that accel alone misses)
Optional HTTP progress reporting -- best-effort POST to external server with retry for network recovery
Graceful error recovery -- returns shelf and goes home on any failure

Deployment via MCP tools (preferred -- JupyterLab, no key):

playground_upload(ip, script_content, "offline_route.py")
playground_run(ip, "offline_route.py")
playground_log(ip)

Manual via JupyterLab (browser, no key): open http://<robot-ip>:26501 (password kachaka), upload playground_offline_route.py with the file browser (or drag-drop it in) so it lands in /home/kachaka/, open a Terminal from the launcher, and run:

setsid nohup python3 -u /home/kachaka/playground_offline_route.py > /tmp/route.log 2>&1 < /dev/null &

Dev-time over SSH (needs an SSH key in the container's ~/.ssh/authorized_keys):

scp -P 26500 playground_offline_route.py kachaka@<robot-ip>:/home/kachaka/
ssh -p 26500 kachaka@<robot-ip> "setsid nohup python3 -u /home/kachaka/playground_offline_route.py > /tmp/route.log 2>&1 < /dev/null &"

Remote Access via Tailscale (VPN)

Everything above assumes you share a LAN with the robot. For off-site deployment, maintenance, and fleet operations, install Tailscale inside the robot's Playground container. The container is non-root and has no /dev/net/tun, but Tailscale's official userspace networking mode works without either. This recipe is production-verified on multiple deployed robots (firmware 3.15--3.17).

Network Topology

graph LR
    subgraph laptop["Your Laptop / Server"]
        TSC["Tailscale client"]
    end

    subgraph robot["Kachaka Robot"]
        subgraph cont["Playground Container"]
            TSD["tailscaled<br/>(userspace mode)"]
            SSHD["sshd :22"]
            JUP["JupyterLab :26501"]
        end
        HOST["Robot host services<br/>gRPC :26400 · private API :2021<br/>(100.94.1.1)"]
    end

    TSC -->|"tailnet (DERP relay)"| TSD
    TSD -->|"inbound → container localhost"| SSHD
    TSD -->|"inbound → container localhost"| JUP
    TSD -->|"tailscale serve rules"| HOST

    style laptop fill:#fff3e0
    style cont fill:#e8f5e9
    style robot fill:#f5f5f5

Key consequence of running inside the container: inbound tailnet traffic lands on the container's localhost, not the robot host. Port numbers therefore differ between LAN and tailnet:

Service	Via LAN	Via tailnet
SSH	`<robot-ip>` port 26500 (host-side Docker map)	`100.x.y.z` port 22 (container sshd directly)
JupyterLab	port 26501	port 26501
Robot gRPC	port 26400	port 26400 -- only after the serve rule below
Private API (official mobile app)	port 2021	port 2021 -- only after the serve rule below

Deploy scripts that hard-code SSH_PORT=26500 should parameterize it (LAN 26500 / tailnet 22).

Installation

Prerequisites: SSH access over LAN (port 26500), and outbound internet from the robot -- verify with python3 -c "import urllib.request; print(urllib.request.urlopen('https://www.gstatic.com/generate_204').status)" (expect 204).

# 1. Download the arm64 static build on your laptop (robot is aarch64), then push it
curl -fsSL -o ts_arm64.tgz "https://pkgs.tailscale.com/stable/tailscale_<version>_arm64.tgz"
scp -P 26500 ts_arm64.tgz kachaka@<robot-ip>:/home/kachaka/

# 2. Extract and start the daemon (everything lives under /home/kachaka -- no system dirs)
ssh -p 26500 kachaka@<robot-ip> '
  mkdir -p /home/kachaka/tailscale-bin /home/kachaka/tailscale-state
  tar -xzf /home/kachaka/ts_arm64.tgz -C /home/kachaka/tailscale-bin --strip-components=1
  rm /home/kachaka/ts_arm64.tgz
  nohup /home/kachaka/tailscale-bin/tailscaled --tun=userspace-networking \
    --statedir=/home/kachaka/tailscale-state \
    --socket=/home/kachaka/tailscaled.sock >> /tmp/tailscaled.log 2>&1 &'

# 3. Register the node -- prints an auth URL; open it in a browser logged into your tailnet
ssh -p 26500 kachaka@<robot-ip> \
  '/home/kachaka/tailscale-bin/tailscale --socket=/home/kachaka/tailscaled.sock up --hostname <site>-<robot-name>'

After authorization, tailscale status --json reports "BackendState": "Running" and the node's 100.x.y.z address.

Standard Serve Rules (gRPC + Mobile App)

Inbound tailnet traffic only reaches the container's own localhost. Services on the robot host (100.94.1.1) must be proxied per-port with tailscale serve. Every deployment gets these two rules:

TS="/home/kachaka/tailscale-bin/tailscale --socket=/home/kachaka/tailscaled.sock"
$TS serve --bg --tcp=26400 tcp://100.94.1.1:26400   # robot gRPC (kachaka-api)
$TS serve --bg --tcp=2021  tcp://100.94.1.1:2021    # private API (official mobile app)

Rules persist in the statedir across reboots -- set once per robot. With port 2021 served, a phone running Tailscale can control the robot through the official Kachaka app from anywhere.

Verification from any tailnet machine:

ssh -p 22 kachaka@100.x.y.z echo ok                      # SSH (note: port 22, not 26500)
python3 -c "import kachaka_api; \
  print(kachaka_api.KachakaApiClient('100.x.y.z:26400').get_robot_serial_number())"
python3 -c "import ssl; ssl.get_server_certificate(('100.x.y.z', 2021))"   # no exception = OK

Boot Persistence

The robot reboots daily (default window 04:00--04:10; the time is configurable on Kachaka Pro since firmware 3.16.1, and the restart is skipped when uptime is too short or the clock is unsynced). State in /home/kachaka survives, so no re-authorization is needed -- but the daemon must be restarted. Append (never overwrite -- the file is a shared boot hook that other deployments may also use) to /home/kachaka/kachaka_startup.sh:

# --- tailscale ---
/home/kachaka/tailscale-bin/tailscaled --tun=userspace-networking \
  --statedir=/home/kachaka/tailscale-state \
  --socket=/home/kachaka/tailscaled.sock >> /tmp/tailscaled.log 2>&1 &
# --- end tailscale ---

Operational Notes

Topic	Detail
Latency	Userspace mode behind container NAT almost always relays via DERP (~80 ms). Fine for deployment, scp, and photo retrieval -- do not expect LAN throughput.
Memory	`tailscaled` RSS ~31 MB.
Key expiry	Disable it in the admin console (Machines → ⋯ → Disable key expiry), or the node needs re-authorization after ~180 days.
Firmware updates	May wipe `/home/kachaka` entirely -- binary, state, and startup hook included. Re-run this recipe afterwards (re-authorization required, since the state is gone too).
Security	The tailnet node key lives on the robot. Keep the invariant from the Playground section: API keys and other secrets never land on the robot.

Error Handling

Automatic Retry

All @with_retry-decorated methods use exponential backoff for transient gRPC errors:

gRPC Code	Retried?
`UNAVAILABLE`	Yes
`DEADLINE_EXCEEDED`	Yes
`RESOURCE_EXHAUSTED`	Yes
`INVALID_ARGUMENT`	No -- fails immediately
`NOT_FOUND`	No -- fails immediately
All other codes	No -- fails immediately

Default configuration: 3 attempts, 1s base delay, 10s max delay.

Response Format

Every method in the toolkit returns a dict:

# Success
{"ok": True, "action": "move_to_location", "target": "Kitchen"}

# Failure (gRPC, non-retryable)
{"ok": False, "error": "INVALID_ARGUMENT: unknown location", "retryable": False}

# Failure (gRPC, retries exhausted)
{"ok": False, "error": "UNAVAILABLE: connection refused", "retryable": True, "attempts": 3}

# Failure (robot error with enriched description)
{"ok": False, "error_code": 10253, "error": "error_code=10253: No destinations registered"}

Custom Retry Configuration

from kachaka_core.error_handling import with_retry

@with_retry(max_attempts=5, base_delay=2.0, max_delay=15.0)
def my_operation(sdk):
    return sdk.some_method()

Network Resilience

The toolkit provides five layers of automatic disconnect recovery, designed for robots operating on unstable mesh WiFi networks:

sequenceDiagram
    participant App
    participant Toolkit as kachaka_core
    participant Robot

    Note over App, Robot: Normal operation
    App->>Toolkit: get_pose()
    Toolkit->>Robot: gRPC (5s deadline)
    Robot-->>Toolkit: response (~10ms)
    Toolkit-->>App: {"ok": True, ...}

    Note over Robot: Network drops (e.g. mesh roaming)

    rect rgb(255, 230, 230)
        Note over Toolkit: Layer 1: TimeoutInterceptor (5s)
        App->>Toolkit: get_pose()
        Toolkit->>Robot: gRPC (5s deadline)
        Note right of Toolkit: No response — deadline fires at 5s
        Toolkit-->>Toolkit: DEADLINE_EXCEEDED

        Note over Toolkit: Layer 2: @with_retry (exponential backoff)
        Toolkit->>Robot: Retry 1 (1s delay)
        Note right of Toolkit: Still down — 5s timeout
        Toolkit->>Robot: Retry 2 (2s delay)
        Note right of Toolkit: Still down — 5s timeout
        Toolkit-->>App: {"ok": False, "retryable": True}
    end

    rect rgb(255, 245, 230)
        Note over Toolkit: Layer 3: ConnectionState monitoring (~7s detection)
        Toolkit->>Robot: Background ping fails
        Toolkit-->>Toolkit: state → DISCONNECTED
        Toolkit-->>App: on_state_change(DISCONNECTED)

        Note over Toolkit: Layer 4 & 5: Components pause
        Note right of Toolkit: RobotController._state_loop → sleep<br/>CameraStreamer._run → sleep<br/>(no wasted 5s timeouts)
    end

    Note over Robot: Network recovers

    rect rgb(230, 255, 230)
        Toolkit->>Robot: Background ping succeeds
        Toolkit-->>Toolkit: state → CONNECTED
        Toolkit-->>App: on_state_change(CONNECTED)
        Note right of Toolkit: RobotController & CameraStreamer<br/>resume immediately
    end

Layer	Component	Behaviour	Timing
1	`TimeoutInterceptor`	Injects 5s deadline on all unary calls	Fires at exactly deadline
2	`@with_retry`	Retries UNAVAILABLE/DEADLINE_EXCEEDED with exponential backoff	3 attempts default
3	`ConnectionState` monitoring	Background ping detects disconnect	~7s (ping interval + timeout)
4	`RobotController._state_loop`	Skips polling during DISCONNECTED	Immediate resume on CONNECTED
5	`CameraStreamer._run`	Skips capture during DISCONNECTED	Immediate resume on CONNECTED

The gRPC channel itself survives all disconnect types and does not require rebuilding. Recovery is automatic once the network path is restored.

Testing

The test suite uses pytest with unittest.mock to mock all gRPC calls. No live robot connection is required.

# Run all tests
uv run pytest

# Or with pip-installed project
pytest

# Verbose output
pytest -v

# Run a specific test file
pytest tests/test_camera.py

# Run a specific test class
pytest tests/test_commands.py::TestRetry

Test Coverage

Module	Tests	Covers
`test_connection.py`	42	Connection pool, normalisation, ping, resolver, monitoring, caching
`test_commands.py`	86	Movement (incl. mute_sensors / move_by_velocity_muted / source_location_id), shelf ops, fire-and-accept + command_id tracking, poll_until_complete verification, speech, shortcuts, map, torch, laser, retry, cancel, stop semantics (`stop_manual_drive` + deprecated `stop` + `set_emergency_stop` kept off MCP)
`test_queries.py`	40	Status, locations (incl. digest + undock flags + `type_name`), shelves (incl. digest + `HasField` pose + `speed_mode`), camera (incl. fresh-frame), intrinsics, ToF, map, errors (incl. `error_type`), info
`test_error_handling.py`	13	Retry modes (count + deadline), backoff, non-retryable errors
`test_interceptors.py`	12	TimeoutInterceptor default deadline injection, passthrough
`test_camera.py`	34	Lifecycle, capture, stats, errors, callbacks, thread safety, disconnect skip
`test_controller.py`	53	State polling, command execution, metrics, racing conditions, disconnect handling
`test_detection.py`	14	Detections, capture+detect, annotation, label mapping, error handling
`test_server_controller.py`	13	MCP controller tools: start/stop lifecycle, idempotency, state dict, controller_rotate
`test_transform.py`	12	TransformStreamer lifecycle, auto-reconnect, stats, thread safety
`test_playground.py`	14	Playground SSH upload/run/stop/log/status (subprocess ssh, mocked; ssh failure ≠ idle)
`test_playground_jupyter.py`	33	Playground JupyterLab transport (contents API + terminal, mocked)
Total	387 tests

All tests use the _clean_pool autouse fixture to ensure isolation between tests.

HIL tests live in tests/hil/ and need a real robot. With only --robot-ip the stationary tests run; tests that drive the robot require an extra opt-in:

pytest tests/hil/ -v --robot-ip=192.168.50.133                # stationary only
pytest tests/hil/ -v --robot-ip=192.168.50.133 --hil-motion   # + driving tests (~3 m)

tests/hil/test_stop_semantics.py proves that cancel_command stops a moving robot and stop_manual_drive does not (--hil-motion). Its emergency-stop case needs a further --ems-latch because it leaves the robot latched until someone presses the physical power button.

Project Structure

graph LR
    ROOT["kachaka-sdk-toolkit/"]

    subgraph core["kachaka_core/ — Shared core library"]
        C_INIT["__init__.py — Public exports"]
        C_CONN["connection.py — Thread-safe pooled gRPC + monitoring + cache"]
        C_INT["interceptors.py — TimeoutInterceptor (5s default)"]
        C_CMD["commands.py — Movement, shelf, speech, torch, laser, map"]
        C_QRY["queries.py — Status, camera intrinsics, ToF, map queries"]
        C_CAM["camera.py — CameraStreamer daemon thread"]
        C_CTRL["controller.py — RobotController + polling"]
        C_DET["detection.py — ObjectDetector + annotation"]
        C_TF["transform.py — TransformStreamer (dynamic TF)"]
        C_PG["playground.py — Playground facade + PlaygroundSSH"]
        C_PGJ["playground_jupyter.py — PlaygroundJupyter (JupyterLab :26501)"]
        C_ERR["error_handling.py — @with_retry"]
    end

    subgraph mcp["mcp_server/ — MCP Server layer"]
        M_INIT["__init__.py"]
        M_SRV["server.py — 88 tools, stdio transport"]
    end

    subgraph skills["skills/kachaka-sdk/ — Plugin skill"]
        S_MD["SKILL.md — Full API reference"]
        S_EX["examples/typical_usage.py"]
        S_PG["examples/playground_offline_route.py"]
    end

    subgraph plugin[".claude-plugin/ — Plugin metadata"]
        P_JSON["plugin.json"]
    end

    subgraph tests["tests/ — pytest suite (311 tests)"]
        T_CONN["test_connection.py"]
        T_CMD["test_commands.py"]
        T_QRY["test_queries.py"]
        T_ERR["test_error_handling.py"]
        T_INT["test_interceptors.py"]
        T_CAM["test_camera.py"]
        T_CTRL["test_controller.py"]
        T_DET["test_detection.py"]
        T_SCTRL["test_server_controller.py"]
        T_TF["test_transform.py"]
    end

    subgraph cli["kachaka_sdk_toolkit/ — Setup CLI"]
        CLI_INIT["__init__.py"]
        CLI_SETUP["setup_cli.py — kachaka-setup command"]
    end

    subgraph examples["examples/"]
        EX_CAM["camera_web.py — FastAPI MJPEG viewer"]
    end

    ROOT --- core
    ROOT --- mcp
    ROOT --- skills
    ROOT --- plugin
    ROOT --- tests
    ROOT --- cli
    ROOT --- examples
    ROOT --- MCP_JSON[".mcp.json"]
    ROOT --- TOML["pyproject.toml"]

Anti-patterns

Don't	Do Instead
`KachakaApiClient(ip)` directly	`KachakaConnection.get(ip)`
Write your own retry logic	Use `@with_retry` decorator
Forget to poll command status	Use `poll_until_complete()`
Call camera in a tight loop	Use `CameraStreamer` for continuous capture
Hard-code robot IP	Pass as parameter or env var
Ignore `result["ok"]`	Always check before proceeding
Call `sdk.move_to_location()` raw	Use `cmds.move_to_location()` (handles resolver)
Use `KachakaCommands` for patrol sequences	Use `RobotController` for multi-step patrols with metrics

Design Caveats

Caveat 1: Non-blocking Command Execution (`wait_for_completion=False`)

The upstream kachaka-api SDK provides a wait_for_completion parameter (default True) on all action commands. When enabled, the SDK internally holds open a gRPC streaming call until the command finishes.

This is convenient for quick testing but dangerous in production.

When the robot moves to an area with poor WiFi, the gRPC connection does not fail immediately -- it silently blocks. In real-world testing, a single gRPC call was observed blocking for 522 seconds before timing out. During this period, the main thread is completely deadlocked.

kachaka_core always uses wait_for_completion=False and manages command lifecycle through its own mechanisms:

graph LR
    subgraph "kachaka-api SDK (dangerous in production)"
        A["sdk.move_to_location(id)"] -->|"wait_for_completion=True"| B["gRPC streaming blocks\nuntil command completes"]
        B -->|"WiFi drops"| C["Blocks 500+ seconds\nMain thread deadlocked"]
    end

    subgraph "kachaka_core (production-safe)"
        D["cmds.move_to_location(name)"] -->|"wait=False"| E["Returns immediately"]
        E --> F["poll_until_complete()\nor RobotController polling"]
        F -->|"Each poll"| G["TimeoutInterceptor\n5s deadline per call"]
        G -->|"WiFi drops"| H["Fails in 5s\nRetries or reports error"]
    end

The three layers of protection:

Layer	Component	What It Does
1	`TimeoutInterceptor`	Injects a 5-second deadline on every gRPC call
2	`KachakaCommands.poll_until_complete()`	Polls `GetCommandState` every 0.5s with a configurable timeout, verifying completion against the dispatched `command_id`
3	`RobotController._execute_command()`	Captures `command_id`, polls with verification, confirms result belongs to our command

Caveat 2: Shelf Drop Detection Limitations

The Kachaka API does not provide a native event or callback for shelf drops. There is no ShelfDropEvent streaming RPC, no error code emitted at drop time, and no webhook mechanism.

kachaka_core's approximation: RobotController implements polling-based shelf drop detection:

sequenceDiagram
    participant BG as Background Thread<br/>(_state_loop)
    participant API as Kachaka gRPC
    participant State as RobotState

    Note over BG: move_shelf() called → _monitoring_shelf = True

    loop Every fast_interval (1s)
        BG->>API: get_moving_shelf_id()
        API-->>BG: shelf_id or ""

        alt shelf_id present AND not yet confirmed
            BG->>State: _shelf_confirmed_docked = True
            BG->>State: moving_shelf_id = shelf_id
        else shelf_id == "" AND _shelf_confirmed_docked == True
            BG->>State: shelf_dropped = True
            Note over State: Drop detected!
        end
    end

    Note over BG: return_shelf() called → _monitoring_shelf = False

Known limitations:

Limitation	Detail
Detection delay	Up to `fast_interval` seconds (default 1s) between drop and detection
RobotController only	Not available in `KachakaCommands` (no background thread)
No MCP push notification	MCP is request-response; `shelf_dropped` only visible via `get_controller_state()`
No automatic cancellation	Application must read `shelf_dropped=True` and decide whether to cancel

Adding New Functionality

Implement in kachaka_core/commands.py (actions) or kachaka_core/queries.py (read-only)
Add a corresponding @mcp.tool() in mcp_server/server.py
Update skills/kachaka-sdk/SKILL.md
Add tests in tests/

Example:

# kachaka_core/commands.py
@with_retry()
def my_new_command(self, param: str) -> dict:
    result = self.sdk.some_sdk_method(param)
    return self._result_to_dict(result, action="my_new_command", target=param)

# mcp_server/server.py
@mcp.tool()
def my_new_command(ip: str, param: str) -> dict:
    """Description for Claude to understand when to use this tool."""
    return KachakaCommands(KachakaConnection.get(ip)).my_new_command(param)

License

This project is licensed under the Apache License 2.0 - see the LICENSE file for details.

This project wraps the kachaka-api SDK. Refer to that project for SDK licensing terms.

Name		Name	Last commit message	Last commit date
.claude-plugin		.claude-plugin
.github/workflows		.github/workflows
kachaka_core		kachaka_core
kachaka_sdk_toolkit		kachaka_sdk_toolkit
mcp_server		mcp_server
skills/kachaka-sdk		skills/kachaka-sdk
tests		tests
.gitignore		.gitignore
.mcp.json		.mcp.json
ARCHITECTURE.md		ARCHITECTURE.md
LICENSE		LICENSE
README.md		README.md
pyproject.toml		pyproject.toml
pytest.ini		pytest.ini
setup.cfg		setup.cfg
uv.lock		uv.lock

Latest commit

History

Folders and files

Repository files navigation

kachaka-sdk-toolkit

Overview

Architecture

Features

Tech Stack

Getting Started

Prerequisites

Installation

Option A: Claude Code Plugin (recommended)

Option B: Local Development

Quick Start

Core API Reference

KachakaConnection

Timeout Protection

Connection Monitoring

Disconnect Behaviour (measured on real hardware)

Two-Tier Device-Info Cache

KachakaCommands

KachakaQueries

CameraStreamer

Basic Usage

With Callback

Back Camera

With Detection Overlay

Properties

Design Notes

TransformStreamer

Properties

Design Notes

RobotController

Why prefer this over KachakaCommands?

Basic Usage

Movement API

Constructor Parameters

When to Use

Command Execution Flow

ObjectDetector

Basic Usage

Labels

Notes

MCP Server

Running the Server

Claude Desktop Configuration

Tool Reference

Connection (3 tools)

Status Queries (6 tools)

Locations and Shelves (5 tools)

Movement (7 tools)

Shelf Operations (6 tools)

Controller (8 tools)

Speech (3 tools)

Custom Sounds (5 tools)

Command Control (4 tools)

Camera (8 tools)

Object Detection (2 tools)

Map (6 tools)

Shortcuts and History (3 tools)

Manual Control (4 tools)

Torch / Lighting (2 tools)

Laser Scan (1 tool)

Auto Homing (2 tools)

Readiness (1 tool)

Localization (2 tools)

Recovery (1 tool)

Transforms (4 tools)

Playground (5 tools)

Playground Offline Execution

Why Playground -- Offline-First Robot Control

When to Use Playground vs. kachaka_core

Container Environment

MCP Tool Workflow

Example: Offline Multi-Stop Route with IMU Shake Detection

Remote Access via Tailscale (VPN)

Network Topology

Installation

Standard Serve Rules (gRPC + Mobile App)

Why prefer this over `KachakaCommands`?

Caveat 1: Non-blocking Command Execution (`wait_for_completion=False`)