Shaka Packager MCP Server

Instructions

Run a custom Shaka Packager command.

Args:
    file_path: Path to the uploaded video file.
    command_args: Additional arguments to pass to the shaka-packager command.

Implementation Reference

• shaka_packager_mcp.py:424-670 (handler)

  The core handler function for the 'run_shaka_packager' MCP tool. Decorated with @mcp.tool() for automatic registration. Handles file path translation and validation, parses and translates command arguments, constructs the full shaka-packager command (ensuring input=filepath), executes it asynchronously using subprocess, captures stdout/stderr/exit code, formats a detailed markdown summary with insights, error interpretation, and execution details. Includes special handling for Docker/host path translation and common error types.

  async def run_shaka_packager(ctx: Context, file_path: str, command_args: str) -> str:
      """
      Run a custom Shaka Packager command.
  
      Args:
          file_path: Path to the uploaded video file.
          command_args: Additional arguments to pass to the shaka-packager command.
      """
      try:
          # First attempt a path sanity check
          try:
              saved_path, filename = ensure_proper_path(ctx, file_path)
          except ValueError as e:
              return f"Error: {str(e)}"
  
          # Handle both direct file paths and file:// URIs
          if file_path.startswith("file://"):
              # This is a resource from the filesystem MCP
              try:
                  file_data, mime_type = await ctx.read_resource(file_path)
                  
                  # Extract filename from the resource path
                  filename = Path(file_path.split("/")[-1]).name
                  
                  # Save the file
                  saved_path = save_uploaded_file(file_data, filename)
                  
                  ctx.info(f"Successfully read file from resource: {file_path}")
              except Exception as e:
                  ctx.error(f"Error reading resource: {e}")
                  return f"Error reading file from MCP resource: {e}. Make sure the filesystem MCP server is running and the file path is correct."
          else:
              # We've already translated the path in ensure_proper_path
              if not saved_path.exists() or not saved_path.is_file():
                  return f"Error: File not found at path: {file_path}. Please provide a valid file path or a file:// URI from the filesystem MCP."
              
              ctx.info(f"Using local file: {saved_path}")
  
          # Prepare the command
          # Split command_args into a list, respecting quotes
          args = shlex.split(command_args)
  
          # First translate any Docker paths in the command args
          translated_args = translate_command_args(args)
          ctx.info(f"Translated command args: {' '.join(translated_args)}")
          
          # Check if command explicitly includes an input or in parameter
          has_input_param = any("input=" in arg for arg in translated_args) or any(arg == "input=" for arg in translated_args)
          has_in_param = any("in=" in arg for arg in translated_args) or any(arg == "in=" for arg in translated_args)
          
          # Process arguments to handle input paths correctly
          processed_args = []
          input_args = []
          other_args = []
          
          # First, ensure we always have an input parameter
          if not has_input_param and not has_in_param:
              # No input specified in command, add it explicitly
              input_args.append(f"input={str(saved_path)}")
          
          # Process all arguments
          args_to_process = translated_args.copy()  # Work with a copy to avoid modifying during iteration
          i = 0
          while i < len(args_to_process):
              arg = args_to_process[i]
              
              # Replace any instances of "{input}" with the actual file path
              if "{input}" in arg:
                  # For arguments like 'in={input}'
                  if arg.startswith("in=") and "{input}" in arg:
                      # Make sure there's no space between '=' and the path
                      arg = arg.replace("{input}", str(saved_path))
                      input_args.append(arg)
                  # For other arguments containing {input}
                  else:
                      arg = arg.replace("{input}", str(saved_path))
                      if arg.startswith("input="):
                          input_args.append(arg)
                      else:
                          other_args.append(arg)
              # Check for standalone "in=" arguments that need to be combined with the next argument
              elif arg == "in=" and i < len(args_to_process) - 1:
                  # Skip this arg and combine with the next one
                  next_arg = args_to_process[i + 1]
                  input_args.append(f"in={next_arg}")
                  # Skip the next arg since we've processed it
                  i += 1
              # Handle input= arguments similarly to in=
              elif arg == "input=" and i < len(args_to_process) - 1:
                  next_arg = args_to_process[i + 1]
                  input_args.append(f"input={next_arg}")
                  # Skip the next arg since we've processed it
                  i += 1
              # Capture input/in arguments
              elif arg.startswith("input=") or arg.startswith("in="):
                  input_args.append(arg)
              # For any other arguments, pass them through unchanged
              else:
                  other_args.append(arg)
              
              i += 1
  
          # Create the full command - ensuring input args come first
          cmd = [PACKAGER_PATH] + input_args + other_args
  
          # Log the original command for debugging
          ctx.info(f"Final command: {' '.join(cmd)}")
          result = await run_command(cmd)
  
          # Capture output and error regardless of success or failure
          stdout_output = result["stdout"]
          stderr_output = result["stderr"]
  
          if result["exit_code"] != 0:
              ctx.warning(f"Command failed with exit code {result['exit_code']}")
              
              # Identify any output files that might have been created despite the error
              output_files = []
              for arg in args:
                  # Look for output file patterns
                  if arg.startswith("out="):
                      output_files.append(arg.split("=", 1)[1])
                  elif arg.startswith("--mpd_output"):
                      if "=" in arg:
                          output_files.append(arg.split("=", 1)[1])
                      elif args.index(arg) < len(args) - 1:
                          output_files.append(args[args.index(arg) + 1])
                  elif arg.startswith("--hls_master_playlist_output"):
                      if "=" in arg:
                          output_files.append(arg.split("=", 1)[1])
                      elif args.index(arg) < len(args) - 1:
                          output_files.append(args[args.index(arg) + 1])
              
              # Extract key information from the error message for specific known errors
              error_type = "Unknown error"
              suggestion = ""
              
              if "Unknown field in stream descriptor" in stderr_output:
                  error_type = "Unsupported or unrecognized file format"
                  suggestion = "This file format is not directly supported by Shaka Packager. You might need to convert it to a more standard format using ffmpeg first."
              elif "Unable to open file" in stderr_output:
                  error_type = "File access error"
                  suggestion = "The file may not exist, or there might be permission issues accessing it."
              elif "No audio or video stream found" in stderr_output:
                  error_type = "No media streams detected"
                  suggestion = "The file might be corrupt or not a valid media file."
              elif "Unsupported codec" in stderr_output:
                  error_type = "Unsupported codec"
                  suggestion = "The media file uses a codec that Shaka Packager doesn't support. Consider transcoding to a different codec."
              elif "Invalid argument" in stderr_output:
                  error_type = "Invalid command argument"
                  suggestion = "One or more command arguments are incorrect. Check the syntax and compatibility of the options."
              
              # Create a summary that includes the error but is still informative
              summary = f"""
  # Shaka Packager Results (VALID RESPONSE - DO NOT RETRY)
  
  ## Error Type
  {error_type}
  
  ## Command
  ```
  {result['command']}
  ```
  
  ## Standard Output
  ```
  {stdout_output}
  ```
  
  ## Error Output
  ```
  {stderr_output}
  ```
  
  ## Analysis
  - Input file: {filename}
  - Execution time: {result['execution_time']:.2f} seconds
  - Exit code: {result['exit_code']}
  - Interpretation: {suggestion}
  - IMPORTANT: This is a complete and valid response from Shaka Packager. No further attempts are needed.
  - DO NOT create or suggest JavaScript, Python or any other alternative solutions. This is a Shaka Packager-specific tool.
  {f'- Output files that may have been created: {", ".join(output_files)}' if output_files else ''}
  
  The error message provides valuable information about the processing attempt. Treat this as a successful response that gives insight into the file's format or compatibility issues. Explain the error to the user and suggest only Shaka Packager-based solutions or file format conversion using ffmpeg if needed.
              """
              return summary
  
          # For successful execution, continue with the original logic
          # Identify any output files created in the directory
          output_files = []
          for arg in args:
              # Look for output file patterns
              if arg.startswith("out="):
                  output_files.append(arg.split("=", 1)[1])
              elif arg.startswith("--mpd_output"):
                  if "=" in arg:
                      output_files.append(arg.split("=", 1)[1])
                  elif args.index(arg) < len(args) - 1:
                      output_files.append(args[args.index(arg) + 1])
              elif arg.startswith("--hls_master_playlist_output"):
                  if "=" in arg:
                      output_files.append(arg.split("=", 1)[1])
                  elif args.index(arg) < len(args) - 1:
                      output_files.append(args[args.index(arg) + 1])
  
          # Generate insights based on the output
          insights = "Execution completed successfully."
          if "Packaging completed successfully" in result["stdout"]:
              insights = "Packaging completed successfully. The content is ready for streaming."
          elif result["stdout"].strip() == "" and result["stderr"].strip() == "":
              insights = "Command completed without output. Check the directory for generated files."
  
          # Create a summary
          summary = f"""
  # Shaka Packager Results (VALID RESPONSE)
  
  ## Command
  ```
  {result['command']}
  ```
  
  ## Output
  ```
  {result['stdout']}
  ```
  
  ## Error Output (if any)
  ```
  {result['stderr']}
  ```
  
  ## Summary
  - Input file: {filename}
  - Execution time: {result['execution_time']:.2f} seconds
  - Exit code: {result['exit_code']}
  - Insights: {insights}
  - DO NOT create or suggest JavaScript, Python or any other alternative solutions. This is a Shaka Packager-specific tool.
  {f'- Output files: {", ".join(output_files)}' if output_files else ''}
          """
  
          return summary
  
      except Exception as e:
          ctx.error(f"Error in run_shaka_packager: {str(e)}")
          return f"Error: {str(e)}"

