get_task

Instructions

Implementation Reference

• mcp/vap_mcp_proxy.py:313-385 (handler)

  Main handler for 'get_task' tool. Fetches task status from the V3 API, extracts video/image/audio URLs from the result, and builds a formatted text response showing status and output URLs.

  def _handle_get_task(arguments: Dict) -> Dict:
      """
      Handle get_task tool call (Directive #241).
  
      Fetches task status from V3 API with proper video_url extraction.
      """
      task_id = arguments.get("task_id", "")
  
      if not task_id:
          return {
              "isError": True,
              "content": [{"type": "text", "text": "Error: task_id is required"}]
          }
  
      # Fetch task from V3 API
      response = make_v3_get_request(f"/v3/tasks/{task_id}")
  
      if "error" in response:
          return {
              "isError": True,
              "content": [{"type": "text", "text": f"Error: {response['error']}"}]
          }
  
      # Extract fields
      status = response.get("status", "unknown")
      task_type = response.get("type", "unknown")
      estimated_cost = response.get("estimated_cost", "N/A")
      actual_cost = response.get("actual_cost", "N/A")
      error_message = response.get("error_message")
      result = response.get("result", {}) or {}
  
      # Extract output URL (video, image, or audio)
      video_url = result.get("video_url") or result.get("output_url")
      image_url = result.get("image_url") or result.get("output_url")
  
      # D#508: Music tasks have audio_url in items array
      audio_url = None
      items = result.get("items", [])
      if items and isinstance(items, list) and len(items) > 0:
          audio_url = items[0].get("audio_url")
          # Also check for image_url in items (for image tasks)
          if not image_url:
              image_url = items[0].get("image_url")
  
      # Build response text
      lines = [
          f"Task: {task_id}",
          f"Type: {task_type}",
          f"Status: {status}",
          f"Estimated Cost: ${estimated_cost}",
      ]
  
      if actual_cost and actual_cost != "N/A":
          lines.append(f"Actual Cost: ${actual_cost}")
  
      if status == "completed":
          if audio_url:
              lines.append(f"\n🎵 Audio URL: {audio_url}")
          elif video_url:
              lines.append(f"\n🎬 Video URL: {video_url}")
          elif image_url:
              lines.append(f"\n🖼️ Image URL: {image_url}")
      elif status == "failed" and error_message:
          lines.append(f"\n❌ Error: {error_message}")
      elif status in ("pending", "queued", "executing"):
          lines.append(f"\n⏳ Task is still {status}. Check again shortly.")
  
      return {
          "content": [{
              "type": "text",
              "text": "\n".join(lines)
          }]
      }

• mcp/vap_mcp_proxy.py:388-411 (helper)

  Helper function that performs the HTTP GET request to the V3 API. Used by _handle_get_task to fetch task data.

  def make_v3_get_request(endpoint: str) -> Dict[str, Any]:
      """Make HTTP GET request to VAP V3 API (Directive #241)."""
      url = f"{API_BASE_URL}{endpoint}"
      headers = {
          "Content-Type": "application/json",
      }
      if API_KEY:
          headers["Authorization"] = f"Bearer {API_KEY}"
  
      logger.debug(f"V3 GET Request: {url}")
  
      try:
          if hasattr(httpx, 'Client'):
              with httpx.Client(timeout=30.0) as client:
                  response = client.get(url, headers=headers)
                  response.raise_for_status()
                  return response.json()
          else:
              response = httpx.get(url, headers=headers, timeout=30)
              response.raise_for_status()
              return response.json()
      except Exception as e:
          logger.error(f"V3 GET error: {e}")
          return {"error": str(e)}

• mcp/vap_mcp_proxy.py:210-211 (registration)

  Dispatches the 'get_task' tool call to _handle_get_task in the tools/call handler.

  if tool_name == "get_task":
      return _handle_get_task(arguments)

• sdk/vape_client/models.py:198-223 (schema)

  TaskResult data model used by the SDK client's get_task method to parse API responses.

  class TaskResult:
      """Result of task status query."""
      task_id: str
      status: str
      task_type: Optional[str] = None
      result_url: Optional[str] = None
      cost: float = 0.0
      created_at: Optional[str] = None
      completed_at: Optional[str] = None
      error: Optional[str] = None
      metadata: Optional[Dict[str, Any]] = None
  
      @classmethod
      def from_response(cls, data: dict) -> "TaskResult":
          """Create from API response."""
          return cls(
              task_id=data.get("task_id", ""),
              status=data.get("status", "unknown"),
              task_type=data.get("task_type") or data.get("type"),
              result_url=data.get("result_url") or data.get("image_url") or data.get("video_url") or data.get("audio_url"),
              cost=data.get("cost", 0.0),
              created_at=data.get("created_at"),
              completed_at=data.get("completed_at"),
              error=data.get("error"),
              metadata=data.get("metadata"),
          )

• sdk/vape_client/client.py:371-397 (handler)

  SDK client's synchronous get_task method that makes a GET request to /v3/tasks/{task_id} and returns a TaskResult.

          duration: Target duration in seconds (30-480, default 120)
          instrumental: Generate without vocals
          loudness_preset: Normalization (streaming, apple, broadcast)
          audio_format: Output format (mp3 or wav)
  
      Returns:
          MusicResult with task_id for async tracking. Cost: $0.68
      """
      payload = {
          "type": "music_generation",
          "params": {
              "description": prompt,
              "duration": duration,
              "instrumental": instrumental,
              "loudness_preset": loudness_preset,
              "audio_format": audio_format,
          }
      }
  
      data = self._request("POST", "/v3/tasks", json=payload)
      return MusicResult.from_response(data)
  
  # ============================================
  # Task Management
  # ============================================
  
  def get_task(self, task_id: str) -> TaskResult: