execute_s_line_no_site_task_workflow

Automates S-line robot task workflows by mapping, selecting subareas, building tasks, and submitting them using robot serial numbers and parameters.

Instructions

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

Automated process: Map list → Map selection → Subarea retrieval → Task building → Task submission

Args:
    robot_sn: The serial number 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_sn`	Yes
`task_parameters`	Yes

Implementation Reference

src/gs_openapi/main.py:280-299 (registration)

MCP tool registration using @mcp.tool() decorator. This is the entry point for the tool invocation, delegating to GausiumMCP instance.

@mcp.tool()
async def execute_s_line_no_site_task_workflow(
    robot_sn: str,
    task_parameters: dict
):
    """Executes complete S-line robot task workflow without site information.
    
    Automated process: Map list → Map selection → Subarea retrieval → Task building → Task submission

    Args:
        robot_sn: The serial number 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_no_site_task_workflow(
        robot_sn=robot_sn,
        task_parameters=task_parameters
    )

src/gs_openapi/mcp/gausium_mcp.py:709-729 (handler)

GausiumMCP class method that wraps and delegates the workflow execution to TaskExecutionEngine.

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

src/gs_openapi/workflows/task_engine.py:164-232 (handler)

Core handler implementing the S-line no-site task workflow: fetches robot maps, selects map, retrieves subareas, builds task data, and submits temporary no-site task via API.

async def execute_s_line_no_site_task(
    self,
    robot_sn: str,
    task_parameters: Dict[str, Any]
) -> Dict[str, Any]:
    """
    执行S线无站点任务。
    
    工作流：
    1. 获取机器人地图列表
    2. 获取目标地图分区
    3. 构建并下发无站点临时任务
    
    Args:
        robot_sn: 机器人序列号
        task_parameters: 任务参数
        
    Returns:
        任务执行结果
    """
    logger.info(f"Starting S-line no-site task execution for robot: {robot_sn}")
    
    async with GausiumAPIClient() as client:
        try:
            # 1. 获取机器人地图列表
            maps_response = await client.call_endpoint(
                'list_maps',
                json_data={'robotSn': robot_sn}
            )
            
            # 处理Gausium特殊响应格式
            if maps_response.get('code') != 0:
                raise RuntimeError(f"Failed to get maps: {maps_response.get('msg')}")
            
            available_maps = maps_response.get('data', [])
            if not available_maps:
                raise ValueError("No maps found for robot")
            
            # 2. 选择目标地图
            target_map_id = self._select_map_from_list(
                available_maps,
                task_parameters.get('map_criteria', {})
            )
            
            # 3. 获取地图分区
            subareas = await client.call_endpoint(
                'get_map_subareas',
                path_params={'map_id': target_map_id}
            )
            
            # 4. 构建任务数据
            task_data = self._build_no_site_task_data(
                target_map_id,
                subareas,
                task_parameters
            )
            
            # 5. 下发无站点临时任务
            task_result = await client.call_endpoint(
                'submit_temp_no_site_task',
                json_data=task_data
            )
            
            logger.info(f"S-line no-site task executed successfully: {task_result}")
            return task_result
            
        except Exception as e:
            logger.error(f"S-line no-site task execution failed: {str(e)}")
            raise

src/gs_openapi/workflows/task_engine.py:383-409 (helper)

Helper function to construct the task data dictionary for submitting no-site temporary tasks.

def _build_no_site_task_data(
    self,
    map_id: str,
    subareas: Dict[str, Any],
    task_parameters: Dict[str, Any]
) -> Dict[str, Any]:
    """
    构建无站点任务数据。
    
    Args:
        map_id: 地图ID
        subareas: 地图分区信息
        task_parameters: 任务参数
        
    Returns:
        任务数据
    """
    # TODO: 根据实际API要求构建任务数据
    task_data = {
        "mapId": map_id,
        "subareas": subareas,
        "taskType": task_parameters.get('task_type', 'cleaning'),
        "cleaningMode": task_parameters.get('cleaning_mode', '__middle_cleaning'),
        **task_parameters
    }
    
    return task_data

src/gs_openapi/workflows/task_engine.py:330-353 (helper)

Helper function to select the target map from the robot's available maps based on criteria.

def _select_map_from_list(
    self,
    available_maps: List[Dict[str, Any]],
    criteria: Dict[str, Any]
) -> str:
    """
    从地图列表中选择目标地图。
    
    Args:
        available_maps: 可用地图列表  
        criteria: 选择条件
        
    Returns:
        选中的地图ID
    """
    if not criteria:
        return available_maps[0]['mapId']
    
    # TODO: 实现地图选择逻辑
    for map_info in available_maps:
        if self._matches_criteria(map_info, criteria):
            return map_info['mapId']
    
    return available_maps[0]['mapId']

GS Robot MCP Server