AuthorPascal DateJuly 29, 2026 Rev1.4

robot-mcp-kit — Implementation Spec¶

robot-mcp-kit is the shared, git-installable Python fleet library that the brain and the robot-side services depend on: grasp-service, mcp-perception-buffer, robot-mcp-memory, and ros2-memory (the brain consumes its ToolClient). It bundles the MCP client (ToolClient), the HMAC-signed event poster (EventPoster), the fleet-learning store (ParamStore/lessons), the RecoveryAgent, server helpers, and the Async Task Contract — the convention that lets long-running robot work run over plain MCP + HTTP with progress, triggers, completion, failure, and cancellation.

It is a private dependency (github.com/binabik-ai/robot-mcp-kit), pinned by git commit in each consumer's lockfile; CI authenticates it via a read-only deploy key. The Async Task Contract is used only for push events — chiefly the watch_events the mcp-perception-buffer posts when a watcher fires (and long-running ROS completions); illustrative examples like a navigation eta_below are not part of the current stack.

Audience

Developers of the brain and the robot-side capability/skill services that depend on this library (grasp-service, mcp-perception-buffer, robot-mcp-memory, ros2-memory). §4 is the complete public API — every name here is re-exported from robot_mcp_kit/__init__.py.

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 brain app (robot-voice-chat, FastAPI + Socket.IO, one container per colleague on the brain host) contains an agentic MissionExecutor that plans multi-step robot missions with an LLM and calls tools on capability MCP servers (HTTP transport, on the robot). MCP tool calls are request/response — fine for wave(), wrong for a 60-second navigation. So:

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.
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.
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).
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.

Since BIN-549 it also owns the stack's shared persistence layer (§4.9): the SQLite store class, the kind vocabulary and the retention shape that BIN-546 makes every service use. That is not part of the wire contract above — it is the machinery four repos would otherwise each have written, kept in one place so the guards that make BIN-546's taxonomy enforceable are inherited rather than re-argued.

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() / verify() — shared with the chat app
│   ├── usage.py                 # UsageRecord — what one billable call cost (§4.4, BIN-240)
│   ├── usage_report.py          # UsageCollector + UsageReporter — how it gets reported (§4.4)
│   ├── fastmcp_ext.py           # register_task_tools(mcp, manager)
│   ├── client.py                # ToolClient — async MCP client for calling other servers (§4.6)
│   ├── learning.py              # ParamStore + Signature/Override/Recall — fleet lessons (§4.7)
│   ├── recovery.py              # RecoveryAgent + Decision — bounded LLM recovery loop (§4.8)
│   └── store/                   # the shared persistence layer (§4.9, BIN-546/BIN-549)
│       ├── kinds.py             #   Kind + Schema: every table declares which kind it holds
│       ├── schema.py            #   open_versioned: migrate forward, refuse a downgrade
│       ├── sqlite_store.py      #   SqliteStore: one connection, one writer thread, prune()
│       └── lifecycle.py         #   RetentionPolicy/run_startup + the one `store_ready` line
└── tests/
    ├── test_tasks.py
    ├── test_events.py           # against a local aiohttp/respx fake endpoint
    ├── test_signing.py
    └── test_writes_go_to_the_database.py   # the BIN-546 CI guard, one copy per repo

