Video & Audio Editing MCP Server

Instructions

Implementation Reference

• server.py:775-1034 (handler)

  The core handler function for the 'concatenate_videos' MCP tool. Handles both simple concatenation of multiple videos (via normalization and concat demuxer) and advanced xfade transitions for exactly two videos. Automatically registered via @mcp.tool() decorator using FastMCP.

  @mcp.tool()
  def concatenate_videos(video_paths: list[str], output_video_path: str,
                         transition_effect: str = None, transition_duration: float = None) -> str:
      """Concatenates multiple video files into a single output file.
      Supports optional xfade transition when concatenating exactly two videos.
  
      Args:
          video_paths: A list of paths to the video files to concatenate.
          output_video_path: The path to save the concatenated video file.
          transition_effect (str, optional): The xfade transition type. Options:
              - 'dissolve': Gradual blend between clips
              - 'fade': Simple fade through black
              - 'fadeblack': Fade through black
              - 'fadewhite': Fade through white
              - 'fadegrays': Fade through grayscale
              - 'distance': Distance transform transition
              - 'wipeleft', 'wiperight': Horizontal wipe
              - 'wipeup', 'wipedown': Vertical wipe
              - 'slideleft', 'slideright': Horizontal slide
              - 'slideup', 'slidedown': Vertical slide
              - 'smoothleft', 'smoothright': Smooth horizontal slide
              - 'smoothup', 'smoothdown': Smooth vertical slide
              - 'circlecrop': Rectangle crop transition
              - 'rectcrop': Rectangle crop transition
              - 'circleopen', 'circleclose': Circle open/close
              - 'vertopen', 'vertclose': Vertical open/close
              - 'horzopen', 'horzclose': Horizontal open/close
              - 'diagtl', 'diagtr', 'diagbl', 'diagbr': Diagonal transitions
              - 'hlslice', 'hrslice': Horizontal slice
              - 'vuslice', 'vdslice': Vertical slice
              - 'pixelize': Pixelize effect
              - 'radial': Radial transition
              - 'hblur': Horizontal blur
              Only applied if exactly two videos are provided. Defaults to None (no transition).
          transition_duration (float, optional): The duration of the xfade transition in seconds. 
                                               Required if transition_effect is specified. Defaults to None.
      
      Returns:
          A status message indicating success or failure.
      """
      if not video_paths:
          return "Error: No video paths provided for concatenation."
      if len(video_paths) < 1: # Allow single video to be "concatenated" (effectively copied/re-encoded)
          return "Error: At least one video is required."
      
      if transition_effect and transition_duration is None:
          return "Error: transition_duration is required when transition_effect is specified."
      if transition_effect and transition_duration <= 0:
          return "Error: transition_duration must be positive."
  
      # Validate transition_effect
      valid_transitions = {
          'dissolve', 'fade', 'fadeblack', 'fadewhite', 'fadegrays', 'distance',
          'wipeleft', 'wiperight', 'wipeup', 'wipedown',
          'slideleft', 'slideright', 'slideup', 'slidedown',
          'smoothleft', 'smoothright', 'smoothup', 'smoothdown',
          'circlecrop', 'rectcrop', 'circleopen', 'circleclose',
          'vertopen', 'vertclose', 'horzopen', 'horzclose',
          'diagtl', 'diagtr', 'diagbl', 'diagbr',
          'hlslice', 'hrslice', 'vuslice', 'vdslice',
          'pixelize', 'radial', 'hblur'
      }
      if transition_effect and transition_effect not in valid_transitions:
          return f"Error: Invalid transition_effect '{transition_effect}'. Valid options: {', '.join(sorted(valid_transitions))}"
  
      # Check if all input files exist
      for video_path in video_paths:
          if not os.path.exists(video_path):
              return f"Error: Input video file not found at {video_path}"
  
      # Handle single video case (copy or re-encode to target)
      if len(video_paths) == 1:
          try:
              # Simple copy if no processing needed, or re-encode to a standard format.
              # For now, let's assume re-encoding to ensure it matches expectations of a processed file.
              # This could be enhanced to use target_props like in add_b_roll if needed.
              ffmpeg.input(video_paths[0]).output(output_video_path, vcodec='libx264', acodec='aac').run(capture_stdout=True, capture_stderr=True)
              return f"Single video processed and saved to {output_video_path}"
          except ffmpeg.Error as e:
              return f"Error processing single video: {e.stderr.decode('utf8') if e.stderr else str(e)}"
  
      # Handle xfade transition for exactly two videos
      if transition_effect and len(video_paths) == 2:
          # Create a temporary directory for intermediate files
          temp_dir = tempfile.mkdtemp()
          try:
              video1_path = video_paths[0]
              video2_path = video_paths[1]
              
              props1 = _get_media_properties(video1_path)
              props2 = _get_media_properties(video2_path)
  
              if not props1['has_video'] or not props2['has_video']:
                  return "Error: xfade transition requires both inputs to be videos."
              if transition_duration >= props1['duration']:
                   return f"Error: Transition duration ({transition_duration}s) cannot be equal or longer than the first video's duration ({props1['duration']})."
  
              # Check if both videos have audio
              has_audio = props1['has_audio'] and props2['has_audio']
              if not has_audio:
                  print("Warning: At least one video lacks audio. Xfade will be video-only or silent audio.")
  
              # Determine common target properties for normalization before xfade
              # Preferring higher resolution/fps from inputs, or defaulting.
              target_w = max(props1['width'], props2['width'], 640) 
              target_h = max(props1['height'], props2['height'], 360)
              # Ensure a common FPS, e.g., highest of the two, or a default like 30
              target_fps = max(props1['avg_fps'], props2['avg_fps'], 30)
              if target_fps <= 0: 
                  target_fps = 30 # safety net
  
              # Normalize input videos to have same dimensions and properties
              # First video
              norm_video1_path = os.path.join(temp_dir, "norm_video1.mp4")
              try:
                  # Scale and set properties
                  subprocess.run([
                      'ffmpeg',
                      '-i', video1_path,
                      '-vf', f'scale={target_w}:{target_h}',
                      '-r', str(target_fps),
                      '-c:v', 'libx264',
                      '-c:a', 'aac',
                      '-y',
                      norm_video1_path
                  ], check=True, capture_output=True)
              except subprocess.CalledProcessError as e:
                  return f"Error normalizing first video: {e.stderr.decode('utf8') if e.stderr else str(e)}"
  
              # Second video
              norm_video2_path = os.path.join(temp_dir, "norm_video2.mp4")
              try:
                  # Scale and set properties
                  subprocess.run([
                      'ffmpeg',
                      '-i', video2_path,
                      '-vf', f'scale={target_w}:{target_h}',
                      '-r', str(target_fps),
                      '-c:v', 'libx264',
                      '-c:a', 'aac',
                      '-y',
                      norm_video2_path
                  ], check=True, capture_output=True)
              except subprocess.CalledProcessError as e:
                  return f"Error normalizing second video: {e.stderr.decode('utf8') if e.stderr else str(e)}"
  
              # Get normalized video 1 duration
              norm_props1 = _get_media_properties(norm_video1_path)
              norm_video1_duration = norm_props1['duration']
              
              if transition_duration >= norm_video1_duration:
                  return f"Error: Transition duration ({transition_duration}s) is too long for the normalized first video ({norm_video1_duration}s)."
  
              # Calculate offset (where second video starts relative to first)
              offset = norm_video1_duration - transition_duration
              
              # Create filter complex for xfade transition
              filter_complex = f"[0:v][1:v]xfade=transition={transition_effect}:duration={transition_duration}:offset={offset}"
              
              # Base command for video transition
              cmd = [
                  'ffmpeg',
                  '-i', norm_video1_path,
                  '-i', norm_video2_path,
                  '-filter_complex'
              ]
              
              # Add appropriate filters for video and audio
              if has_audio:
                  # Audio transition (crossfade)
                  filter_complex += f",[0:a][1:a]acrossfade=d={transition_duration}:c1=tri:c2=tri"
                  cmd.extend([filter_complex, '-map', '[v]', '-map', '[a]'])
              else:
                  # Video only
                  filter_complex += "[v]"
                  cmd.extend([filter_complex, '-map', '[v]'])
              
              # Add output file and encoding parameters
              cmd.extend([
                  '-c:v', 'libx264',
                  '-c:a', 'aac',
                  '-y',
                  output_video_path
              ])
              
              try:
                  subprocess.run(cmd, check=True, capture_output=True)
                  return f"Videos concatenated successfully with '{transition_effect}' transition to {output_video_path}"
              except subprocess.CalledProcessError as e:
                  return f"Error during xfade process: {e.stderr.decode('utf8') if e.stderr else str(e)}"
                  
          except Exception as e:
              return f"An unexpected error occurred during xfade concatenation: {str(e)}"
          finally:
              # Clean up temporary directory
              shutil.rmtree(temp_dir)
      
      elif transition_effect and len(video_paths) > 2:
          return f"Error: xfade transition ('{transition_effect}') is currently only supported for exactly two videos. Found {len(video_paths)} videos."
  
      # Standard concatenation for 2+ videos without xfade
      # We'll use the concat demuxer approach
      temp_dir = tempfile.mkdtemp()
      try:
          # Normalize all videos to the same format/codec/resolution
          normalized_paths = []
          
          # Get target properties from first video
          first_props = _get_media_properties(video_paths[0])
          target_w = first_props['width'] if first_props['width'] > 0 else 1280
          target_h = first_props['height'] if first_props['height'] > 0 else 720
          target_fps = first_props['avg_fps'] if first_props['avg_fps'] > 0 else 30
          if target_fps <= 0:
              target_fps = 30
          
          # Process each video
          for i, video_path in enumerate(video_paths):
              norm_path = os.path.join(temp_dir, f"norm_{i}.mp4")
              try:
                  subprocess.run([
                      'ffmpeg',
                      '-i', video_path,
                      '-vf', f'scale={target_w}:{target_h}',
                      '-r', str(target_fps),
                      '-c:v', 'libx264',
                      '-c:a', 'aac',
                      '-y',
                      norm_path
                  ], check=True, capture_output=True)
                  normalized_paths.append(norm_path)
              except subprocess.CalledProcessError as e:
                  return f"Error normalizing video {i}: {e.stderr.decode('utf8') if e.stderr else str(e)}"
          
          # Create a concat file
          concat_list_path = os.path.join(temp_dir, "concat_list.txt")
          with open(concat_list_path, 'w') as f:
              for path in normalized_paths:
                  f.write(f"file '{path}'\n")
          
          # Run ffmpeg concat
          try:
              subprocess.run([
                  'ffmpeg',
                  '-f', 'concat',
                  '-safe', '0',
                  '-i', concat_list_path,
                  '-c', 'copy',
                  '-y',
                  output_video_path
              ], check=True, capture_output=True)
              return f"Videos concatenated successfully to {output_video_path}"
          except subprocess.CalledProcessError as e:
              return f"Error during concatenation: {e.stderr.decode('utf8') if e.stderr else str(e)}"
              
      except Exception as e:
          return f"An unexpected error occurred during standard concatenation: {str(e)}"
      finally:
          # Clean up temporary directory
          shutil.rmtree(temp_dir)

