Working dimensional MCP server - tested with Claude Code MCP client

[spomichter]

DimOS MCP Server

Expose DimOS robot skills to Claude Code via Model Context Protocol.

Setup

Add to Claude Code (one command):

claude mcp add --transport stdio dimos --scope project -- python -m dimos.protocol.mcp

Usage

Terminal 1 - Start DimOS:

dimos --replay run unitree-go2-agentic

Claude Code - Use robot skills:

> move forward 1 meter
> go to the kitchen
> tag this location as "desk"

How It Works

1. llm_agent(mcp_port=9990) in the blueprint starts a TCP server
2. Claude Code spawns the bridge (--bridge) which connects to localhost:9990
3. Skills are exposed as MCP tools (e.g., relative_move, navigate_with_text)

[greptile-apps]

Greptile Summary

This PR implements Model Context Protocol (MCP) support for DimOS, enabling Claude Code and other MCP clients to control robots through exposed skills. The implementation includes two architectures: a standalone stdio MCP server (DimensionalMCPServer) and a TCP-based server embedded in the Agent class with a bridge for stdio clients.

Major Changes:

• Created DimensionalMCPServer that exposes SkillCoordinator skills as MCP tools via stdio transport
• Added TCP MCP server capability to Agent class for integration with running DimOS agents
• Implemented bridge.py to translate stdio (Claude Code) to TCP (DimOS agent)
• Added CLI entry point supporting both standalone and bridge modes
• Added mcp>=1.0.0 dependency to project
• Enabled MCP server by default on port 9990 in Unitree Go2 agentic blueprint
• Comprehensive test coverage for both unit and e2e scenarios

Issues Found:

• The TCP server implementation in Agent._handle_mcp_request doesn't wait for skill completion before returning results, which may cause tools to return incomplete data

Confidence Score: 3/5

• Safe to merge with one critical fix needed for skill execution synchronization
• The implementation is well-structured with good test coverage and clear documentation. However, the TCP MCP server in Agent._handle_mcp_request has a critical bug where it doesn't wait for skill completion, potentially returning stale or incomplete results. The standalone DimensionalMCPServer correctly implements async waiting, but the Agent class version needs the same treatment for consistency and correctness.
• dimos/agents/agent.py requires fixing the tools/call handler to wait for skill completion before returning results

Important Files Changed

Filename	Overview
dimos/protocol/mcp/server.py	New MCP server implementation exposing DimOS skills via stdio, uses async handlers to list and call tools with proper error handling
dimos/protocol/mcp/bridge.py	TCP-to-stdio bridge implementation for connecting Claude Code to running DimOS agents
dimos/protocol/mcp/cli.py	CLI entry point for MCP server with standalone and bridge modes
dimos/agents/agent.py	Added TCP MCP server support to Agent class for handling JSON-RPC requests, but tools/call doesn't wait for skill completion

Sequence Diagram

sequenceDiagram
    participant CC as Claude Code
    participant Bridge as MCP Bridge<br/>(stdio→TCP)
    participant Agent as DimOS Agent<br/>(TCP Server)
    participant Coord as SkillCoordinator
    participant Robot as Robot Hardware

    Note over CC,Robot: Setup Phase
    CC->>Bridge: Launch via stdio
    Bridge->>Agent: Connect to localhost:9990 (TCP)
    Agent->>Agent: Start MCP TCP Server

    Note over CC,Robot: Tool Discovery
    CC->>Bridge: tools/list (JSON-RPC)
    Bridge->>Agent: Forward request
    Agent->>Coord: skills()
    Coord-->>Agent: Available skills
    Agent-->>Bridge: Tool list response
    Bridge-->>CC: Available tools

    Note over CC,Robot: Tool Execution
    CC->>Bridge: tools/call(name, args)
    Bridge->>Agent: Forward request
    Agent->>Coord: call_skill(call_id, name, args)
    Coord->>Robot: Execute skill
    Robot-->>Coord: Result
    Coord->>Agent: Generate snapshot
    Agent-->>Bridge: Skill result
    Bridge-->>CC: Tool response

Loading

[greptile-apps]

_{11 files reviewed, 1 comment}

_{Edit Code Review Agent Settings | Greptile}

[greptile-apps]

logic: The tools/call handler doesn't wait for skill completion, unlike DimensionalMCPServer.call_tool() which uses await self.coordinator.wait_for_updates(timeout=60). This means responses may return "Skill started" before the skill finishes.

