Shaka Packager MCP Server

Instructions

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: