Media-infrastructure

Instructions

Implementation Reference

• mcp/vap_mcp_proxy.py:221-286 (handler)

  The handler for video generation, which is the closest functionality to 'video_merge' available in the codebase, although 'video_merge' itself is not found.

  def _handle_generate_video(arguments: Dict) -> Dict:
      """
      Handle generate_video tool call (Directive #242: Veo 3.1).
  
      Creates a video generation task via V3 API.
      """
      prompt = arguments.get("prompt", "")
      duration = arguments.get("duration", 8)
      aspect_ratio = arguments.get("aspect_ratio", "16:9")
      generate_audio = arguments.get("generate_audio", True)
      resolution = arguments.get("resolution", "720p")
      negative_prompt = arguments.get("negative_prompt", "")
  
      if not prompt:
          return {
              "isError": True,
              "content": [{"type": "text", "text": "Error: prompt is required"}]
          }
  
      # Validate duration (Veo 3.1: 4, 6, 8 seconds)
      if duration not in (4, 6, 8):
          duration = 8
  
      # Validate aspect ratio (Veo 3.1: 16:9, 9:16)
      if aspect_ratio not in ("16:9", "9:16"):
          aspect_ratio = "16:9"
  
      # Validate resolution
      if resolution not in ("720p", "1080p"):
          resolution = "720p"
  
      # Create task via V3 API
      logger.info(f"Creating Veo 3.1 video task: duration={duration}s, audio={generate_audio}")
  
      params = {
          "prompt": prompt,
          "duration": duration,
          "aspect_ratio": aspect_ratio,
          "generate_audio": generate_audio,
          "resolution": resolution
      }
      if negative_prompt:
          params["negative_prompt"] = negative_prompt
  
      response = make_v3_request("/v3/tasks", {
          "type": "video",
          "params": params
      })
  
      if "error" in response:
          return {
              "isError": True,
              "content": [{"type": "text", "text": f"Error: {response['error']}"}]
          }
  
      # Format success response
      task_id = response.get("task_id", "unknown")
      cost_table = VIDEO_COSTS_WITH_AUDIO if generate_audio else VIDEO_COSTS_NO_AUDIO
      estimated_cost = response.get("estimated_cost", cost_table.get(duration, 4.80))
  
      return {
          "content": [{
              "type": "text",
              "text": f"Video generation task created (Veo 3.1)!\n\nTask ID: {task_id}\nDuration: {duration} seconds\nAspect Ratio: {aspect_ratio}\nResolution: {resolution}\nAudio: {'Yes' if generate_audio else 'No'}\nEstimated Cost: ${estimated_cost}\n\nUse get_task with this task_id to check status and get the video URL when complete."
          }]
      }