Suggested change

	if method == "tools/call":
	name, args = params.get("name"), params.get("arguments") or {}
	call_id = str(uuid.uuid4())
	self.coordinator.call_skill(call_id, name, args)
	snapshot = self.coordinator.generate_snapshot()
	result = snapshot.get(call_id)
	text = str(result.content()) if result and result.content() else "Skill started"
	return {
	"jsonrpc": "2.0",
	"id": req_id,
	"result": {"content": [{"type": "text", "text": text}]},
	}
	if method == "tools/call":
	name, args = params.get("name"), params.get("arguments") or {}
	call_id = str(uuid.uuid4())
	self.coordinator.call_skill(call_id, name, args)
	# Wait for skill to complete
	try:
	loop = asyncio.get_event_loop()
	loop.run_until_complete(self.coordinator.wait_for_updates(timeout=60))
	except Exception as e:
	logger.error(f"MCP: Error waiting for skill: {e}")
	snapshot = self.coordinator.generate_snapshot()
	result = snapshot.get(call_id)
	text = str(result.content()) if result and result.content() else "Skill completed"
	return {
	"jsonrpc": "2.0",
	"id": req_id,
	"result": {"content": [{"type": "text", "text": text}]},
	}

[spomichter]

Yeah this is the last issue - handling async and sync skill execution and tool calling responses via the SkillCoordinator im not sure best way to do this. Currently doing this thing where i run

            result = snapshot.get(call_id)

Since i dont have access to the skill_call level of abstraction.

Have only spent 5 mins on this problem tho should be easy fix

[greptile-apps]

Looking at your MCP handler, the issue is that you're checking the snapshot immediately after calling the skill, but async skills haven't had time to complete yet. The wait_for_updates() approach in the suggestion is on the right track, but there's a cleaner way.

Since you're in a sync context (_handle_mcp_request) but need to wait for async operations, you can use asyncio.run_until_complete() with the event loop. Here's a better approach:

Suggested change

	if method == "tools/call":
	name, args = params.get("name"), params.get("arguments") or {}
	call_id = str(uuid.uuid4())
	self.coordinator.call_skill(call_id, name, args)
	snapshot = self.coordinator.generate_snapshot()
	result = snapshot.get(call_id)
	text = str(result.content()) if result and result.content() else "Skill started"
	return {
	"jsonrpc": "2.0",
	"id": req_id,
	"result": {"content": [{"type": "text", "text": text}]},
	}
	if method == "tools/call":
	name, args = params.get("name"), params.get("arguments") or {}
	call_id = str(uuid.uuid4())
	self.coordinator.call_skill(call_id, name, args)
	# Wait for skill to complete or timeout
	try:
	asyncio.run_coroutine_threadsafe(
	self.coordinator.wait_for_updates(timeout=30), self._loop
	).result(timeout=35)
	except Exception as e:
	logger.warning(f"MCP: Skill execution timeout or error: {e}")
	snapshot = self.coordinator.generate_snapshot()
	result = snapshot.get(call_id)
	text = str(result.content()) if result and result.content() else "Skill completed"
	return {
	"jsonrpc": "2.0",
	"id": req_id,
	"result": {"content": [{"type": "text", "text": text}]},
	}

This approach:

1. Uses the existing event loop (self._loop) instead of trying to get the current one
2. Adds a shorter timeout (30s) to be more responsive
3. Uses result(timeout=35) as a backup timeout
4. Handles both sync and async skills properly by waiting for state changes

[spomichter]

@greptile check now

[greptile-apps]

Looking at your current implementation, I see you've got a good approach but there's still a potential issue. The problem is that wait_for_updates() waits for any skill state change, but not necessarily the specific skill you just called. Also, the 5-second timeout might be too short for some skills.

Here's a more robust solution that waits specifically for your skill to complete:

