AuthorPascal DateJuly 14, 2026 Rev1.0

robot-mcp-kit — Implementation Spec

robot-mcp-kit is a small, pip-installable Python library that every robot capability MCP server (robot-mcp-navigation, robot-mcp-perception, robot-mcp-memory, and the VLA/manipulation servers) depends on. It implements the Async Task Contract — the convention that lets long-running robot work (navigation, scene analysis, watchers) run over plain MCP + HTTP with progress, triggers, completion, failure, and cancellation — plus the HTTP event poster and the shared pydantic models.

Published as v0.1.0 and consumed by the brain. The Async Task Contract is used only for push events — chiefly the watch_events posted by mcp-perception-buffer when a watcher skill fires (and long-running ROS completions). Per-capability async tasks such as navigation's eta_below are not used.

Canonical contract

This document is self-contained: the whole library can be implemented from this file alone. It is also the canonical definition of the contract — the other repos' specs restate the parts they consume, but if anything ever disagrees, this file wins.

---

1. System context

The overall system: a chat/voice app (robot-voice-chat, FastAPI + Socket.IO, runs on each robot) contains an agentic MissionExecutor that plans multi-step robot missions with an LLM and calls tools on capability MCP servers (one systemd service each, HTTP/SSE transport, on-robot). MCP tool calls are request/response — fine for wave(), wrong for a 60-second navigation. So:

1. A long-running tool (navigate_to, find_object, watch_for) returns immediately with a task_id; the work continues in an asyncio task inside the server.
2. The server POSTs events (progress, fired triggers, completion, failure) to the chat app's POST /api/agent/events endpoint, authenticated with a per-server HMAC secret.
3. The chat app can always reconcile by polling: every task-capable server also exposes get_task, cancel_task, list_tasks MCP tools. Push is the optimization; poll is the truth (used after chat-app restarts or event gaps).
4. Triggers: the caller can attach conditions when starting a task (e.g. {"type": "eta_below", "seconds": 10} on a navigation). The server evaluates them (it owns the signal) and emits a trigger_fired event. This is how "start scene understanding 10 s before arrival" works without polling.

This library packages all of that so a capability server is just FastMCP + robot-mcp-kit + domain code.

---

2. Repo layout

robot-mcp-kit/
├── pyproject.toml               # name: robot-mcp-kit, package: robot_mcp_kit
├── README.md                    # short usage doc (condense §5 of this spec)
├── src/robot_mcp_kit/
│   ├── __init__.py              # re-export public API (everything in §4)
│   ├── models.py                # pydantic models: TaskEvent, TaskInfo, TaskFailure, ...
│   ├── tasks.py                 # TaskManager, TaskContext, TaskError
│   ├── events.py                # EventPoster (HTTP + HMAC + retry)
│   ├── signing.py               # sign_body() / verify_signature() — shared with the chat app
│   └── fastmcp_ext.py           # register_task_tools(mcp, manager)
└── tests/
    ├── test_tasks.py
    ├── test_events.py           # against a local aiohttp/respx fake endpoint
    └── test_signing.py

Tooling (match the robot-voice-chat backend conventions): Python ≥ 3.11, uv for env + lockfile, ruff + ruff format, mypy --strict, pytest + pytest-asyncio. CI: GitHub Actions running all four gates on push/PR.

Dependencies: pydantic>=2, httpx>=0.27, mcp (only for fastmcp_ext, keep it an optional import so the models can be used without MCP installed). No other runtime deps.

Versioning: semver, git tags. Consumers install via uv add "robot-mcp-kit @ git+https://github.com/<org>/robot-mcp-kit@v0.1.0". The wire contract carries schema_version (below) so the library and chat app can evolve independently.

---

3. The wire contract

3.1 Event POST (server → chat app)

Every event is one HTTP request:

POST {RVC_CALLBACK_URL}            # e.g. http://127.0.0.1:8000/api/agent/events
Content-Type: application/json
X-RVC-Server: navigation           # server name, matches chat app's mcp.yaml entry
X-RVC-Timestamp: 1765532000        # unix seconds, ±300s tolerance
X-RVC-Signature: sha256=<hex>      # HMAC-SHA256(secret, f"{timestamp}.{raw_body}")

Body (TaskEvent):

{
  "schema_version": 1,
  "event_id": "8c1f0e2a-...",          // uuid4, unique per event
  "task_id": "nav-7f3a2b",             // "<server-short>-<6 hex>" — generated by TaskManager
  "server": "navigation",
  "tool": "navigate_to",               // tool that started the task
  "type": "progress" | "trigger_fired" | "task_done" | "task_failed" | "watch_event",
  "seq": 4,                            // monotonically increasing per task, starts at 1
  "ts": "2026-06-12T14:03:00.123Z",    // UTC ISO 8601
  "payload": { ... }                   // type-specific, see below
}