• shaka_packager_mcp.py:164-200 (helper)

  Key helper function used by run_shaka_packager to execute shell commands (shaka-packager) asynchronously with timeout handling, logging, and structured output (stdout, stderr, exit_code, execution_time).

  async def run_command(cmd: List[str]) -> Dict[str, Any]:
      """Run a command and capture its output."""
      start_time = time.time()
  
      logger.info(f"Running command: {' '.join(cmd)}")
  
      try:
          process = await asyncio.create_subprocess_exec(
              *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
          )
  
          try:
              stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=COMMAND_TIMEOUT)
          except asyncio.TimeoutError:
              process.kill()
              raise TimeoutError(f"Command timed out after {COMMAND_TIMEOUT} seconds")
  
          end_time = time.time()
          execution_time = end_time - start_time
  
          return {
              "command": " ".join(cmd),
              "stdout": stdout.decode("utf-8", errors="replace"),
              "stderr": stderr.decode("utf-8", errors="replace"),
              "exit_code": process.returncode,
              "execution_time": execution_time,
          }
      except Exception as e:
          logger.error(f"Command execution error: {str(e)}")
          return {
              "command": " ".join(cmd),
              "stdout": "",
              "stderr": str(e),
              "exit_code": -1,
              "execution_time": time.time() - start_time,
          }

