AuthorPascal DateJuly 14, 2026 Rev1.0

robot-mcp-perception — Implementation Spec (the vision module)¶

Repo: robot-mcp-perception (github.com/binabik-ai/robot-mcp-perception).

What it is: the robot's single configurable vision module. A grounder (Gemini Robotics-ER) decides what/where; a segmenter (SAM 3) produces pixels. It is configured at runtime via skills and publishes results on the ROS data plane (/vision/** topics) so other ROS modules (manipulation, and so on) consume masks directly — the brain is never in that path.

This is the data-plane vision node of the control/data split (AGENT_ORCHESTRATION_CONCEPT.md §1, §3). It is reached only through two MCP bridge servers: mcp-skill-control (configure skills, via this node's ROS services) and mcp-perception-buffer ("what do you see", via this node's ROS topics). The brain stays ROS-free.

Models: grounder = Gemini Robotics-ER (gemini-robotics-er-1.6-preview, strong pointing/boxes); segmenter = SAM 3 (sam3.pt; sam2.1_* as faster fallbacks). Both are wired in config/grounder/gemini_er.yaml + config/segmenter/sam.yaml. The pluggable ports keep model swaps to a one-line config change (SAM ↔ SAM2, or ER ↔ grounded-SAM). Groq llama-vision is not a grounder here — its localization is insufficient.

1. System context (read once)¶

The conversational/mission brain (robot-voice-chat, ROS-free, MCP client only) configures the ROS stack through MCP bridge servers and observes it; ROS modules talk to each other over ROS. This node is the perception capability. It is configured at runtime — skills are created/managed/ listed over ROS services, never hardcoded — and publishes results to ROS topics.

/camera/image_raw ─► ┌──────────────── robot-mcp-perception (ROS node) ───────────────┐
                     │  grounder: Gemini ER ─┐  router (per skill mode, §5)            │
                     │  segmenter: SAM 3 ────┴─► reason / tiled / watch pipelines      │
                     └───────────────┬───────────────────────────┬────────────────────┘
   ROS services (configure):         │ ROS topics (results):      │
   ~/create_vision_task              ▼                            ▼
   ~/manage_vision_task     /vision/{skill}/detections   /vision/{skill}/{output}
   ~/list_vision_tasks      (VisionDetectionArray:        (StampedString: text/
                             label,box,mask,point)         number/bool/list)
        ▲                            ▲                            ▲
        │ ROS srv                    │ ROS sub                    │ ROS sub
  mcp-skill-control          mcp-perception-buffer        (future) manipulation node
  (brain's hands)            (brain's eyes)                consumes masks DIRECTLY

This node exposes no Async Task Contract surface. A long-running perception capability is a standing skill publishing to a topic; the brain observes the buffer. watch_event push (a watcher firing) is owned by mcp-perception-buffer (specs/mcp-perception-buffer.md §5), which uses robot-mcp-kit to POST to the brain.

1.1 ROS distro / packaging¶

ROS 2 Humble+, rclpy. ament_python package robot_mcp_perception. Heavy deps (torch, SAM 3, google-genai) via pip in an [sam] / base extras split; CUDA Docker base for the GPU build. The node must also run camera-free on frame_source: folder for dev/CI.

1.2 Where the node runs (on-robot vs off-robot)¶

This node is the biggest consumer of the camera stream, so its placement is a data-plane decision (architecture.md §3a). Two shapes are supported, and frame_source is the switch:

On-robot (default) — frame_source: ros. The node lives on the robot, subscribes the ROS camera topic directly, and reaches out to a cloud model only where a stage's cost_class is cloud (today: Gemini-ER grounding; §8). This is Pattern A: only per-call frames leave the robot; the ROS graph never does.
Off-robot — frame_source: rtsp (or v4l2/folder). If the robot's onboard GPU is too small for SAM 3, run the whole node on a cloud / LAN GPU box and point it at a streamed camera (a thin robot-side RTSP/WebRTC egress bridge feeds it) instead of a local ROS topic. This is Pattern B: the consumer moves off-robot, but a robot-side process still publishes the stream, so ROS itself stays robot-local. Prefer a GPU box on the robot's own LAN over the public internet for anything latency-sensitive.