• server.py:1237-1269 (helper)

  Helper function used by concatenate_videos to retrieve media properties (duration, resolution, FPS, etc.) for normalization before concatenation and transition checks.

  def _get_media_properties(media_path: str) -> dict:
      """Probes media file and returns key properties."""
      try:
          probe = ffmpeg.probe(media_path)
          video_stream_info = next((s for s in probe['streams'] if s['codec_type'] == 'video'), None)
          audio_stream_info = next((s for s in probe['streams'] if s['codec_type'] == 'audio'), None)
          
          props = {
              'duration': float(probe['format'].get('duration', 0.0)),
              'has_video': video_stream_info is not None,
              'has_audio': audio_stream_info is not None,
              'width': int(video_stream_info['width']) if video_stream_info and 'width' in video_stream_info else 0,
              'height': int(video_stream_info['height']) if video_stream_info and 'height' in video_stream_info else 0,
              'avg_fps': 0, # Default, will be calculated if possible
              'sample_rate': int(audio_stream_info['sample_rate']) if audio_stream_info and 'sample_rate' in audio_stream_info else 44100,
              'channels': int(audio_stream_info['channels']) if audio_stream_info and 'channels' in audio_stream_info else 2,
              'channel_layout': audio_stream_info.get('channel_layout', 'stereo') if audio_stream_info else 'stereo'
          }
          if video_stream_info and 'avg_frame_rate' in video_stream_info and video_stream_info['avg_frame_rate'] != '0/0':
              num, den = map(int, video_stream_info['avg_frame_rate'].split('/'))
              if den > 0:
                  props['avg_fps'] = num / den
              else:
                  props['avg_fps'] = 30 # Default if denominator is 0
          else: # Fallback if avg_frame_rate is not useful
              props['avg_fps'] = 30 # A common default
  
          return props
      except ffmpeg.Error as e:
          raise RuntimeError(f"Error probing file {media_path}: {e.stderr.decode('utf8') if e.stderr else str(e)}")
      except Exception as e:
          raise RuntimeError(f"Unexpected error probing file {media_path}: {str(e)}")

• server.py:1223-1235 (helper)

  Utility helper for parsing time strings to seconds, used in B-roll but available for timing in video tools including transitions.

  def _parse_time_to_seconds(time_str: str) -> float:
      """Converts HH:MM:SS.mmm or seconds string to float seconds."""
      if isinstance(time_str, (int, float)):
          return float(time_str)
      if ':' in time_str:
          parts = time_str.split(':')
          if len(parts) == 3:
              return int(parts[0]) * 3600 + int(parts[1]) * 60 + float(parts[2])
          elif len(parts) == 2:
              return int(parts[0]) * 60 + float(parts[1])
          else:
              raise ValueError(f"Invalid time format: {time_str}")
      return float(time_str)