get_operation
Check operation status and retrieve results in Media-infrastructure. Returns output URL when processing completes for media tasks like editing or format conversion.
Instructions
Get the status and result of an operation. Returns output URL when completed.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| operation_id | Yes | Operation UUID returned from an operation tool |
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) }] }