Either way the node's ROS/MCP contract (§6, and the bridge servers fronting it) is identical — only where the process runs and how frames reach it change. A VLA wrapped as an MCP server follows the same two patterns.

2. Repo layout¶

robot-mcp-perception/
├── package.xml                      # ament_python; rclpy, sensor_msgs, ros2_vision_interfaces, …
├── setup.py / pyproject.toml        # console_scripts: vision_node = robot_mcp_perception.vision_node:main
├── README.md  CHANGELOG.md  .pre-commit-config.yaml  .github/   # Appendix A conventions
├── config/                          # Hydra tree
│   ├── config.yaml                  # defaults: grounder/segmenter/frame_source + pipeline/watch knobs
│   ├── grounder/   gemini_er.yaml  grounded_sam.yaml
│   ├── segmenter/  sam.yaml  none.yaml
│   └── frame_source/ ros.yaml  rtsp.yaml  v4l2.yaml  folder.yaml
├── src/robot_mcp_perception/
│   ├── vision_node.py               # the ROS node — camera sub, skill registry, N-Hz loop,
│   │                                #   ROS services (create/manage/list), ROS publishers
│   ├── skills.py                    # Skill/SkillOutput dataclasses + JSON (de)serialize (§3)
│   ├── prompt_builder.py            # grounding + semantic-output JSON
│   ├── composition.py               # build grounder/segmenter/frame_source from Hydra
│   ├── frames.py                    # FrameSource ABC + ros/rtsp/v4l2/folder (§4)
│   ├── models/                      # base.py, gemini_er.py, sam.py, grounded_sam.py (§4)
│   ├── pipelines/                   # reason.py, tiled.py, watch.py, common.py (§5)
│   ├── postprocess.py  viz.py  artifacts.py   # masks postproc, overlay, optional artifacts
│   └── publishers.py                # per-skill ROS publisher set + msg packing (§6)
├── deploy/robot-mcp-perception.service
├── Dockerfile                       # CUDA base for SAM 3
└── tests/{unit,integration}/

The interfaces package ros2_vision_interfaces (its own repo or a sibling ament package) holds the messages + service (§6.2). All of robot-mcp-perception, mcp-skill-control, and mcp-perception-buffer depend on it.

2.1 Design patterns¶

The module builds on patterns from the ros2-vlm-groq sample:

Dynamic task system: a task/skill = {name, description, outputs[]} created/enabled/ disabled/deleted at runtime via ROS services (create_vlm_task → create_vision_task here), with publishers created on the fly per skill. This is the "configure the ROS stack" mechanism.
Prompt builder: turns a skill's grounding + outputs[] into one prompt and parses a single JSON response ({"results":[{"name","value"}]}). Semantic output types: text | number | boolean | list | position | pose.
Pipelined async processing: N worker tasks, max_in_flight bound, drop-oldest when the queue is full, per-skill rate_hz, futures resolved on a callback. Keeps the node real-time.
Timestamp preservation: every published result carries the source frame's stamp + frame_id.
The StampedString per-output topic convention for semantic outputs.

The Groq provider/model (llama-3.2-*-vision) is not used — its localization is insufficient. The grounder is Gemini ER, and semantic outputs come from ER. The sample's MCP wrapper is also not carried over; that role belongs to the separate mcp-skill-control server.

3. Skill = the unit of configuration¶

A skill is a declarative config (JSON, passed to create_vision_task). It extends the ros2-vlm-groq task with a mode, a grounding prompt, and a segment flag:

{
  "name": "find_object",            // unique; topic namespace /vision/find_object/*
  "description": "Locate and segment a queried object.",
  "mode": "ground_and_segment",     // describe | ground_and_segment | segment_all | region
  "grounding": "the scissors lying on top of the pile",  // ER what/where prompt (grounding modes)
  "segment": true,                  // run SAM 3 on the grounded target(s)
  "outputs": [                       // OPTIONAL semantic side-channel (from ER, via prompt_builder)
    {"name": "found", "type": "boolean", "description": "whether the object is visible"}
  ],
  "rate_hz": 2.0,                   // per-skill processing rate (default = node param)
  "once": true                     // one-shot: run once, publish, auto-disable (brain's find_object)
}

