Shaka Packager MCP Server

Instructions

Implementation Reference

• shaka_packager_mcp.py:423-669 (handler)

  The core handler function for the 'run_shaka_packager' tool. Decorated with @mcp.tool() for automatic registration. Handles file path resolution (including MCP resources and Docker path translation), command argument processing and path substitution ({input}), executes shaka-packager via run_command, and returns formatted markdown output with stdout/stderr/insights regardless of success.
  @mcp.tool() 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:423-423 (registration)

  The @mcp.tool() decorator registers the run_shaka_packager function as an MCP tool.
  @mcp.tool()
• shaka_packager_mcp.py:164-200 (helper)

  Key helper function used by the tool to execute subprocess commands (shaka-packager) asynchronously with timeout, capturing stdout/stderr/exit_code.
  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:49-82 (helper)

  Helper for translating Docker container paths (/projects/video-drop) to host paths based on VIDEO_PATH environment variable.
  def translate_path(file_path): """Translate paths between Docker container and host system formats""" # Log the translation attempt for debugging logger.debug(f"Translating path: {file_path}, VIDEO_PATH={VIDEO_PATH}, DOCKER_PATH={DOCKER_PATH}") # Skip if not a string if not isinstance(file_path, str): return file_path # Always check for Docker path and replace it if found if VIDEO_PATH and file_path.startswith(DOCKER_PATH): translated = file_path.replace(DOCKER_PATH, VIDEO_PATH) logger.info(f"Translated Docker path '{file_path}' to host path '{translated}'") return translated # If file starts with /projects but not exactly DOCKER_PATH, it might still be a Docker path if VIDEO_PATH and file_path.startswith("/projects/"): # Try to extract the relative path rel_path = file_path[len("/projects/"):] translated = os.path.join(VIDEO_PATH, rel_path) logger.info(f"Translated likely Docker path '{file_path}' to host path '{translated}'") return translated # Check if the path exists in the Docker path structure but wasn't translated possible_docker_path = os.path.join(DOCKER_PATH, os.path.basename(file_path)) if VIDEO_PATH and os.path.basename(file_path) and not os.path.exists(file_path): # Try as a child of VIDEO_PATH possible_host_path = os.path.join(VIDEO_PATH, os.path.basename(file_path)) if os.path.exists(possible_host_path): logger.info(f"Remapped path '{file_path}' to '{possible_host_path}' based on file existence") return possible_host_path # If not translated and not a video path, return original return file_path