Robotics MCP Server

Instructions

Implementation Reference

• src/robotics_mcp/tools/vbot_crud.py:72-183 (handler)

  Main handler function for the vbot_crud tool. Dispatches CRUD operations (create, read, update, delete, list) for virtual robots, integrating with Unity/VRChat platforms via mounted MCP servers.
  @self.mcp.tool() async def vbot_crud( operation: Literal["create", "read", "update", "delete", "list"], robot_type: Optional[str] = None, robot_id: Optional[str] = None, platform: Literal["unity", "vrchat"] = "unity", position: Optional[Dict[str, float]] = None, scale: Optional[float] = None, metadata: Optional[Dict[str, Any]] = None, model_path: Optional[str] = None, ) -> Dict[str, Any]: """CRUD operations for virtual robots (vbots). This tool provides complete lifecycle management for virtual robots: - Create: Spawn and register a new virtual robot - Read: Get details of an existing virtual robot - Update: Modify virtual robot properties (scale, position, metadata, etc.) - Delete: Remove and unregister a virtual robot - List: List all virtual robots with optional filtering Supported robot types: - "scout": Moorebot Scout (mecanum wheels, indoor) - "scout_e": Moorebot Scout E (tracked, waterproof, outdoor) - "go2": Unitree Go2 (quadruped) - "g1": Unitree G1 (humanoid with arms) - "robbie": Robbie from Forbidden Planet (classic sci-fi robot) - "custom": Custom robot type (requires model_path) Args: operation: CRUD operation to perform: - "create": Create/spawn a new virtual robot - "read": Read/get details of an existing virtual robot - "update": Update properties of an existing virtual robot - "delete": Delete/remove a virtual robot - "list": List all virtual robots (optionally filtered) robot_type: Type of robot (required for "create", optional for "list"). Must be one of: "scout", "scout_e", "go2", "g1", "robbie", "custom". robot_id: Virtual robot identifier (required for "read", "update", "delete"). Auto-generated for "create" if not provided. platform: Target platform ("unity" or "vrchat"). Default: "unity". position: Spawn/update position (x, y, z) for "create" or "update". scale: Size multiplier for "create" or "update" (e.g., 1.0 = original size). metadata: Additional metadata dictionary for "create" or "update". model_path: Path to 3D model file (.glb, .fbx, .vrm) for "create" with "custom" robot_type. Returns: Dictionary containing operation result with robot details. Examples: Create a Scout vbot: result = await vbot_crud( operation="create", robot_type="scout", platform="unity", position={"x": 0.0, "y": 0.0, "z": 0.0}, scale=1.0 ) Create Robbie from Forbidden Planet: result = await vbot_crud( operation="create", robot_type="robbie", platform="unity", position={"x": 1.0, "y": 0.0, "z": 1.0}, scale=1.0 ) Read vbot details: result = await vbot_crud( operation="read", robot_id="vbot_scout_01" ) Update vbot scale and position: result = await vbot_crud( operation="update", robot_id="vbot_scout_01", scale=1.5, position={"x": 2.0, "y": 0.0, "z": 2.0} ) Delete a vbot: result = await vbot_crud( operation="delete", robot_id="vbot_scout_01" ) List all vbots: result = await vbot_crud(operation="list") List only Scout vbots: result = await vbot_crud( operation="list", robot_type="scout" ) """ try: if operation == "create": return await self._create_vbot(robot_type, robot_id, platform, position, scale, metadata, model_path) elif operation == "read": return await self._read_vbot(robot_id) elif operation == "update": return await self._update_vbot(robot_id, position, scale, metadata) elif operation == "delete": return await self._delete_vbot(robot_id) elif operation == "list": return await self._list_vbots(robot_type, platform) else: return format_error_response(f"Unknown operation: {operation}", error_type="validation_error") except Exception as e: return handle_tool_error("vbot_crud", e, action=operation, context={"robot_type": robot_type, "robot_id": robot_id})
• src/robotics_mcp/server.py:490-491 (registration)

  Registration of the VbotCrudTool instance with the FastMCP server, making the vbot_crud tool available.
  self.vbot_crud.register() # Virtual robot CRUD operations logger.debug("Registered vbot_crud tool")
• src/robotics_mcp/tools/vbot_crud.py:42-49 (schema)

  Schema definition listing supported robot types for validation in create/list operations.
  SUPPORTED_ROBOT_TYPES = [ "scout", # Moorebot Scout "scout_e", # Moorebot Scout E (tracked, waterproof) "go2", # Unitree Go2 "g1", # Unitree G1 "robbie", # Robbie from Forbidden Planet "custom", # Custom robot type ]
• src/robotics_mcp/tools/vbot_crud.py:184-261 (helper)

  Helper method implementing the 'create' operation: validates inputs, registers virtual robot, spawns in Unity/VRChat, handles errors.
  async def _create_vbot( self, robot_type: Optional[str], robot_id: Optional[str], platform: str, position: Optional[Dict[str, float]], scale: Optional[float], metadata: Optional[Dict[str, Any]], model_path: Optional[str], ) -> Dict[str, Any]: """Create/spawn a new virtual robot.""" if not robot_type: return format_error_response("robot_type is required for create operation", error_type="validation_error") if robot_type not in SUPPORTED_ROBOT_TYPES: return format_error_response( f"Unsupported robot_type: {robot_type}. Supported types: {', '.join(SUPPORTED_ROBOT_TYPES)}", error_type="validation_error", ) if robot_type == "custom" and not model_path: return format_error_response( "model_path is required for custom robot_type", error_type="validation_error" ) # Generate robot_id if not provided if not robot_id: robot_id = f"vbot_{robot_type}_{len(self.state_manager.list_robots(is_virtual=True)) + 1:02d}" # Check if robot_id already exists if self.state_manager.get_robot(robot_id): return format_error_response(f"Robot {robot_id} already exists", error_type="validation_error") # Default position if position is None: position = {"x": 0.0, "y": 0.0, "z": 0.0} # Default scale if scale is None: scale = 1.0 # Prepare metadata vbot_metadata = { "spawned": True, "platform": platform, "position": position, "scale": scale, "model_path": model_path, **(metadata or {}), } # Register robot in state manager try: robot = self.state_manager.register_robot(robot_id, robot_type, platform=platform, metadata=vbot_metadata) except ValueError as e: return format_error_response(str(e), error_type="validation_error") # Spawn in Unity/VRChat via mounted servers spawn_result = await self._spawn_in_platform(robot_id, robot_type, platform, position, scale, model_path) spawn_data = extract_result_data(spawn_result) if spawn_data.get("status") != "success": # Cleanup registration if spawn failed self.state_manager.unregister_robot(robot_id) return spawn_result return format_success_response( f"Virtual robot {robot_id} created successfully", data={ "robot_id": robot_id, "robot_type": robot_type, "platform": platform, "position": position, "scale": scale, "metadata": vbot_metadata, }, robot_id=robot_id, )
• mcpb/server/server.py:115-291 (registration)

  Alternative registration in mcpb/server variant, instantiates and registers VbotCrudTool.
  self.vbot_crud = VbotCrudTool(self.mcp, self.state_manager, self.mounted_servers) # Register all tools self._register_tools() # Setup HTTP routes after tools are registered if self.config.enable_http: self._setup_http_routes() logger.info("Robotics MCP server initialized", http_enabled=self.config.enable_http) def _setup_http_routes(self): """Set up FastAPI HTTP routes.""" router = APIRouter(prefix="/api/v1") @router.get("/health") async def health(): """Health check endpoint.""" return {"status": "healthy", "version": "0.1.0"} @router.get("/robots") async def list_robots(): """List all registered robots.""" robots = self.state_manager.list_robots() return {"robots": [r.to_dict() for r in robots]} @router.get("/robots/{robot_id}") async def get_robot(robot_id: str): """Get robot information.""" robot = self.state_manager.get_robot(robot_id) if not robot: raise HTTPException(status_code=404, detail=f"Robot {robot_id} not found") return robot.to_dict() @router.post("/robots/{robot_id}/control") async def control_robot(robot_id: str, request: Dict[str, Any] = None): """Control a robot via HTTP.""" if request is None: request = {} try: action = request.get("action", "get_status") params = {k: v for k, v in request.items() if k != "action"} # Use the robot_control tool result = await self.robot_control.handle_action(robot_id, action, params) return result except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get("/tools") async def list_tools(): """List all available MCP tools.""" tools = [] for tool_name, tool_info in self.mcp.list_tools().items(): tools.append( { "name": tool_name, "description": tool_info.get("description", ""), "inputSchema": tool_info.get("inputSchema", {}), } ) return {"tools": tools} @router.post("/tools/{tool_name}") async def call_tool(tool_name: str, params: Dict[str, Any] = None): """Call an MCP tool via HTTP.""" if params is None: params = {} try: # Execute tool using MCP instance # Note: FastMCP 2.13 tool calling interface result = await self.mcp.call_tool(tool_name, **params) return {"result": result} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.get("/status") async def get_status(): """Get server status.""" robots = self.state_manager.list_robots() return { "version": "0.1.0", "status": "healthy", "robots": [r.to_dict() for r in robots], "mounted_servers": list(self.mounted_servers.keys()), "http_enabled": self.config.enable_http, } @router.post("/robots") async def register_robot(request: Dict[str, Any]): """Register a new robot.""" try: robot_id = request.get("robot_id") robot_type = request.get("robot_type") platform = request.get("platform") metadata = request.get("metadata", {}) if not robot_id or not robot_type: raise HTTPException(status_code=400, detail="robot_id and robot_type required") robot = self.state_manager.register_robot( robot_id, robot_type, platform=platform, metadata=metadata ) return robot.to_dict() except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @router.delete("/robots/{robot_id}") async def unregister_robot(robot_id: str): """Unregister a robot.""" try: self.state_manager.unregister_robot(robot_id) return {"status": "success", "message": f"Robot {robot_id} unregistered"} except Exception as e: raise HTTPException(status_code=500, detail=str(e)) self.http_app.include_router(router) def _mount_mcp_servers(self): """Mount external MCP servers for composition.""" try: # Try to mount osc-mcp try: from oscmcp.mcp_server import server as osc_mcp_server self.mcp.mount(osc_mcp_server, prefix="osc", as_proxy=True) self.mounted_servers["osc"] = osc_mcp_server logger.info("Mounted osc-mcp server") except ImportError: logger.warning("osc-mcp not available, skipping mount") # Try to mount unity3d-mcp try: from unity3d_mcp.server import Unity3DMCP unity_server = Unity3DMCP() self.mcp.mount(unity_server.app, prefix="unity", as_proxy=True) self.mounted_servers["unity"] = unity_server logger.info("Mounted unity3d-mcp server") except ImportError: logger.warning("unity3d-mcp not available, skipping mount") # Try to mount vrchat-mcp try: from vrchat_mcp import VRChatMCP vrchat_server = VRChatMCP() self.mcp.mount(vrchat_server.mcp, prefix="vrchat", as_proxy=True) self.mounted_servers["vrchat"] = vrchat_server logger.info("Mounted vrchat-mcp server") except ImportError: logger.warning("vrchat-mcp not available, skipping mount") # Try to mount avatar-mcp try: from avatarmcp.server import AvatarMCPServer avatar_server = AvatarMCPServer() self.mcp.mount(avatar_server.mcp, prefix="avatar", as_proxy=True) self.mounted_servers["avatar"] = avatar_server logger.info("Mounted avatar-mcp server") except ImportError: logger.warning("avatar-mcp not available, skipping mount") except Exception as e: logger.error("Error mounting MCP servers", error=str(e)) def _register_tools(self): """Register all MCP tools.""" # Mount external MCP servers first self._mount_mcp_servers() # Register portmanteau tools self.robot_control.register() self.virtual_robotics.register() self.vbot_crud.register()