submit_temp_no_site_task

Instructions

Args:
    task_data: Task data including map and area information.

Returns:
    A dictionary containing the task submission result.

Implementation Reference

• src/gs_openapi/main.py:131-142 (registration)

  MCP tool registration for submit_temp_no_site_task via @mcp.tool() decorator, delegates to mcp.submit_temp_no_site_task()

  # Define submit_temp_no_site_task tool
  @mcp.tool()
  async def submit_temp_no_site_task(task_data: dict):
      """Submits a temporary task without site information for S-line robots.
  
      Args:
          task_data: Task data including map and area information.
  
      Returns:
          A dictionary containing the task submission result.
      """
      return await mcp.submit_temp_no_site_task(task_data=task_data)

• src/gs_openapi/mcp/gausium_mcp.py:288-313 (handler)

  Handler method in GausiumMCP class that calls the API endpoint 'submit_temp_no_site_task' via GausiumAPIClient

  async def submit_temp_no_site_task(
      self,
      task_data: Dict[str, Any]
  ) -> Dict[str, Any]:
      """
      S线无站点临时任务下发。
  
      Args:
          task_data: 任务数据，包含地图、区域等信息
  
      Returns:
          任务下发结果
  
      Raises:
          ValueError: 任务数据无效
          httpx.HTTPStatusError: API调用错误
          httpx.RequestError: 网络问题
      """
      if not task_data:
          raise ValueError("Task data cannot be empty")
  
      async with GausiumAPIClient() as client:
          return await client.call_endpoint(
              'submit_temp_no_site_task',
              json_data=task_data
          )

• src/gs_openapi/core/endpoints.py:133-139 (schema)

  API endpoint schema definition for submit_temp_no_site_task under ROBOT_TASK_ENDPOINTS - POST to v1alpha1/tasks/temporary/no-site

  'submit_temp_no_site_task': APIEndpoint(
      name="submit_temp_no_site_task", 
      path="tasks/temporary/no-site",
      method=HTTPMethod.POST,
      version=APIVersion.V1_ALPHA1,
      description="S线无站点临时任务下发"
  )

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

  Workflow engine method that uses submit_temp_no_site_task as part of the S-line no-site task workflow

  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