Skip to content
AuthorPascal DateJuly 14, 2026 Rev1.0

mcp-skill-control — Implementation Spec

mcp-skill-control is the control-plane MCP server through which the brain discovers what the robot can be configured to do and instantiates those capabilities ("skills") on the ROS stack. It is a ROS node that also serves MCP: the brain speaks MCP to it; it speaks ROS services to the robot-mcp-perception node. The brain never imports rclpy; this server is the boundary.

See AGENT_ORCHESTRATION_CONCEPT.md §2 (the skill model) and §4.1, and specs/robot-mcp-perception.md (the node it configures).

1. Context

  • A skill is a declarative capability config that, when instantiated, configures a ROS node to start publishing a stream on a ROS topic (concept §2). All skills are vision skills — configurations of the robot-mcp-perception node (specs/robot-mcp-perception.md §3).
  • This server owns the skill template catalog (parameterized skills) and turns a template + params into a running skill by calling robot-mcp-perception's create_vision_task service, then returns the ROS topic the result will appear on.
  • It does not read results — that's the separate mcp-perception-buffer server. Clean split: this is write/config (ROS services); the buffer is read (ROS topics).

The server bridges MCP to ROS via an MCP Server, a hidden rclpy node with service clients to create_vision_task/manage_vision_task/list_vision_tasks, and a ThreadPoolExecutor spinning ROS. Names are generalized to "skill", discovery/instantiation tools are explicit, and the catalog is data-driven (YAML) rather than a hardcoded dict.

2. Repo layout

mcp-skill-control/                   (ament_python OR plain pip pkg; needs rclpy at runtime)
├── package.xml / setup.py
├── config/
│   └── skills.yaml                  # the skill template catalog (data, not code)
├── src/mcp_skill_control/
│   ├── server.py                    # MCP Server + rclpy node + service clients
│   ├── catalog.py                   # load/validate skill templates; param filling
│   └── ros_bridge.py                # thin async wrapper over the VisionTask service clients
└── tests/

Dependencies: mcp, rclpy, ros2_vision_interfaces (the VisionTask srv), pyyaml, robot-mcp-kit (only if this server posts events; see §5). Transport: stdio by default (the brain spawns it) or sse (MCP_TRANSPORT=sse, MCP_PORT default 9201), matching the chat app's tools/mcp.yaml conventions. FastMCP/mcp server name skill_control → dotted tools skill_control.list_skill_templates, etc.

3. The skill template catalog (config/skills.yaml)

The catalog is data-driven. Each template is a robot-mcp-perception skill (specs/robot-mcp-perception.md §3) with {param} placeholders and a declared param list:

templates:
  - name: find_object
    description: "Locate and segment a specific object the user names."
    params: [query]
    skill:
      mode: ground_and_segment
      grounding: "the {query}"
      segment: true
      outputs:
        - {name: found, type: boolean, description: "whether the {query} is visible"}

  - name: segment_all
    description: "Segment every object in view."
    params: []
    skill: {mode: segment_all, segment: true}

  - name: free_space
    description: "Find a placement region for something."
    params: [what]
    skill:
      mode: region
      grounding: "free flat space large enough for the {what}"

  - name: describe_scene
    description: "Describe what the camera sees (no segmentation)."
    params: []
    skill:
      mode: describe
      outputs:
        - {name: scene_description, type: text, description: "detailed description of the scene"}
        - {name: scene_objects,     type: list, description: "main objects, up to 10"}

  - name: watch_for
    description: "Watch until a condition becomes true, then notify."
    params: [condition]
    skill: {mode: describe, once: false, outputs: [
            {name: triggered, type: boolean, description: "whether: {condition}"}]}
    watch: {output: triggered, equals: true}   # turns this skill into a push watcher (§5)

catalog.py: load(), list(), fill(template_name, params) -> skill_dict (substitutes {param}, validates all params present, validates the resulting skill against the robot-mcp-perception schema).

Future

Learned-good skills may be merged in from robot-mcp-memory (concept §2); the catalog interface stays the same.

4. MCP tools

All return JSON strings; docstrings written for the planner LLM (prescriptive — "Call this when…").

Tool Args Returns / effect
list_skill_templates() — {templates:[{name, description, params}]} — what the robot can be configured to do (discovery).
list_running_skills() — {skills:[{skill_id, name, topic, enabled}]} — calls list_vision_tasks; what's currently publishing.
instantiate_skill(template, params) template name + param dict fills the template (catalog.fill), calls create_vision_task, returns {skill_id, name, topic:"/vision/{name}/detections", semantic_topics:[…]}. This is "dynamically make use of a skill."
instantiate_raw_skill(skill) a full skill object (§3 of vision spec) escape hatch for an ad-hoc skill not in the catalog; same create path.
stop_skill(skill_id) — calls manage_vision_task {delete}; tears down the skill + its publishers.
set_skill_enabled(skill_id, enabled) — manage_vision_task {enable\|disable} — pause without losing the config.

skill_id is the skill's unique name (the vision node keys tasks by name); the server may suffix a counter to avoid collisions and track the mapping. Each tool wraps the corresponding VisionTask service call (ros_bridge.py) with the sample's thread-pool + timeout pattern, and returns the topic name so the brain can hand it to the perception-buffer (get_latest) or a ROS consumer — the brain never touches pixels.

5. Watchers → push events (optional, via robot-mcp-kit)

A template with a watch: block (e.g. watch_for) is a watcher: once instantiated, it notifies the brain when its condition flips, rather than the brain polling. There are two possible ownership models:

  • A. The perception-buffer owns watch→push. This server just instantiates the skill; the mcp-perception-buffer (which is already subscribed to the topic) evaluates the watch predicate and POSTs a watch_event to the brain via robot-mcp-kit. This keeps this server write-only and all topic-reading in one place.
  • B. This server subscribes too. It keeps its own subscription for watched skills and posts the event itself. Simpler ownership, but now both bridge servers read topics.

Either way the push uses the brain's existing Async Task Contract (POST /api/agent/events, HMAC-signed, robot-mcp-kit), so the brain consumes it identically to any task event.

This service uses option A: mcp-perception-buffer (specs/mcp-perception-buffer.md §5) carries the watch logic, and this server only records the watch intent in the skill it creates.

6. Config & wiring

Env Meaning Default
MCP_TRANSPORT / MCP_PORT stdio | sse / sse port stdio / 9201
VISION_NODE_NAME the robot-mcp-perception node name to target services on vision_node
SKILLS_CATALOG path to skills.yaml config/skills.yaml
ROS_DOMAIN_ID ROS domain inherit

Wire into the brain's backend/config/tools/mcp.yaml like any MCP server:

servers:
  - name: skill_control
    transport: sse
    url: "http://127.0.0.1:9201/sse"

No event_secret here unless option B is chosen (only event-posting servers need it).

7. Tests

  1. catalog: load skills.yaml; fill("find_object", {query:"scissors"}) substitutes and validates; missing param → error; unknown template → error.
  2. tools (mock the VLMinterface service clients): list_skill_templates returns the catalog; instantiate_skill calls create_vision_task with the filled skill JSON and returns the right /vision/{name}/detections topic; stop_skill calls manage_vision_task {delete}; list_running_skills parses list_vision_tasks.
  3. bridge: service-unavailable / timeout surfaces a clean MCP error, not a crash (mirror the sample's _call_ros_service handling).

(Run against a fake rclpy service node in CI; a real robot-mcp-perception for the integration check.)

8. Acceptance checklist

  • Server starts, connects to a running robot-mcp-perception, list_skill_templates returns the YAML catalog.
  • instantiate_skill("find_object", {query:"scissors on top"}) makes /vision/find_object/detections appear (verified with ros2 topic list) and returns that topic; stop_skill removes it.
  • Brain (chat app) lists + instantiates a skill end-to-end via MCP, receiving only the topic name (no pixels cross MCP).

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.