Media-infrastructure

Instructions

Implementation Reference

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

  Implementation of the get_task tool handler, which fetches task details from the VAP API. Note that the prompt mentions `get_task`, which is the correct tool name, and it is handled in `handle_tools_call` via `_handle_get_task`.

  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)
          }]
      }