skills.py: Skill / SkillOutput dataclasses with from_json / to_json + validation (unique name, valid mode, valid semantic types). once: true is how the brain does a one-shot find_object (instantiate → read one /detections → auto-stop); once: false is a standing skill that keeps publishing for a ROS consumer (manipulation).

Three representative instructions map onto one node, three skills:

Instruction	mode	grounding	segment	publishes
"detect the scissors lying on top"	`ground_and_segment`	"the scissors on top of the pile"	yes	mask+box+point of that object
"segment anything"	`segment_all`	—	yes	a mask per instance
"where can I put the package"	`region`	"free flat space large enough for the package"	no	placement rectangle

4. FrameSource + model adapters¶

FrameSource (ABC; ros default on-robot, folder for dev/CI, rtsp/v4l2 options) yields the latest EXIF-corrected, downscaled frame. Grounder / Segmenter ABCs in models/base.py:

@dataclass
class GroundedTarget:
    label: str
    box: tuple[int,int,int,int] | None      # xyxy px
    points: list[tuple[int,int]]            # ER pointing
    kind: Literal["object","region"]
    confidence: float | None

class Grounder(ABC):
    async def ground(self, image, question: str) -> GroundingResult: ...   # answer+reasoning+targets
    async def point(self, image, query: str) -> list[GroundedTarget]: ...   # tiled pointing

class Segmenter(ABC):
    def segment_box(self, image, box) -> np.ndarray: ...      # bool mask, full-image
    def segment_points(self, image, points) -> np.ndarray: ...

GeminiERGrounder (gemini-robotics-er-1.6-preview, thinking on; GEMINI_API_KEY; retry + timeout). The grounder for all grounding modes; also produces the semantic outputs[].
SamSegmenter (SAM 3 sam3.pt; sam2.1_b.pt fast fallback; GPU; runs in a single-worker thread pool, GPU access serialized). NullSegmenter (segmenter=none) returns box-shaped masks for no-GPU dev.
GroundedSamGrounder is the fully-local fallback grounder (no cloud).

Exact model IDs, prompts, tile/crop parameters, and plausibility thresholds live in the repo's playground/ and example configs; treat those as the source of truth for parameters.

5. Pipelines → modes¶

The router selects a pipeline per skill. pipelines/{reason,tiled,watch}.py map onto skill modes:

`mode`	grounder	segmenter	pipeline	publishes
`describe`	yes (answers `grounding`/`outputs`)	—	`reason` (no-seg)	semantic `StampedString` only
`ground_and_segment`	yes → point/box	yes (SAM 3: crop→box-prompt, margin `crop_margin`; fallbacks full-image box→point; plausibility check)	`reason` (object)	`detections` + semantic outputs
`segment_all`	per-tile pointing	yes (SAM once on full image; point-dedupe + confidence-NMS)	`tiled`	`detections` (one per instance)
`region`	yes → free-space rect	no	`reason` (region)	`detections` `kind:"region"` (box=region, empty mask, point=anchor)

Plus watcher behavior (the watch pipeline): a skill with a boolean output + a cheap frame-difference gate (watch.gate_threshold) that escalates to ER only when the frame changes — publishes the boolean to its topic; the buffer turns the rising edge into a watch_event (§1). Postprocessing (hole-fill, speck removal, oversized-background filter) and the prewarm cache (prewarm_ttl_s) round out the pipeline behavior.

6. ROS interface (the data-plane contract)¶

6.1 Topics¶

Direction	Topic	Type
sub	`camera_topic` (default `/camera/image_raw`)	`sensor_msgs/Image` (BEST_EFFORT, depth 1)
pub	`/vision/{skill}/detections`	`ros2_vision_interfaces/VisionDetectionArray` with masks — live consumers (manipulation, buffer) subscribe here
pub	`/vision/{skill}/detections_meta`	same `VisionDetectionArray` with the `mask` field empty (labels/boxes/scores/points only) — the recordable topic for `ros2-memory`; ~100× smaller. Always published alongside `/detections`.
pub	`/vision/{skill}/{output}`	`ros2_vision_interfaces/StampedString` (per semantic output)
pub	`/vision/{skill}/overlay`	`sensor_msgs/Image` (annotated; debug/UI)
pub	`/vision/stats`, `/vision/latency`	diagnostics (from `ros2-vlm-groq`)

Masks live on the data plane, not in history

/detections carries pixel masks for live consumers (a grasp uses the current mask). For episodic recording (ros2-memory) and any logging consumer, subscribe to /detections_meta (mask-free) — old masks are useless once the scene changed, and recording them blows up bag size. Publishing meta is trivial (clear one field), so it is always on; cost is negligible.

6.2 `ros2_vision_interfaces` (fresh package)¶

# VisionDetection.msg
string label            # "scissors"
float32 score
string kind             # "object" | "region"
sensor_msgs/RegionOfInterest box     # pixel bbox
sensor_msgs/Image           mask     # mono8 full-frame (0/255); empty for region-only
geometry_msgs/Point         point    # grasp/centroid px (z=0); region anchor for regions
string extra            # per-skill JSON extras

# VisionDetectionArray.msg
std_msgs/Header     header           # stamp + frame_id preserved from the source image
string              skill
VisionDetection[]   detections

# StampedString.msg   (carried over from ros2-vlm-groq)
std_msgs/Header header
string         data

# VisionTask.srv      (configure surface; one generic JSON-in/out service, 3 names)
string data            # JSON request
---
bool   success
string message         # JSON payload for list

mono8 full-frame masks at ≤2 Hz are ~0.3 MB at 640×480 — fine on-robot. Future: switch to RLE behind the same message if the manipulation hand-off needs it.

6.3 Services (configure; called by `mcp-skill-control`)¶

~/create_vision_task (data = a skill, §3) · ~/manage_vision_task ({task_name, action: enable|disable|delete}) · ~/list_vision_tasks ({} → JSON of skills + enabled + topics). Same shape as ros2-vlm-groq's three services; data gains mode/grounding/segment.

7. Config & launch¶

config/config.yaml selects grounder/segmenter/frame_source and carries pipeline/watch knobs. Node params: camera_topic, topic_namespace (/vision), processing_frequency, max_in_flight, default_skills (created at startup, e.g. a describe "scene" skill). Env: GEMINI_API_KEY (required for ER), PERCEPTION_CONFIG_OVERRIDES (Hydra overrides, e.g. frame_source=folder segmenter=none). Launch: ros2 launch robot_mcp_perception vision.launch.py. The optional artifact HTTP endpoint (artifacts.py, port 9112) is available for UI/overlay fetch, but masks travel over ROS, not HTTP.

8. Cost-control policy (meter every cloud call)¶

Why this section exists. Two facts make naive "run the VLM at 2 Hz" wrong:

"What do you see" is not free to produce. Reading the perception-buffer is free, but the buffer only holds a scene description if a describe-mode skill ran a model to make one. So answering "what do you see" cheaply requires some producing skill to be warm; otherwise it needs a fresh model call.
Any stage may be a cloud call. Today the grounder (Gemini ER) is cloud and the segmenter (SAM 3) is local — but segmentation may move to the cloud too. So the policy must meter whichever stages cost money/latency, not assume "grounding = the only cost."

The policy therefore meters and gates every cloud stage, driven by Hydra config — no model free-runs unless a deployment explicitly opts in.

Each model adapter declares its cost class (in its Hydra YAML):

# config/grounder/gemini_er.yaml   → cost_class: cloud
# config/segmenter/sam.yaml        → cost_class: local
# config/segmenter/sam_cloud.yaml  → cost_class: cloud   (future)
cost_class: cloud   # local | cloud   — local stages are never metered/gated

config/config.yaml gains a cost: block:

cost:
  max_cloud_calls_per_min: 60       # global budget across all skills; over → skip + WARN
  max_cloud_rate_hz: 1.0            # ceiling on a single skill's cloud-stage rate
  cloud_requires_gate: true          # a continuous (non-once) skill may only fire a cloud stage
                                      #   when the watch frame-diff gate passes (scene changed)
  ambient_scene:                     # optional: keep "what do you see" warm cheaply
    enabled: false
    rate_hz: 0.2                      # behind the gate, so a static scene costs ~nothing
    grounding: "a brief description of the scene and the main objects"

Rules the node enforces: - A skill's local stages run at the skill's rate_hz unmetered. - A skill's cloud stages run only if: the skill is once, or the watch gate passed (when cloud_requires_gate), and the skill's effective rate ≤ max_cloud_rate_hz, and the global max_cloud_calls_per_min budget has headroom. Over budget → skip this tick, WARN cloud_call_skipped, keep publishing the last result (stamped stale). - ambient_scene (off by default): a single gated, low-rate describe skill the node runs so what_do_you_see() is usually warm for a few cents/hour. The frame-diff gate means a still scene triggers almost no calls. Turn it on for "always answerable"; leave it off for "answer on demand". - For closed-loop manipulation, prefer skills whose continuous stage is local (segment/ track) and whose cloud stage (grounding "which object") runs once up front — ground once, then segment locally at rate. This keeps a grasp loop token-free even at high rate.

"What do you see" then has three honest modes (the brain picks per the buffer's stale flag, specs/mcp-perception-buffer.md §4): warm buffer → free read; cold buffer → instantiate a once describe skill (one cloud call); historical → ask ros2-memory.

Observability: log every model call as model_call {stage, skill, cost_class, duration_ms} and every skip as cloud_call_skipped {reason} so the admin metrics can show calls/min and spend.

9. Tests¶

CPU-only CI (mock Grounder, NullSegmenter, frame_source: folder): 1. skills.py (de)serialize + validate (mode/types/duplicate name). 2. router/pipelines: ground_and_segment crop→box-prompt + fallbacks + plausibility; region passthrough; segment_all tile→full mapping, dedupe + NMS, SAM once on full image; describe semantic-only. 3. services: create_vision_task registers a skill + creates /detections + per-output publishers; manage enable/disable/delete; list reports topics. 4. publishing: header stamp + frame_id preserved; mask is mono8 full-frame; overlay published; once:true auto-disables after one publish. 5. watcher gate: identical folder frames suppress ER calls; a changed frame escalates. 6. prewarm: cache hit within TTL → one grounder call. 7. cost policy (§8): a continuous skill with a cloud stage skips ticks when the gate doesn't pass and when over max_cloud_calls_per_min (assert cloud_call_skipped); a once skill always fires once; local stages are never throttled; ambient_scene enabled keeps the buffer warm with calls only on frame change.

Integration (not CI): real GEMINI_API_KEY + SAM 3, frame_source: folder over playground/ images — a find_object("biggest package") one-shot skill yields a plausible mask on /vision/find_object/detections.

10. Acceptance checklist¶

colcon build && colcon test green; ruff/mypy clean (Appendix A gates).
ros2 run robot_mcp_perception vision_node (frame_source: folder) publishes /vision/scene/*.
create_vision_task for find_object(scissors on top) makes /vision/find_object/detections carry a mask + grasp point; segment_all and region skills work end-to-end on a topic.
A trivial subscriber (manipulation stand-in) receives the VisionDetectionArray with a usable mask — no brain involved.
manage_vision_task delete removes the skill and its publishers.

Appendix A — Repository & tooling conventions (apply to this repo)¶

Standard "how we build repos here". For this repo the package is an ament_python ROS 2 node (not a pure MCP server), so the MCP-server specifics below apply to the bridge servers that front it; everything else (layout, tooling, ports-and-adapters, Hydra, secrets, logging) applies directly.

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 deps (--extra sam). 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 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 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 (for the bridge servers fronting this node) — 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 ~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 logs.