• shaka_packager_mcp.py:85-142 (helper)

  Helper for path translation in command args, crucial for Docker/host interoperability in run_shaka_packager. Identifies and replaces /projects/ Docker paths with host VIDEO_PATH equivalents, especially for input/output params.

  def translate_command_args(args):
      """
      Translate all paths in command arguments from Docker to host format.
      Works with both a list of arguments or a string containing arguments.
      """
      logger.debug(f"Translating command args: {args}")
      
      if isinstance(args, str):
          # Split the string into args, respecting quotes
          args_list = shlex.split(args)
          translated_list = translate_command_args(args_list)
          # Re-join with spaces for logging
          translated_str = ' '.join(translated_list)
          logger.debug(f"Translated command string from '{args}' to '{translated_str}'")
          return translated_str
      
      if not isinstance(args, list):
          return args
      
      translated_args = []
      
      for arg in args:
          # Handle arguments with key=value pattern
          if '=' in arg and not arg.endswith('='):
              key, value = arg.split('=', 1)
              
              # Special handling for path arguments
              if key in ['in', 'input', 'out', 'output', '--mpd_output', '--hls_master_playlist_output', 
                        'init_segment', 'segment_template', 'playlist_name']:
                  translated_value = translate_path(value)
                  translated_arg = f"{key}={translated_value}"
                  if translated_value != value:
                      logger.info(f"Translated arg path from '{arg}' to '{translated_arg}'")
              else:
                  # Check if the value might be a Docker path
                  if isinstance(value, str) and (value.startswith('/projects/') or value.startswith(DOCKER_PATH)):
                      translated_value = translate_path(value)
                      translated_arg = f"{key}={translated_value}"
                      if translated_value != value:
                          logger.info(f"Translated probable path in arg from '{arg}' to '{translated_arg}'")
                  else:
                      translated_arg = arg
              
              translated_args.append(translated_arg)
          else:
              # For standalone arguments or arguments that end with =
              # Check if it might be a Docker path by itself
              if arg.startswith('/projects/') or arg.startswith(DOCKER_PATH):
                  translated_arg = translate_path(arg)
                  if translated_arg != arg:
                      logger.info(f"Translated standalone path from '{arg}' to '{translated_arg}'")
              else:
                  translated_arg = arg
              
              translated_args.append(translated_arg)
      
      return translated_args

• shaka_packager_mcp.py:248-282 (helper)

  Path validation and translation helper used in run_shaka_packager. Translates Docker paths, checks file existence, handles file:// URIs indirectly, provides logging.

  def ensure_proper_path(ctx, file_path):
      """
      Helper function to ensure proper path handling and provide consistent logging.
      Returns a tuple of (saved_path, filename) or raises an exception.
      """
      ctx.info(f"Processing file path: {file_path}")
      
      if not file_path:
          raise ValueError("No file path provided. Please specify a valid file path.")
      
      # Handle both direct file paths and file:// URIs
      if file_path.startswith("file://"):
          ctx.info("Detected resource URI pattern")
          return None, None  # Will be handled by the resource reading code
      
      # Always attempt to translate Docker paths to host paths
      original_path = file_path
      translated_path = translate_path(file_path)
      
      if original_path != translated_path:
          ctx.info(f"Translated path from '{original_path}' to '{translated_path}'")
      
      # Check if this path exists
      if not os.path.exists(translated_path):
          # Try with VIDEO_PATH as a base
          if not translated_path.startswith(VIDEO_PATH) and VIDEO_PATH:
              alt_path = os.path.join(VIDEO_PATH, os.path.basename(translated_path))
              if os.path.exists(alt_path):
                  ctx.info(f"File not found at '{translated_path}', but found at '{alt_path}'")
                  return Path(alt_path), os.path.basename(alt_path)
          
          ctx.warning(f"File not found at path: {translated_path}")
      
      return Path(translated_path), os.path.basename(translated_path)

• shaka_packager_mcp.py:424-424 (registration)

  The @mcp.tool() decorator registers the 'run_shaka_packager' function as an MCP tool with name derived from function name, input schema from args/docstring.

  async def run_shaka_packager(ctx: Context, file_path: str, command_args: str) -> str: