get_task
Check the status and retrieve results of media generation tasks, including image URLs when processing is complete.
Instructions
Get the status and result of a generation task. Returns image URL when completed.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| task_id | Yes | Task UUID returned from generate_image |
Implementation Reference
- mcp/vap_mcp_proxy.py:313-385 (handler)The handler function `_handle_get_task` in `mcp/vap_mcp_proxy.py` implements the logic for the `get_task` tool, fetching data from the V3 API and formatting the result.
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) }] }