execute_s_line_site_task_workflow

Automates S-line robot task workflows by processing site information to select maps, retrieve subareas, build tasks, and submit them for execution.

Instructions

Executes complete S-line robot task workflow with site information.

Automated process: Site info → Map selection → Subarea retrieval → Task building → Task submission

Args:
    robot_id: The ID of the target robot.
    task_parameters: Task parameters including map criteria and task settings.

Returns:
    A dictionary containing the workflow execution result.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`robot_id`	Yes
`task_parameters`	Yes

Implementation Reference

src/gs_openapi/main.py:259-278 (handler)

Primary MCP tool handler and registration for 'execute_s_line_site_task_workflow'. Delegates to GausiumMCP instance.

@mcp.tool()
async def execute_s_line_site_task_workflow(
    robot_id: str,
    task_parameters: dict
):
    """Executes complete S-line robot task workflow with site information.
    
    Automated process: Site info → Map selection → Subarea retrieval → Task building → Task submission

    Args:
        robot_id: The ID of the target robot.
        task_parameters: Task parameters including map criteria and task settings.

    Returns:
        A dictionary containing the workflow execution result.
    """
    return await mcp.execute_s_line_site_task_workflow(
        robot_id=robot_id,
        task_parameters=task_parameters
    )

src/gs_openapi/mcp/gausium_mcp.py:687-707 (handler)

GausiumMCP class method implementing the workflow by delegating to TaskExecutionEngine.

async def execute_s_line_site_task_workflow(
    self,
    robot_id: str,
    task_parameters: Dict[str, Any]
) -> Dict[str, Any]:
    """
    执行S线有站点任务完整工作流。
    
    自动化流程：站点信息 → 地图选择 → 分区获取 → 任务构建 → 任务下发
    
    Args:
        robot_id: 机器人ID
        task_parameters: 任务参数
        
    Returns:
        工作流执行结果
    """
    return await self.task_engine.execute_s_line_site_task(
        robot_id=robot_id,
        task_parameters=task_parameters
    )

src/gs_openapi/workflows/task_engine.py:96-162 (handler)

Core implementation of the S-line site task workflow in TaskExecutionEngine, handling site info retrieval, map selection, subarea fetching, task building, and submission.

async def execute_s_line_site_task(
    self,
    robot_id: str,
    task_parameters: Dict[str, Any]
) -> Dict[str, Any]:
    """
    执行S线有站点任务。
    
    工作流：
    1. 获取站点信息
    2. 解析可用地图
    3. 获取目标地图分区
    4. 构建并下发有站点临时任务
    
    Args:
        robot_id: 机器人ID
        task_parameters: 任务参数
        
    Returns:
        任务执行结果
    """
    logger.info(f"Starting S-line site task execution for robot: {robot_id}")
    
    async with GausiumAPIClient() as client:
        try:
            # 1. 获取站点信息
            site_info = await client.call_endpoint(
                'get_site_info',
                path_params={'robot_id': robot_id}
            )
            
            # 2. 解析可用地图
            available_maps = self._extract_maps_from_site(site_info)
            if not available_maps:
                raise ValueError("No maps found in site information")
            
            # 3. 选择目标地图
            target_map_id = self._select_map(
                available_maps, 
                task_parameters.get('map_criteria', {})
            )
            
            # 4. 获取地图分区
            subareas = await client.call_endpoint(
                'get_map_subareas',
                path_params={'map_id': target_map_id}
            )
            
            # 5. 构建任务数据
            task_data = self._build_site_task_data(
                target_map_id, 
                subareas, 
                task_parameters
            )
            
            # 6. 下发有站点临时任务
            task_result = await client.call_endpoint(
                'submit_temp_site_task',
                json_data=task_data
            )
            
            logger.info(f"S-line site task executed successfully: {task_result}")
            return task_result
            
        except Exception as e:
            logger.error(f"S-line site task execution failed: {str(e)}")
            raise

GS Robot MCP Server