veo_get_task

Instructions

Use this to check if a generation is complete and retrieve the resulting
video URLs and metadata.

Use this when:
- You want to check if a generation has completed
- You need to retrieve video URLs from a previous generation
- You want to get the full details of a generated video

Task states:
- 'processing': Generation is still in progress
- 'succeeded': Generation finished successfully
- 'failed': Generation failed (check error message)

Returns:
    Task status and generated video information including URLs and state.

Implementation Reference

• tools/task_tools.py:14-51 (handler)

  The main handler function for the veo_get_task tool. Queries a video generation task by ID via client.query_task, throttles polling with 5s sleep if incomplete, and formats the result.

  async def veo_get_task(
      task_id: Annotated[
          str,
          Field(
              description="The task ID returned from a generation request. This is the 'task_id' field from any veo_text_to_video, veo_image_to_video, or veo_get_1080p tool response."
          ),
      ],
  ) -> str:
      """Query the status and result of a video generation task.
  
      Use this to check if a generation is complete and retrieve the resulting
      video URLs and metadata.
  
      Use this when:
      - You want to check if a generation has completed
      - You need to retrieve video URLs from a previous generation
      - You want to get the full details of a generated video
  
      Task states:
      - 'processing': Generation is still in progress
      - 'succeeded': Generation finished successfully
      - 'failed': Generation failed (check error message)
  
      Returns:
          Task status and generated video information including URLs and state.
      """
      result = await client.query_task(
          id=task_id,
          action="retrieve",
      )
      # Throttle polling: sleep 5s for incomplete tasks so LLM clients
      # don't burn through poll attempts in seconds.
      response = result.get("response", {})
      is_complete = response.get("success", False)
      if not is_complete:
          await asyncio.sleep(5)
      return format_task_result(result)

• tools/task_tools.py:14-21 (schema)

  Input schema using Pydantic's Field and Annotated to define the task_id parameter.

  async def veo_get_task(
      task_id: Annotated[
          str,
          Field(
              description="The task ID returned from a generation request. This is the 'task_id' field from any veo_text_to_video, veo_image_to_video, or veo_get_1080p tool response."
          ),
      ],
  ) -> str:

• tools/task_tools.py:13-51 (registration)

  Registration of veo_get_task via the @mcp.tool() decorator, making it available as an MCP tool.

  @mcp.tool()
  async def veo_get_task(
      task_id: Annotated[
          str,
          Field(
              description="The task ID returned from a generation request. This is the 'task_id' field from any veo_text_to_video, veo_image_to_video, or veo_get_1080p tool response."
          ),
      ],
  ) -> str:
      """Query the status and result of a video generation task.
  
      Use this to check if a generation is complete and retrieve the resulting
      video URLs and metadata.
  
      Use this when:
      - You want to check if a generation has completed
      - You need to retrieve video URLs from a previous generation
      - You want to get the full details of a generated video
  
      Task states:
      - 'processing': Generation is still in progress
      - 'succeeded': Generation finished successfully
      - 'failed': Generation failed (check error message)
  
      Returns:
          Task status and generated video information including URLs and state.
      """
      result = await client.query_task(
          id=task_id,
          action="retrieve",
      )
      # Throttle polling: sleep 5s for incomplete tasks so LLM clients
      # don't burn through poll attempts in seconds.
      response = result.get("response", {})
      is_complete = response.get("success", False)
      if not is_complete:
          await asyncio.sleep(5)
      return format_task_result(result)

• core/utils.py:74-87 (helper)

  Helper function used by veo_get_task to format the result as JSON with polling guidance metadata.

  def format_task_result(data: dict[str, Any]) -> str:
      """Format task query result as JSON.
  
      Args:
          data: API response dictionary
  
      Returns:
          JSON string representation of the result
      """
      return json.dumps(
          _with_task_guidance(data, "veo_get_task", "veo_get_tasks_batch"),
          ensure_ascii=False,
          indent=2,
      )

• core/client.py:183-191 (helper)

  The HTTP client method that sends the task query request to the /veo/tasks endpoint.

      async def query_task(self, **kwargs: Any) -> dict[str, Any]:
          """Query task status using the tasks endpoint."""
          task_id = kwargs.get("id") or kwargs.get("ids", [])
          logger.info(f"🔍 Querying task(s): {task_id}")
          return await self.request("/veo/tasks", kwargs)
  
  
  # Global client instance
  client = VeoClient()