Tooling (match the robot-voice-chat backend conventions): Python ≥ 3.10 — the robot decides this floor, because binabik-r1-vision and galaxea_agent's shared MCP venv import the kit on the ROS distro's interpreter, which is 3.10 on Humble (BIN-426); a >=3.11 floor does not get those hosts a newer Python, it gets them no kit at all. CI runs the matrix 3.10–3.13. 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 and client.py, kept an optional import — the mcp extra, floored at mcp>=1.28.1 since the ToolClient speaks streamable-HTTP — so the models can be used without MCP installed), and anthropic>=0.40 behind the recovery extra (only RecoveryAgent's default controller imports it, deferred and guarded; see §4.8). No other runtime deps.

Versioning: semver, one git tag per release. The releases are tagged v0.1.0 … v0.7.0; v0.1.0–v0.5.1 were tagged retroactively under BIN-142, each on its version-bump commit, because the repo had shipped six versions with no tags at all — which left every compare link in CHANGELOG.md and the README's install line pointing at refs that did not resolve. Consumers install a release with uv add "robot-mcp-kit[mcp] @ git+https://github.com/binabik-ai/robot-mcp-kit@v0.6.0", or track main through [tool.uv.sources] and let uv.lock pin the commit (what the fleet does today — see Compatibility). 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}            # path: /api/agent/events on the brain's HTTP origin
Content-Type: application/json
X-RVC-Server: navigation           # server name, matches the brain's attach-manifest entry
X-RVC-Timestamp: 1765532000        # unix seconds, ±300s tolerance
X-RVC-Signature: sha256=<hex>      # HMAC-SHA256(secret, f"{timestamp}.{raw_body}")

8000 is the container-internal port — don't hard-code it in RVC_CALLBACK_URL

The brain listens on :8000 inside its own container, so only code in that container may use http://127.0.0.1:8000/api/agent/events. Instances are published on the host from 8001 upward (BRAIN_PORT_BASE, one port per colleague), so a robot-side server posting events must target the host-visible origin of that brain — http://<brain-host>:<instance-port>/api/agent/events. The path is always /api/agent/events; only the origin varies. Neither brainctl nor r1ctl sets this today — it is a robot-side server's own env var to configure, and nothing attached to a manifest currently pushes a TaskEvent this way (see Integrating a robot §3f).

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 Usage POST (server → brain, BIN-240 phase 2)¶

The kit's other outbound channel, used by UsageReporter for spend a tool result can never carry (a bounded recovery loop that hits its deadline and returns nothing at all, having already paid for every step). Its default counterpart, UsageCollector, needs no wire format at all — it hands UsageRecord.model_dump(mode="json") back to the caller to attach under a usage key on whatever the tool already returns, and the brain harvests it off every result.

POST {RVC_USAGE_URL}               # path: /api/agent/usage on the brain's HTTP origin
Content-Type: application/json
X-RVC-Server: grasp-service        # attribution only — see the secret note below
X-RVC-Timestamp: 1765532000        # unix seconds, ±300s tolerance
X-RVC-Signature: sha256=<hex>      # HMAC-SHA256(secret, f"{timestamp}.{raw_body}")

Same signing scheme as §3.1's Event POST, and the same 8000-vs-published-port trap (see the warning there) — but authenticated by a different, fleet-wide secret, RVC_USAGE_SECRET, never the per-server RVC_EVENT_SECRET/event_secret pair. That secret's manifest entry forces the brain to emit the server's name as a tool-name prefix — attaching it to a robot server that already has bare tools (grasp) would rename them to grasp.grasp the moment it started reporting spend. X-RVC-Server here is attribution carried into a rejection log only; the record's own service field is what the brain's ledger actually reads.

Body (UsageRecord, from robot_mcp_kit.usage) is sent unpriced: confidence: "unpriced", cost_micro_usd: null. The brain prices it on arrival from its own table — there is no price table on a robot, and one there would let a stale rate silently out-vote the brain's. A record that already carries a price, or whose robot_id disagrees with the brain's own, is rejected (400), not trusted.

3.3 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.4 Reconciliation tools (server → exposed over MCP)¶

Every task-capable server exposes these three tools (added by register_task_tools, §4.5). 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.5 Environment variables (consumed by the kit; each server's launcher sets them)¶

Var	Meaning	Default
`RVC_CALLBACK_URL`	The brain's events endpoint, `<brain origin>/api/agent/events` (see the port warning in §3.1). 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 brain's attach-manifest entry for this server. Required if `RVC_CALLBACK_URL` is set.	unset
`RVC_USAGE_URL`	The brain's usage-ingest endpoint, `<brain origin>/api/agent/usage` (§3.2). Unset ⇒ `UsageReporter` is a no-op — the return-field channel (`UsageCollector`) needs this unset.	unset
`RVC_USAGE_SECRET`	Fleet-wide HMAC secret for `UsageReporter`, matching the brain's own `RVC_USAGE_SECRET`. Not the per-server pair above — see §3.2 for why. Required if `RVC_USAGE_URL` is set.	unset

None of these four are wired by brainctl or r1ctl today — each is a robot-side server's own env var to set, matched by value against the corresponding brain-side variable (documented on Running a brain). RVC_USAGE_SECRET is the one pair that is actually used in the current stack; nothing attached today pushes a TaskEvent (Integrating a robot §3f).

4. Public API¶

Everything below is re-exported from robot_mcp_kit/__init__.py, in six groups: the async task contract (§4.1–4.2), events + signing (§4.3), usage reporting (§4.4), the FastMCP task tools (§4.5), the robot client (§4.6), learning + lessons (§4.7), and recovery (§4.8).

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.usage` and `robot_mcp_kit.usage_report` (BIN-240 phase 2)¶

# usage.py — one shape every billable call in the stack reports
class UsageRecord(BaseModel):
    """See the module docstring for the full field list. `cost_micro_usd is None` and
    `confidence == "unpriced"` imply each other, enforced by a validator — a reporter sends
    tokens plus identity, never a price."""

# usage_report.py
class UsageSink(Protocol):
    async def report(self, record: UsageRecord) -> None: ...

class UsageCollector(UsageSink):
    """Accumulates records in memory. drain() -> list[dict] (JSON-ready), for attaching to a
    tool result under a `usage` key — the return-field channel, needing no configuration."""
    def drain(self) -> list[dict[str, object]]: ...

class UsageReporter(UsageSink):
    def __init__(self, service_name: str,
                 report_url: str | None = None,   # defaults from RVC_USAGE_URL
                 secret: str | None = None):        # defaults from RVC_USAGE_SECRET
        """Same shape as EventPoster (queue, worker, signed retry) — diverges in one place: a
        4xx other than 429 is terminal, not retried. report_url None -> no-op mode."""

    async def report(self, record: UsageRecord) -> None: ...
    async def aclose(self) -> None: ...

RecoveryAgent (§4.8) takes a usage: UsageSink | None = None constructor argument — whichever sink a caller passes, it reports one UsageRecord per controller call, unpriced, with feature="recovery".

4.5 `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."""

4.6 `robot_mcp_kit.client` — the robot client¶

The client side of the kit: a Layer-3 skill builds its competence by calling other servers' tools (a grasp-service drives the robot's primitives with it). Needs the mcp extra; imports are deferred so import robot_mcp_kit stays usable without it.

class ToolClient:
    def __init__(self, url: str, *, headers: dict[str, str] | None = None,
                 timeout_s: float = 180.0, connect_timeout_s: float = 15.0,
                 retries: int = 2, retry_backoff_s: float = 0.5,
                 transport: str | None = None): ...

    async def call_tool(self, name: str, arguments: dict | None = None) -> dict[str, Any]:
        """Call a tool and parse its result as a JSON object."""
    async def call_text(self, name: str, arguments: dict | None = None) -> str:
        """Raw text result; retries transient connect/transport failures."""
    async def list_tools(self) -> list[dict[str, Any]]:
        """`{name, description, input_schema}` per tool (the RecoveryAgent needs the schemas).
        Best-effort — `[]` if unreachable."""
    async def fetch_prompt(self) -> str:
        """Read the server's `prompt://system` fragment (best-effort, never raises)."""

Design — per-call connect. Each call opens a fresh session, invokes the bare-named tool, and closes, all in one task (no anyio cancel-scope pitfalls; self-healing across server restarts). Transport is auto-selected from the URL — a path ending /mcp → streamable-HTTP, else SSE — or set transport= explicitly. That matches the fleet's split: galaxea's servers speak streamable-HTTP at /mcp, binabik's speak SSE at /sse.

Errors (all subclass ToolCallError):

Exception	Meaning	Retried?
`ToolCallError`	Call failed after retries, or the result was not the required JSON object.	—
`ToolReportedError`	The tool ran and reported an error (`isError`).	No — re-raised at once.
`ToolTimeout`	Invoked but no return within `timeout_s` (a wedged motion).	No — fail fast to a higher tier.

4.7 `robot_mcp_kit.learning` — fleet learning + lessons¶

The default → override → learn store behind self-improvement. A Signature keys each lesson to a hierarchical, environment-aware context; ParamStore.recall applies the most-specific confident lesson to bias the next attempt, and record earns generalization over time.

def signature(label: str, dims: Mapping[str, float] | None = None,
              place: str = "") -> Signature: ...
# Signature.keys() → the recall backoff order, most-specific first:
#   L1 obj+size+place (env-specific)  →  L2 obj+size (earned across ≥ promote_places)
#   (L3 shape+size is a reserved future tier)

class Override:          # a learnable adjustment: approach / side / metre offsets
    approach: str | None; side: str | None; offsets: Mapping[str, float]
class Recall:            # what recall() returns: override + source_key + level + streak
    ...

class ParamStore:
    def __init__(self, backend: Backend | None = None, *, enabled: bool = True,
                 confirm_n: int = 3, promote_places: int = 2, history: int = 12): ...
    def recall(self, sig: Signature) -> Recall | None: ...
    def record(self, sig: Signature, override: Override, success: bool) -> dict[str, Any]:
        """Confirm after confirm_n wins; promote L1→L2 across promote_places distinct places;
        demote a confident lesson that then fails. No-op when disabled."""
    @property
    def enabled(self) -> bool: ...
    def set_enabled(self, on: bool) -> None: ...
    def status(self) -> dict[str, Any]: ...
    def lessons(self) -> list[dict[str, Any]]: ...

Backend is a get/put/keys Protocol with two impls: InMemoryBackend and SqliteBackend (BIN-549). grasp-service uses the SQLite backend for a local store; the central robot-mcp-memory composes one ParamStore per scope for fleet-shared learning.

backend = SqliteBackend.open("data/lessons.db")              # owns its store; call .close()
backend = SqliteBackend(existing_store, scope="group:zurich")  # shares a service's store

JsonFileBackend was removed in BIN-549 — it rewrote the whole file on every put, into params/<scope>.json, which under BIN-546 is not somewhere we write. There was no migration and none was owed: measured 2026-09-09, neither that directory nor grasp_lessons.json had ever been written to in production, six weeks after BIN-183 said the same. scope is now a column, so one table with a (scope, key) primary key serves every fleet scope with one connection, one retention policy and one backup. A put is synchronous and undroppable — the store's queue drops on overflow, which is right for a diagnostic and wrong for a lesson, since a dropped lesson is silently relearning it. Retention on the table is declared and off by default, for recoveries' reason: a lesson is cheap to keep and expensive to re-learn.

Which store wins is unchanged. MEMORY_URL still decides: set, and learning is group-scoped and fleet-shared through robot-mcp-memory; unset, and it is a local store gated by GRASP_SELF_IMPROVE. BIN-549 changed only what the local half writes to (BIN-183).

4.8 `robot_mcp_kit.recovery` — the recovery sub-agent¶

When a skill's fast static path fails, it "converts into a sub-agent": a bounded LLM tool-use loop over the skill's own tool clients, verifying after each call.

class Decision:          # one controller step: call tool(args), declare done, or give_up
    tool: str | None; args: dict[str, Any]; done: bool; give_up: bool; reason: str

class RecoveryAgent:
    def __init__(self, goal: str, clients: Sequence[ToolClient], *,
                 decide: DecideFn | None = None, verify: VerifyFn | None = None,
                 model: str | None = None, max_steps: int = 8, deadline_s: float = 180.0,
                 progress: ProgressCB | None = None): ...
    async def run(self) -> dict[str, Any]:
        """Drive the tools toward `goal`, verifying with `verify`. Returns
        {ok, steps, reason, winning?} — `winning` is the last clean tool call, so the caller
        can *learn* what fixed it. Bounded (step budget + deadline); abort-safe (only catches
        ToolCallError, so a CancelledError from the brain unwinds the loop at once)."""

def recovery_backend_status() -> str | None:
    """None if the DEFAULT controller can run, else why not — for a startup check.
    Resolves the module spec only: no SDK import, no network."""

decide is injectable — the default calls the planner model (RECOVERY_MODEL, default claude-opus-4-8); tests inject a fake and need no network. It lists each server's tool schemas (ToolClient.list_tools) and lets the controller pick one tool at a time.

run() never raises because the LLM backend is unhappy (BIN-156, ≥ 0.7.0)

A recovery tier is reached only after something already went wrong, so a rung that can raise takes down the very skill it exists to rescue. The default controller therefore turns every backend failure into a give-up Decision, logged at ERROR: the anthropic package missing, no key in the environment, or an auth / rate-limit / network error mid-run. The caller gets {"ok": false, "reason": "recovery backend unavailable: …"} and keeps its own failure reason. CancelledError is a BaseException, so a brain abort still propagates.

The import was always deferred; its failure wasn't handled, and that was enough — on rap-1 every grasp() returned Error executing tool grasp: No module named 'anthropic'.

Two consequences for consumers:

Ask recovery_backend_status() at startup if your recovery tier is enabled by config, so an unusable backend is reported at launch rather than at the first failed skill run. It distinguishes SDK not installed from no key, because the fixes differ.
Declare the recovery extra (robot-mcp-kit[recovery] → anthropic>=0.40) when your recovery tier is on by default — grasp-service does, since GRASP_POLICY=tiered is its default. A caller injecting its own decide (the brain) needs nothing.

4.9 `robot_mcp_kit.store` — the shared store (BIN-546/BIN-549)¶

Everything a Binabik service writes goes in a database, never a JSON or markdown file. That standing rule is BIN-546, stated in full on the brain app's spec, which is also where the engine choice (SQLite, one file per service) and the config-versus-data test are argued. This section is the mechanism four repos share, lifted here from the brain's BrainStore (BIN-531/BIN-547) so robot-mcp-memory, r1-abstraction and every skill service consume it instead of reimplementing it.

class Kind(str, Enum):   # RECORD | DEFINITION | BELIEF | DERIVED — a str mixin, not StrEnum (3.10)
    ...

@dataclass(frozen=True)
class Schema:
    version: int                                  # only an ALTER earns a bump; a new table does not
    ddl: str                                      # idempotent, replayed on every open
    kinds: Mapping[str, Kind]                     # every table declares which kind it holds
    pruned: frozenset[str] = frozenset()          # retention's reach — records only
    migrations: Mapping[int, tuple[str, ...]] = {}  # keyed by the version each step upgrades TO
    ignored: frozenset[str] = frozenset()         # present, and not ours to give a kind
    columns: Mapping[str, Mapping[str, Kind]] = {}  # per-COLUMN kinds, opt-in per table (BIN-574)

    def kind_of(table) / all_kinds() / tables_of_kind(kind)
    def columns_of_kind(table, kind) -> frozenset[str]   # a row's belief half, addressable
    def not_ours() -> frozenset[str]        # `ignored` + the fts5 shadows of every DERIVED table

def undeclared_tables(present: Iterable[str], schema: Schema) -> frozenset[str]: ...
def undeclared_columns(present: Mapping[str, Iterable[str]], schema: Schema) -> frozenset[str]: ...
def conservative_kind(kinds: Iterable[Kind]) -> Kind    # RECORD < BELIEF < DEFINITION
def fts5_shadow_tables(indexes: Iterable[str]) -> frozenset[str]

class SqliteStore:
    def __init__(self, db_path: str | Path, schema: Schema, *, name: str = "store"): ...
    def enqueue(self, sql, params=()) -> None:    # writer thread, batched, DROPPABLE
    def execute(self, sql, params=()) -> int:     # caller's thread, undroppable
    def query_one(...) / query_all(...) / transaction() / flush(timeout_s)
    def prune(self, table, sql, params=()) -> int:  # the one chokepoint; raises NotPrunable
    def table_names(self) -> frozenset[str]
    def column_names(table) -> tuple[str, ...]    # live PRAGMA table_info, never the DDL
    def columns_present() -> dict[str, tuple[str, ...]]   # for the tables that declare columns
    def checkpoint() -> int / wal_bytes() -> int / vacuum() / count(table) / close()

def run_startup(store, *, policy=NO_RETENTION, summarise=None,
                migrated=None, extra=()) -> BootSummary: ...

The four kinds decide cleanup: a record (what happened) is pruned by age or by count; a definition (what a human authored) is removed only by an explicit human act; a belief (an observation with a validity window) is not persisted unless stamped with the epoch it was observed in, and a foreign epoch reads as unknown rather than as old (BIN-533); a derived table (a search index) is a projection of another with no content of its own, so it is never pruned and its owner rebuilds it.

DERIVED is not a nicer word for "unprunable" — DEFINITION was already that, and is what both consumers declared while the vocabulary was three. It says the thing no other kind can: pruning a projection independently of its source is a bug in either direction, and losing one is recoverable, so a backup may skip it and a corrupt one is fixed by a reindex rather than a restore (BIN-574; robot-mcp-memory's processes_fts is the case).

A kind is declared per table, and per column where one row holds more than one. Schema.kinds is the declaration retention reads, because a DELETE takes a whole row. Schema.columns is the finer, opt-in per table declaration for the row that mixes kinds, and r1-abstraction's waypoints is the one that needs it: name a definition a human authored, x/y/yaw_deg/ session a belief valid only in the bring-up that observed it, saved_at a record. The table-level kind must be exactly conservative_kind() of the columns — strongest wins under RECORD < BELIEF < DEFINITION — and Schema refuses a declaration where it is not, so the two granularities cannot drift. DERIVED is deliberately not on that scale and a projection may not declare column kinds: it has no content of its own, so a row mixing it with authored content would be a contradiction rather than a conservative case.

Two properties make that enforceable rather than aspirational, and both are mechanisms:

A table — or a column — that declares no kind fails a test. undeclared_tables(store.table_names(), SCHEMA) and undeclared_columns(store.columns_present(), SCHEMA) are pure functions over what the file holds (sqlite_master, PRAGMA table_info) — not over the DDL string, which would only confirm that a constant agrees with itself — so a consumer's test is two lines each. The store also logs a WARNING at boot for each, which is the case no test is watching: one that arrives on a deployed file from a stale build or a hand-run CREATE/ALTER. A table with no entry in Schema.columns is checked at table granularity and nothing demands otherwise; a consumer whose guard is the bare two-liner should assert its premise too, since an opt-in map that was dropped makes the guard pass while checking nothing.

An fts5 index needs no exemption: declare it DERIVED. sqlite creates five shadow tables behind one (<index>_data, _idx, _content, _docsize, _config), as ordinary tables without the sqlite_ prefix that would have covered them, so an undeclared-table guard would demand a declaration for five tables nobody wrote on every boot — and a warning that always fires is how a real one goes unread. Schema.not_ours() derives those names from the declaration; Schema.ignored remains for anything else a file holds that is somebody else's. Suffixes rather than a prefix rule, so a genuinely new table named like a shadow one (processes_fts_extra) is still caught. 2. Retention may only delete records. Schema refuses at construction to declare a definition, a belief or a projection prunable, and SqliteStore.prune — the one chokepoint every retention path goes through — raises NotPrunable for a table the declaration never covered, saying what to do instead per kind: an operator deletes a definition, an epoch invalidates a belief, an owner rebuilds a projection. The two guards catch different mistakes: a bad declaration versus a prune aimed at the wrong table. execute is deliberately unguarded, because an operator deleting their own plan is exactly the explicit human act the taxonomy allows.

The write path is BIN-203's, unchanged: enqueue never touches sqlite on the caller's thread (a synchronous insert from a log processor once had brain-pascal serving a zero-IO REST handler in 10 s), batches consecutive identical statements into one executemany, and drops on a 10 000-deep queue rather than becoming backpressure on the thing it is recording — dropped counts and close warns, so the loss is bounded and visible. Cold paths run synchronously and are not droppable. WAL mode carries journal_size_limit = 1 MiB, because sqlite's default of −1 reuses a checkpointed journal in place and never shrinks it: that is how both live brains were measured at a ~4.1 MB WAL beside a 1.0 MB database.

Versioning refuses rather than corrupts. store_meta.schema_version decides whether a build may open a file: older migrates forward through Schema.migrations, in ascending order and before the DDL is replayed; newer raises SchemaTooNew, because a rolled-back image writing through a schema it does not understand is how a column silently becomes garbage. Adding a table is not a bump — the DDL is replayed on every open, so CREATE TABLE IF NOT EXISTS reaches a deployed file; only an ALTER earns a version, being the one thing an idempotent CREATE cannot do.

run_startup prunes, counts and emits one line — store_ready path=… schema=0->1 loaded={…} invalidated=0 wal_bytes=0 — so every service's boot log reads the same sentence, and a disappearance is noticed at a glance rather than by a service quietly getting worse. loaded and invalidated are always present: "nothing was loaded" and "the field is missing" must not look the same in a log.

What deliberately did not travel is the brain's legacy.migrate_legacy. BIN-546 struck data migration from the whole programme — every port creates its table and drops its file reader, so there is nothing to read once and park. run_startup still accepts a migrated= mapping, so a service that folded something in of its own reports it on the same line.

The CI guard travels as a copy, not as a call. tests/test_writes_go_to_the_database.py walks src/ and fails on a write-shaped call outside a dict-with-reasons allowlist; it imports nothing from the application, so a repo's copy needs exactly two edits — SRC and ALLOWED. It stays a copy because a shared helper would need the kit as a test dependency of repos that already have it as a runtime one, and because each repo's allowlist is the part worth reading. The kit's own allowlist has two entries, both the key store: keys.py's credential cache and cli.py's identity writer. That exemption is deliberate and now written down (BIN-549 item 3) — the cache must be readable at the earliest point of startup, before any store class exists, and its job is to work when everything else is unavailable; putting it behind a database would put the credential path behind the machinery most likely to be what failed. The file-versus-table choice does not change the "not encrypted at rest" position, which is a separate question (BIN-426).

Compatibility¶

robot-mcp-kit is an internal fleet library, not published to any index. Each consumer pins it by git commit in its uv.lock (robot-mcp-kit @ git+https://github.com/binabik-ai/robot-mcp-kit@<sha>), so upgrades are deliberate and reproducible — and since BIN-142 a consumer can pin a tag (tag = "v0.7.0") instead when it wants an explicitly named release. A consumer's declared floor (robot-mcp-kit[mcp]>=X.Y) should name the version that actually introduced the symbols it imports — grasp-service's [mcp,recovery]>=0.7 is recovery_backend_status() + the recovery extra, for instance — not a historical lower bound. Because the whole fleet shares one wire contract (§3) and one public API (§4), breaking changes are coordinated across the fleet: bump SCHEMA_VERSION, land the library change, then advance every consumer's pin together (the brain endpoint and the capability servers must agree on the event envelope + signing).

A declared floor is not a guarantee — the import is

Only robot-voice-chat pins the kit by commit today; the other six track branch = "main", so a removed export reaches them on their next uv lock, with no version gate. BIN-563 measured the failure mode directly: binabik-r1-vision's manifest said >=0.8.0 while its lock resolved a 0.8.0 commit predating the helper it needed, and only from robot_mcp_kit import <name> failing at import caught it. So a consumer's floor is a statement of intent, and the thing that actually proves compatibility is importing the symbol.

BIN-549's removal of JsonFileBackend was the worked example, and BIN-573 closed it. grasp-service (server.py:111) and robot-mcp-memory (params.py:46) constructed it at module scope, so for them the failure was an ImportError at container start rather than a red test; their locks pinned a pre-removal commit, so both ran, and the fuse was whichever routine uv lock came first. Verified rather than assumed: against kit main, the old grasp-service/server.py raises ImportError and its suite reports 5 collection errors.

Both call sites now use SqliteBackend, both manifests floor at >=0.9.0 or above, and the version was cut to 0.9.0 — which is the part worth keeping in mind. While the package said 0.8.0 the incompatibility was unsayable: robot-mcp-kit>=0.8.0 is satisfied by both the kit that exports a symbol and the kit that does not. A removed export is breaking, so it earns a minor bump, and bumping is what turns "it broke when I relocked" into a resolver error.

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:

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.
Triggers: eta_below-style trigger fires exactly once; repeat: true fires repeatedly; unknown types ignored by progress_above auto-evaluation.
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.
Signing: round-trip sign/verify; rejects bad signature, stale timestamp, tampered body.
register_task_tools: tools appear in list_tools(); get_task/cancel_task behave per §3.3 (use the in-memory FastMCP test client).
UsageCollector/UsageReporter (BIN-240 phase 2): drain() round-trips through JSON exactly as model_validate_json expects it back; UsageReporter signs correctly, retries a 5xx/429, but a 4xx other than 429 is terminal — not retried, unlike EventPoster; no-op mode when unconfigured. The grasp recovery tier reports one unpriced record per controller call, under feature="recovery", and a raising sink never fails the recovery it is reporting on.

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, with no server runtime. The layout, tooling and architecture conventions apply to it; the MCP-server ones describe the servers that consume it, not this repo. It takes no config of its own — a caller passes values in — so the Hydra material does not apply to it either.

The full list — repo structure, uv/ruff/mypy/pytest tooling, the pre-commit hook set, ports-and-adapters, the two config shapes (Hydra where implementations are selectable, a frozen env dataclass where they are not), secrets, structlog, the CI gate, and what an MCP server owes its consumer (transport choice, blocking tools, the ~30 s call timeout, reconnect tolerance, the hardened systemd unit) — is Coding guidelines. It lives there and only there: this appendix was one of five near-identical copies, and they drifted from each other and from the code.