Payload by type:

type	payload
progress	free-form dict from the task body, e.g. {"eta_s": 22.4, "distance_m": 14.1, "percent": 60}
trigger_fired	{"trigger": {<the trigger dict as passed in>}, ...extra} e.g. {"trigger": {"type": "eta_below", "seconds": 10}, "eta_s": 9.4}
task_done	{"result": {<dict returned by the task body>}}
task_failed	a TaskFailure (see §4.1)
watch_event	free-form dict from a watcher (perception server) — same envelope, long-lived task

Delivery semantics: at-least-once. The chat app dedupes on event_id (and can detect gaps via seq). Events for one task must be POSTed in seq order by the poster (single queue per process is sufficient). A 2xx response acknowledges; anything else (or a connection error) is retried with exponential backoff (0.5 s, 2 s, 8 s, then every 30 s, capped at 10 minutes total per event, then dropped with an ERROR log — the chat app's reconciliation poll covers the loss).

3.2 Trigger dicts

A trigger is a plain dict with a type key; everything else is type-specific. The kit treats them opaquely — semantics live in each capability server (navigation implements eta_below and zone_entered; perception implements its own). Known types so far:

type	fields	owner
eta_below	seconds: float — fire once when estimated remaining time < seconds	navigation
zone_entered	zone: str — fire once when the robot enters the named zone	navigation
progress_above	percent: float — fire once when progress ≥ percent	any (generic helper, see §4.2)

Servers must ignore unknown trigger types with a WARNING log (forward compatibility), and fire each trigger at most once per task unless the trigger dict carries "repeat": true.

3.3 Reconciliation tools (server → exposed over MCP)

Every task-capable server exposes these three tools (added by register_task_tools, §4.4). All return JSON strings (MCP tool results are text):

• get_task(task_id: str) → serialized TaskInfo or {"error": "not_found"}
• cancel_task(task_id: str) → {"task_id": ..., "state": "cancelled"} (idempotent; cancelling a finished task returns its terminal state unchanged)
• list_tasks(states: list[str] | None = None) → {"tasks": [TaskInfo, ...]} — with states=["running"] this is what the chat app calls after a restart.

3.4 Environment variables (consumed by the kit, set in each server's systemd unit)

Var	Meaning	Default
RVC_CALLBACK_URL	Chat app events endpoint. Unset ⇒ EventPoster is a no-op (logs events at DEBUG) — lets a server run standalone in dev.	unset
RVC_EVENT_SECRET	HMAC secret shared with the chat app's mcp.yaml entry for this server. Required if RVC_CALLBACK_URL is set.	unset

---

4. Public API

4.1 robot_mcp_kit.models

from datetime import datetime
from typing import Any, Literal
from pydantic import BaseModel, Field

SCHEMA_VERSION = 1

TaskState = Literal["running", "succeeded", "failed", "cancelled"]
EventType = Literal["progress", "trigger_fired", "task_done", "task_failed", "watch_event"]

class TaskFailure(BaseModel):
    """Designed to be read by the planner LLM — machine code + human text + recovery hints."""
    error_code: str                      # e.g. "DOOR_CLOSED", "OBJECT_NOT_FOUND", "TIMEOUT"
    message: str                         # one or two sentences, written for an LLM
    recoverable: bool = True             # False = don't bother replanning around this
    suggestions: list[str] = Field(default_factory=list)   # e.g. ["ask_reception", "retry_after_s:300"]

class TaskEvent(BaseModel):
    schema_version: int = SCHEMA_VERSION
    event_id: str                        # uuid4 hex
    task_id: str
    server: str
    tool: str
    type: EventType
    seq: int
    ts: datetime
    payload: dict[str, Any]

class TaskInfo(BaseModel):
    task_id: str
    server: str
    tool: str
    state: TaskState
    created_at: datetime
    updated_at: datetime
    triggers: list[dict[str, Any]] = Field(default_factory=list)
    est_duration_s: float | None = None          # as passed to TaskManager.start()
    progress: dict[str, Any] | None = None      # last progress payload
    result: dict[str, Any] | None = None        # set when state == "succeeded"
    failure: TaskFailure | None = None          # set when state == "failed"

4.2 robot_mcp_kit.tasks

class TaskError(Exception):
    """Raise from a task body to fail the task with structured info."""
    def __init__(self, error_code: str, message: str, *,
                 recoverable: bool = True, suggestions: list[str] | None = None): ...

class TaskContext:
    """Handed to every task body. All methods are async-safe."""
    task_id: str
    triggers: list[dict[str, Any]]                 # the triggers the caller attached

    async def progress(self, payload: dict[str, Any]) -> None:
        """Record + emit a `progress` event. Also evaluates `progress_above` triggers
        automatically if payload contains a numeric `percent` key."""

    async def fire_trigger(self, trigger: dict[str, Any], extra: dict[str, Any] | None = None) -> None:
        """Emit a `trigger_fired` event. The kit enforces at-most-once per (task, trigger)
        unless trigger has repeat: true."""

    async def emit(self, payload: dict[str, Any]) -> None:
        """Emit a `watch_event` (for long-lived watcher tasks)."""

    def raise_if_cancelled(self) -> None:
        """Cooperative cancellation checkpoint — raises asyncio.CancelledError."""

TaskBody = Callable[[TaskContext], Awaitable[dict[str, Any]]]

class TaskManager:
    def __init__(self, server_name: str, poster: EventPoster,
                 *, max_finished_retained: int = 200): ...

    def start(self, tool: str, body: TaskBody,
              triggers: list[dict[str, Any]] | None = None,
              est_duration_s: float | None = None) -> TaskInfo:
        """Create a task, spawn `body` as an asyncio.Task, return immediately
        (state == "running"). Lifecycle:
          - body returns dict        -> state "succeeded", emit task_done {"result": <dict>}
          - body raises TaskError    -> state "failed",    emit task_failed <TaskFailure>
          - body raises CancelledError -> state "cancelled" (no failure event; cancel_task
            already returned synchronously to the caller)
          - body raises anything else -> state "failed", error_code "INTERNAL",
            message = str(exc), recoverable False; log with traceback
        """

    def get(self, task_id: str) -> TaskInfo | None: ...
    def list(self, states: list[TaskState] | None = None) -> list[TaskInfo]: ...
    async def cancel(self, task_id: str) -> TaskInfo | None:
        """asyncio-cancel the underlying task, await its teardown (5 s grace), mark cancelled."""
    async def aclose(self) -> None:
        """Cancel all running tasks (server shutdown)."""

Implementation notes:

• task_id format: f"{server_short}-{uuid4().hex[:6]}" where server_short is the first 4 chars of the server name (nav-7f3a2b, perc-19c0de, memo-...). Purely cosmetic; uniqueness comes from the manager's dict.
• Finished tasks are retained in memory (ring of max_finished_retained) so get_task works after completion. No persistence — a server restart loses task state by design; the chat app's reconciliation treats a missing task as failed (error_code: "SERVER_RESTARTED" — that mapping happens chat-app-side, the kit just returns not-found).
• seq is per-task, allocated under the manager's lock before enqueueing to the poster.

4.3 robot_mcp_kit.events and robot_mcp_kit.signing

# signing.py — used by both this library and the chat app's endpoint
def sign(secret: str, timestamp: int, body: bytes) -> str:
    """hex(HMAC_SHA256(secret, f"{timestamp}.".encode() + body))"""
def verify(secret: str, timestamp: int, body: bytes, signature_header: str,
           *, max_skew_s: int = 300) -> bool:
    """Constant-time compare (hmac.compare_digest) + timestamp skew check."""

# events.py
class EventPoster:
    def __init__(self, server_name: str,
                 callback_url: str | None = None,    # defaults from RVC_CALLBACK_URL
                 secret: str | None = None):          # defaults from RVC_EVENT_SECRET
        """callback_url None -> no-op mode (log at DEBUG). url set but secret missing -> raise
        at construction (fail fast)."""

    async def post(self, event: TaskEvent) -> None:
        """Enqueue; a single background worker per poster sends in order with the retry policy
        from §3.1. Never raises into the caller."""

    async def aclose(self) -> None: ...

Use one shared httpx.AsyncClient per poster (keep-alive). The worker must preserve per-task ordering; a single global FIFO queue satisfies that and is fine at this volume.

4.4 robot_mcp_kit.fastmcp_ext

def register_task_tools(mcp: "FastMCP", manager: TaskManager) -> None:
    """Adds get_task / cancel_task / list_tasks tools (§3.3) to a FastMCP server.
    Docstrings on the generated tools must tell the LLM these are for status/cancel of
    long-running tasks, not for starting work."""

---

5. Usage example

import asyncio
from mcp.server.fastmcp import FastMCP
from robot_mcp_kit import EventPoster, TaskManager, TaskContext, TaskError, register_task_tools

mcp = FastMCP("sleeper")
poster = EventPoster(server_name="sleeper")            # env-configured
manager = TaskManager("sleeper", poster)
register_task_tools(mcp, manager)

@mcp.tool()
def sleep_task(duration_s: float, fail: bool = False,
               triggers: list[dict] | None = None) -> str:
    """Start a fake long-running task that sleeps for duration_s. Supports trigger type
    {"type": "eta_below", "seconds": N}. Returns a task_id immediately."""
    async def body(ctx: TaskContext) -> dict:
        fired = set()
        for elapsed in range(int(duration_s)):
            ctx.raise_if_cancelled()
            await asyncio.sleep(1)
            eta = duration_s - elapsed - 1
            await ctx.progress({"eta_s": eta, "percent": 100 * (elapsed + 1) / duration_s})
            for t in ctx.triggers:
                if t.get("type") == "eta_below" and eta < t["seconds"] and id(t) not in fired:
                    fired.add(id(t)); await ctx.fire_trigger(t, {"eta_s": eta})
        if fail:
            raise TaskError("FAKE_FAILURE", "The sleeper was asked to fail.",
                            suggestions=["retry_without_fail"])
        return {"slept_s": duration_s}
    info = manager.start("sleep_task", body, triggers=triggers, est_duration_s=duration_s)
    return info.model_dump_json()

Dual-mode pattern

Every capability server follows this pattern: the tool function is the thin synchronous entry (validate args → manager.start → return TaskInfo JSON); the body is the real async work.

---

6. Tests

The suite covers:

1. Task lifecycle: start → progress events in order with increasing seq → done with result; TaskError → failed with structured failure; unexpected exception → failed INTERNAL; cancel → cancelled, body's CancelledError honored, no task_failed emitted.
2. Triggers: eta_below-style trigger fires exactly once; repeat: true fires repeatedly; unknown types ignored by progress_above auto-evaluation.
3. EventPoster: signs correctly (golden-value test for sign()); retries on 500 then succeeds; gives up after the cap and logs; preserves order under concurrent post() calls; no-op mode.
4. Signing: round-trip sign/verify; rejects bad signature, stale timestamp, tampered body.
5. register_task_tools: tools appear in list_tools(); get_task/cancel_task behave per §3.3 (use the in-memory FastMCP test client).

7. Acceptance criteria

• uv run pytest, ruff check, ruff format --check, and mypy --strict all pass.
• The §5 sleeper example runs as a real SSE server (MCP_TRANSPORT=sse MCP_PORT=9100 python example/sleeper.py) and, with RVC_CALLBACK_URL pointed at nc -l or a tiny echo server, emits correctly signed events in order.

---

Appendix A — Repository & tooling conventions

Scope

This repo is a pip-installable library (no server runtime). The layout, tooling, and architecture conventions apply; the MCP-servers-specifically notes describe the servers that consume this library, not this repo itself.

Structure — src/<package>/ layout (never flat); config/ Hydra tree; tests/{unit,integration}/; committed README.md (quickstart + how-to table + troubleshooting table), CHANGELOG.md (Keep-a-Changelog + SemVer), .pre-commit-config.yaml, .github/ (CI + dependabot.yml), optional multi-stage non-root Dockerfile, .env.example committed / .env gitignored.

Tooling — uv for env + deps (uv sync, lockfile committed), Python ≥ 3.11, optional extras for heavy/optional deps (the --extra observability pattern). ruff (lint + format), mypy --strict, pytest + pytest-asyncio, coverage gate ≥ 80%. Pre-commit: trailing-whitespace, end-of-file-fixer, check-yaml/toml/merge-conflict/large-files, ruff (--fix), ruff-format, and a local mypy hook run via uv run (so it picks up the pydantic.mypy plugin). CI on every PR: pytest, ruff check, ruff format --check, mypy, pip-audit. Dependabot weekly for the uv ecosystem.

Architecture — ports-and-adapters: every capability with >1 implementation or that touches hardware/network gets an abc.ABC port in base.py; concrete adapters live beside it; application code imports only ports. Hydra + Pydantic config: one config-group dir per port, one YAML per implementation carrying _target_; swap implementations by changing one line in the root config.yaml defaults list — no if provider == … ladders. Validate the composed config through Pydantic dataclasses at startup (fail fast). Deployment presets under config/deployment/<name>.yaml selected via an env var.

Secrets — never in YAML or git. .env + OmegaConf ${oc.env:KEY} resolvers, typed SecretStr; startup refuses to boot if a required key is empty.

Logging — structlog JSON by default, pretty renderer toggle for dev; bind correlation IDs (request/turn/mission) via contextvars; log duration_ms per stage. Subprocess — always argv lists, never shell=True. Testing — fakes at the port boundary (don't patch internals); respx for HTTP-level mocking; integration via httpx.AsyncClient + ASGITransport.

MCP servers specifically — one repo per capability from a copy-me template; transport stdio if the consumer supervises it (1–2 co-located), else sse/streamable_http with its own systemd unit/container/logs/restart policy and the consumer just holds a URL. Tools namespaced server.tool; mark slow perception tools "blocking", else fire-and-forget; expect a ~30 s call timeout; tolerate clients health-checking via list_tools(). Resilience: the consumer reconnects on an interval — design for clean restart, never assume connection order at boot. Deploy: systemd unit with hardening (NoNewPrivileges, PrivateTmp, ProtectSystem=strict), bind localhost, dedicated user, journald.