Skip to content
AuthorPascal DateJuly 14, 2026 Rev1.0

mcp-perception-buffer — Implementation Spec

mcp-perception-buffer is the brain's "what do you see" server. It subscribes to the /vision/** ROS topics published by the robot-mcp-perception node, keeps a rolling latest snapshot in memory, and exposes read-only MCP tools so the brain can ask "what do you see?" without a round-trip into the perception pipeline. It is a ROS node that also serves MCP, so the brain itself stays ROS-free.

See AGENT_ORCHESTRATION_CONCEPT.md §4.2 and specs/robot-mcp-perception.md §4 for the topics/messages this subscribes to.

1. Context

  • The robot-mcp-perception node publishes, per active skill: /vision/{skill}/detections (VisionDetectionArray: label/score/box/mask/point), /vision/{skill}/{output} (StampedString: semantic text/number/bool/list), and /vision/{skill}/overlay (annotated image).
  • The brain shouldn't poll the pipeline or carry pixels. This server subscribes once to those topics and keeps the latest value of each, so what_do_you_see() is an instant in-RAM read.
  • Read-only and pull-first. It never configures anything (that's mcp-skill-control). It may push watch_events for watcher skills (§5).

Core buffer pattern: a fixed-size rolling history (deque(maxlen=N)), a freshness/age check per entry, a compact human-readable rendering for prompt injection, and graceful handling of stale/missing data.

2. Repo layout

mcp-perception-buffer/               (needs rclpy at runtime)
├── package.xml / setup.py
├── config/
│   └── buffer.yaml                  # topic patterns, history depth, freshness window, watch push
├── src/mcp_perception_buffer/
│   ├── server.py                    # MCP Server + rclpy node + dynamic subscriptions
│   ├── buffer.py                    # rolling snapshot store (latest + short history per stream)
│   ├── render.py                    # compact "scene snapshot" text (state_collector format)
│   └── watch.py                     # watch-predicate evaluation → robot-mcp-kit event poster (§5)
└── tests/

Dependencies: mcp, rclpy, ros2_vision_interfaces (the VisionDetection* msgs), robot-mcp-kit (event poster + signing, for §5). Transport: stdio or sse (MCP_TRANSPORT=sse, MCP_PORT default 9202). MCP server name perception → tools perception.what_do_you_see, etc.

3. Subscription model

  • On startup, subscribe to a configurable set of topic patterns (default /vision/**). Because ROS has no wildcard subscribe, the node periodically reconciles against ros2 topic list (or a graph-event callback): for every /vision/{skill}/detections|{output}|overlay topic seen, ensure a subscription exists; drop subscriptions for topics gone silent past the freshness window.
  • For each topic keep the latest message plus a deque(maxlen=history_depth) of recent values and last_update_ts. Masks/overlays are stored by reference (topic + stamp), never copied into the snapshot text — the brain gets "there is a mask on /vision/find_object/detections", and a ROS consumer fetches the pixels.
  • Optional camera_topic subscription for an "is the camera alive / last frame age" health field.

4. MCP tools (read-only)

Tool Args Returns
what_do_you_see() — a compact JSON+text snapshot across all active skills: per skill the latest semantic outputs (scene description, object list, booleans) and a one-line summary of detections (labels + boxes + scores, mask by reference). The prompt-ready rendering from render.py.
get_latest(skill) skill name the latest VisionDetectionArray for that skill as JSON (label/score/box/point; mask_ref = {topic, stamp}), plus its semantic outputs.
list_streams() — {streams:[{topic, skill, kind, last_update_age_s}]} — what's live and how fresh.
get_overlay_ref(skill) skill name {topic, stamp} for the annotated image (for a UI/consumer to fetch; not pixels over MCP).

Freshness

Every value carries age_s; entries older than freshness_window_s (config, default 5 s) are marked stale: true so the planner doesn't act on dead data.

5. Watchers → push events (owns the watch logic; concept §4.2, skill-control §5 option A)

mcp-skill-control records a watch intent on watcher skills (e.g. watch_for(condition) → {output: triggered, equals: true}). Since this server is already subscribed to that skill's topic, it evaluates the predicate on each incoming message and, on a rising edge, POSTs a watch_event to the brain's POST /api/agent/events using robot-mcp-kit (HMAC-signed, the same Async Task Contract the brain already consumes):

type: "watch_event"
task_id: "<skill_id>"
server: "perception"
payload: { skill, condition, value, detection_summary, overlay_ref }

How does this server learn which skills are watched and their predicates? It reads them from list_vision_tasks (the robot-mcp-perception skill carries the watch block), which keeps it fully decoupled: the vision node is the single source of truth for what's running, and this server already reconciles against it.

event_secret (HMAC) + the brain callback URL are configured here exactly like any robot-mcp-kit-posting server (see specs/robot-mcp-kit.md / brain tools/mcp.yaml event_secret).

6. Config & wiring

# config/buffer.yaml
topic_patterns: ["/vision/"]        # prefixes to auto-subscribe under
history_depth: 10
freshness_window_s: 5.0
reconcile_interval_s: 2.0
camera_topic: /camera/image_raw     # optional health probe
watch_push: true                    # enable §5
Env Meaning Default
MCP_TRANSPORT / MCP_PORT stdio | sse / port stdio / 9202
RVC_CALLBACK_URL / RVC_EVENT_SECRET brain event endpoint + HMAC (for §5) unset → push disabled
ROS_DOMAIN_ID ROS domain inherit

Brain tools/mcp.yaml:

  - name: perception
    transport: sse
    url: "http://127.0.0.1:9202/sse"
    event_secret: ${oc.env:PERCEPTION_EVENT_SECRET}   # only if watch_push

The brain may also inject what_do_you_see() into the planner/system context at mission start, so "the robot knows what it sees" without an explicit tool call — an executor-side convenience.

7. Tests

  1. buffer: feed synthetic VisionDetectionArray + StampedString messages; latest value + history retained; age_s/stale computed against a fake clock.
  2. reconcile: a new /vision/{skill}/detections topic appearing creates a subscription; a silent topic past the window is dropped from list_streams.
  3. tools: what_do_you_see renders semantic outputs + a detection summary with mask by reference (no pixels); get_latest returns the structured detection JSON; stale data flagged.
  4. watch (§5): a rising edge on a watched skill's output POSTs a correctly-signed watch_event (assert with robot_mcp_kit.signing.verify); no event while the condition stays false; no push when watch_push: false.

8. Acceptance checklist

  • With robot-mcp-perception running a describe_scene skill, what_do_you_see() returns the live scene description + object list within the freshness window.
  • After instantiate_skill("find_object", …) (via mcp-skill-control), get_latest("find_object") returns detections with a mask_ref (topic+stamp), never pixels over MCP.
  • A watch_for skill flipping true POSTs one signed watch_event to the brain.
  • Buffer survives the vision node restarting (subscriptions reconcile; stale entries flagged).

Appendix A — Repository & tooling conventions

Standard "how we build repos here". This is an MCP server that is also an rclpy node — all sections apply; for the ROS side, mirror robot-mcp-perception's ament packaging.

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.