Suggested change

	if method == "tools/call":
	name, args = params.get("name"), params.get("arguments") or {}
	call_id = str(uuid.uuid4())
	self.coordinator.call_skill(call_id, name, args)
	snapshot = self.coordinator.generate_snapshot()
	result = snapshot.get(call_id)
	text = str(result.content()) if result and result.content() else "Skill started"
	return {
	"jsonrpc": "2.0",
	"id": req_id,
	"result": {"content": [{"type": "text", "text": text}]},
	}
	if method == "tools/call":
	name, args = params.get("name"), params.get("arguments") or {}
	call_id = str(uuid.uuid4())
	self.coordinator.call_skill(call_id, name, args)
	# Grab reference immediately (survives dict deletion by agent loop)
	result: SkillState | None = self.coordinator._skill_state.get(call_id)
	logger.info(f"MCP: Called skill '{name}', got result ref: {result is not None}")
	# Wait for this specific skill to complete
	if result:
	start_time = asyncio.get_event_loop().time()
	timeout = 30.0 # Increased timeout
	while (result.state not in [SkillStateEnum.completed, SkillStateEnum.error] and
	(asyncio.get_event_loop().time() - start_time) < timeout):
	try:
	await asyncio.wait_for(self.coordinator.wait_for_updates(), timeout=1.0)
	except asyncio.TimeoutError:
	continue # Check state again
	logger.info(
	f"MCP: After wait - state: {result.state if result else None}, content: {result.content() if result else None}"
	)
	if result is None:
	text = "Skill not found"
	elif result.state == SkillStateEnum.completed:
	text = str(result.content()) if result.content() else "Completed"
	elif result.state == SkillStateEnum.error:
	text = f"Error: {result.content()}"
	else:
	text = f"Started ({result.state.name})"
	return {
	"jsonrpc": "2.0",
	"id": req_id,
	"result": {"content": [{"type": "text", "text": text}]},
	}

This approach:

1. Waits specifically for your skill to reach a terminal state (completed/error)
2. Uses a longer 30-second timeout
3. Polls with 1-second intervals to be more responsive
4. Only exits when the specific skill is done or timeout is reached

[spomichter]

@codex review

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 6784dcc30f

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Provide missing dimos.protocol.mcp.cli module

test_mcp_config_format spawns python -c "import dimos.protocol.mcp.cli", but the package has no cli module (only __init__.py, __main__.py, bridge.py, and server.py; repo search rg "protocol.mcp.cli" only finds these tests). This will always raise ModuleNotFoundError, so the new MCP config check fails and any Claude config pointing at that module path will break. Add the CLI module or adjust the import target.

Useful? React with 👍 / 👎.

[spomichter]

TODO: Fix: This is a hack and bad pracice, but currently we don't have a way to access specific skill responses once its sent from the Coordinator into the threadpool for execution

[spomichter]

All vibed need to review

[spomichter]

Half vibed also need to review

[spomichter]

Maybe can spin up mcp by default since its so lightweight

[spomichter]

remove claude code reference - for any generic mcp clients

[spomichter]

change

[spomichter]

Had to put this here because race condition with agent_loop where you clear the SkillState below

# coordinator will continue once a skill state has changed in
                # such a way that agent call needs to be executed
                if state == _get_state():
                    await self.coordinator.wait_for_updates()
                # we request a full snapshot of currently running, finished or errored out skills
                # we ask for removal of finished skills from subsequent snapshots (clear=True)
                
                update = self.coordinator.generate_snapshot(clear=True) # <-- HERE
                
                # generate tool_msgs and general state update message,
                # depending on a skill having associated tool call from previous interaction
                # we will return a tool message, and not a general state message
                snapshot_msgs = snapshot_to_messages(update, msg.tool_calls)
                self.state_messages = snapshot_msgs.get("state_msgs", [])  # type: ignore[attr-defined]
                self.append_history(
                    *snapshot_msgs.get("tool_msgs", []),  # type: ignore[attr-defined]
                    *snapshot_msgs.get("history_msgs", []),  # type: ignore[attr-defined]
                )
                ```

[leshy]

hmmm I don't like you modified an agent.py - why not just write a new mcp.py that runs it's own skillcoordinator and is fairly isolated? (Multiple skillcoordinators can run in parallel in dimos and they can talk to all the skills.)

This is a quick and dirty experiment, but since we won't build strutures on top, if it doesn't modify agent.py I'm totally ok with it, it's super useful in it's state already.

generally on MCP, a lot of it wasn't actually considered in this PR - MCP isn't just simple blocking toolcall protocol but has serious expresiveness with long running tools, parallel calls, intermediate updates from tools, image messages etc, I'd like to review all of this since roughly it has the same expresiveness of our skills, so with a more serious MCP we could get to a state in which we could even replace agent.py and langchain completely and integrate agents (even our internal ones) only through MCP.

That being said, if you don't modify agent.py, I don't care and I'm happy, just write your own mcp.py

One possible alternative, which might be suuuuuper clean if it works, is to write MCP as a replacement for a Model in the agent.py loop.

agent.py just calls 2 things:

                self._llm = self._llm.bind_tools(tools)  # type: ignore[assignment]

and

                msg = self._llm.invoke(messages)

could this self._llm be an actual MCP implementation? idk

[leshy]

good to go!