GS Robot MCP Server

Instructions

Automated process: Status query → Task selection → Command execution

Args:
    serial_number: The serial number of the target robot.
    task_selection_criteria: Optional task selection criteria.

Returns:
    A dictionary containing the workflow execution result.

Implementation Reference

• src/gs_openapi/main.py:238-257 (registration)

  MCP tool registration with @mcp.tool() decorator. Thin wrapper that delegates to GausiumMCP instance method.

  @mcp.tool()
  async def execute_m_line_task_workflow(
      serial_number: str,
      task_selection_criteria: Optional[dict] = None
  ):
      """Executes complete M-line robot task workflow.
      
      Automated process: Status query → Task selection → Command execution
  
      Args:
          serial_number: The serial number of the target robot.
          task_selection_criteria: Optional task selection criteria.
  
      Returns:
          A dictionary containing the workflow execution result.
      """
      return await mcp.execute_m_line_task_workflow(
          serial_number=serial_number,
          task_selection_criteria=task_selection_criteria
      )

• src/gs_openapi/workflows/task_engine.py:34-95 (handler)

  Core handler logic: fetches robot status, extracts and selects available tasks, issues START_TASK command via API client.

  async def execute_m_line_task(
      self,
      serial_number: str,
      task_selection_criteria: Optional[Dict[str, Any]] = None
  ) -> Dict[str, Any]:
      """
      执行M线机器人任务。
      
      工作流：
      1. 获取机器人状态
      2. 从状态中提取可执行任务列表
      3. 根据条件选择任务
      4. 通过Create Robot Command下发任务
      
      Args:
          serial_number: 机器人序列号
          task_selection_criteria: 任务选择条件
          
      Returns:
          任务执行结果
      """
      logger.info(f"Starting M-line task execution for robot: {serial_number}")
      
      async with GausiumAPIClient() as client:
          try:
              # 1. 获取机器人状态
              status = await client.call_endpoint(
                  'get_robot_status_v1',
                  path_params={'serial_number': serial_number}
              )
              
              # 2. 解析可执行任务列表
              available_tasks = self._extract_m_line_tasks(status)
              if not available_tasks:
                  raise ValueError("No executable tasks found in robot status")
              
              # 3. 选择任务
              selected_task = self._select_m_line_task(
                  available_tasks, 
                  task_selection_criteria or {}
              )
              
              # 4. 构建并下发任务指令
              command_result = await client.call_endpoint(
                  'create_command',
                  path_params={'serial_number': serial_number},
                  json_data={
                      "serialNumber": serial_number,
                      "remoteTaskCommandType": "START_TASK",
                      "commandParameter": {
                          "startTaskParameter": selected_task
                      }
                  }
              )
              
              logger.info(f"M-line task executed successfully: {command_result}")
              return command_result
              
          except Exception as e:
              logger.error(f"M-line task execution failed: {str(e)}")
              raise

• src/gs_openapi/mcp/gausium_mcp.py:665-685 (helper)

  GausiumMCP class method that delegates workflow execution to the task_engine instance.

  async def execute_m_line_task_workflow(
      self,
      serial_number: str,
      task_selection_criteria: Optional[Dict[str, Any]] = None
  ) -> Dict[str, Any]:
      """
      执行M线机器人完整任务工作流。
      
      自动化流程：状态查询 → 任务选择 → 指令下发
      
      Args:
          serial_number: 机器人序列号
          task_selection_criteria: 任务选择条件
          
      Returns:
          工作流执行结果
      """
      return await self.task_engine.execute_m_line_task(
          serial_number=serial_number,
          task_selection_criteria=task_selection_criteria
      )