#!/usr/bin/env python3 """ FCPXML MCP Server — Batch operations and analysis for Final Cut Pro XML files. Composition root: wires MCP resources/prompts, concatenates each server_tools/*.py module's TOOLS and HANDLERS into list_tools() and TOOL_HANDLERS, and dispatches. No tool logic lives here — see Engine/docs/03_SERVER_TOOLS.md for the catalog and server_tools/ for the 73 tool definitions + handlers, grouped by category. Author: DareDev256 (https://github.com/DareDev256) """ from __future__ import annotations from pathlib import Path from typing import Any, Sequence from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( GetPromptResult, Prompt, PromptArgument, PromptMessage, Resource, TextContent, ) from server_tools import ( editing, export, generation, live, markers_import, qc, roles, subtitles, timeline, transcript, voice, ) from server_tools._shared import ( _DIARIZATION_INSTALL_HINT, _FEATURES_INSTALL_HINT, _MAX_JSON_DEPTH, _SANDBOX_ENABLED, _TRANSCRIBE_INSTALL_HINT, AUDIO_MEDIA_EXTENSIONS, MAX_FILE_SIZE, MAX_MEDIA_FILE_SIZE, PROJECTS_DIR, TRANSCRIBE_MAX_MEDIA, _apply_placed_action, _check_json_depth, _cut_transcript_spans, _detect_duplicate_groups, _detect_flash_frames, _detect_gaps, _extract_subtitle_blocks, _fmt_suggestions, _format_batch_result, _format_clip_table, _load_or_transcribe, _markdown_table, _no_timeline, _NoTimelineError, _parse_project, _parse_timestamp_parts, _raw_markers_to_batch, _require_timeline, _resolve_io_paths, _setup_generator, _setup_modifier, _speaker_table, _text_result, _transcript_cut_report, _transcript_json_path, _validate_directory, _validate_filepath, _validate_output_path, _voice_analysis_config_text, find_fcpxml_files, format_duration, format_timecode, generate_output_path, parse_srt, parse_transcript_timestamps, parse_vtt, ) from server_tools.editing import ( handle_add_audio, handle_add_connected_clip, handle_add_marker, handle_add_transition, handle_add_zoom, handle_batch_add_markers, handle_change_speed, handle_create_compound_clip, handle_delete_clips, handle_fill_gaps, handle_fix_flash_frames, handle_flatten_compound_clip, handle_insert_clip, handle_rapid_trim, handle_reformat_timeline, handle_reorder_clips, handle_split_clip, handle_trim_clip, ) from server_tools.export import ( handle_export_csv, handle_export_edl, handle_export_fcp7_xml, handle_export_resolve_xml, handle_relink_media, ) from server_tools.generation import ( handle_apply_template, handle_auto_rough_cut, handle_generate_ab_roll, handle_generate_montage, handle_list_templates, ) from server_tools.live import ( handle_list_fcp_libraries, handle_push_to_fcp, ) from server_tools.markers_import import ( handle_import_beat_markers, handle_import_srt_markers, handle_import_transcript_markers, handle_snap_to_beats, handle_transcript_markers, ) from server_tools.qc import ( handle_analyze_pacing, handle_detect_beats, handle_detect_duplicates, handle_detect_flash_frames, handle_detect_gaps, handle_detect_media_silence, handle_detect_silence_candidates, handle_find_long_clips, handle_find_short_cuts, handle_remove_media_silence, handle_remove_silence_candidates, handle_validate_timeline, ) from server_tools.roles import ( handle_assign_role, handle_export_role_stems, handle_filter_by_role, ) from server_tools.subtitles import ( handle_generate_dynamic_subtitles, handle_generate_plain_subtitles, handle_validate_subtitle_layout, ) from server_tools.timeline import ( handle_analyze_timeline, handle_diff_timelines, handle_list_clips, handle_list_compound_clips, handle_list_connected_clips, handle_list_effects, handle_list_keywords, handle_list_library_clips, handle_list_markers, handle_list_projects, handle_list_roles, ) from server_tools.transcript import ( handle_edit_by_transcript, handle_remove_filler_words, handle_transcribe_media, ) from server_tools.voice import ( handle_analyze_voice_features, handle_apply_voice_actions, handle_build_voice_timeline, handle_generate_voice_script, handle_diarize_media, handle_get_voice_analysis_config, handle_refine_voice_timeline, handle_remove_speakers, handle_save_voice_analysis_config, ) # Re-exported below (via __all__) so `from server import handle_x` / `_helper` # keeps working for tests and any external caller that imported the old # monolith directly — this module is the stable public surface, # server_tools/ is the implementation. __all__ tells the linter these # imports are the point, not dead code. __all__ = [ "server", "main", "main_sync", "call_tool", "list_tools", "list_resources", "read_resource", "list_prompts", "get_prompt", "TOOL_HANDLERS", "PROJECTS_DIR", "_SANDBOX_ENABLED", "MAX_FILE_SIZE", "MAX_MEDIA_FILE_SIZE", "_MAX_JSON_DEPTH", "_check_json_depth", "_validate_filepath", "_validate_output_path", "_validate_directory", "find_fcpxml_files", "format_timecode", "format_duration", "_format_clip_table", "_markdown_table", "_format_batch_result", "_fmt_suggestions", "generate_output_path", "_parse_project", "_text_result", "_no_timeline", "_require_timeline", "_NoTimelineError", "_resolve_io_paths", "_setup_modifier", "_setup_generator", "_parse_timestamp_parts", "_raw_markers_to_batch", "_extract_subtitle_blocks", "parse_srt", "parse_vtt", "parse_transcript_timestamps", "_detect_flash_frames", "_detect_gaps", "_detect_duplicate_groups", "AUDIO_MEDIA_EXTENSIONS", "_DIARIZATION_INSTALL_HINT", "_FEATURES_INSTALL_HINT", "_voice_analysis_config_text", "_apply_placed_action", "_speaker_table", "TRANSCRIBE_MAX_MEDIA", "_TRANSCRIBE_INSTALL_HINT", "_transcript_json_path", "_load_or_transcribe", "_cut_transcript_spans", "_transcript_cut_report", "handle_list_projects", "handle_analyze_timeline", "handle_list_clips", "handle_list_markers", "handle_list_keywords", "handle_list_library_clips", "handle_list_connected_clips", "handle_list_compound_clips", "handle_list_roles", "handle_diff_timelines", "handle_list_effects", "handle_find_short_cuts", "handle_find_long_clips", "handle_analyze_pacing", "handle_detect_flash_frames", "handle_detect_duplicates", "handle_detect_gaps", "handle_validate_timeline", "handle_detect_media_silence", "handle_detect_beats", "handle_remove_media_silence", "handle_detect_silence_candidates", "handle_remove_silence_candidates", "handle_add_marker", "handle_batch_add_markers", "handle_trim_clip", "handle_reorder_clips", "handle_add_transition", "handle_change_speed", "handle_add_zoom", "handle_delete_clips", "handle_split_clip", "handle_insert_clip", "handle_fix_flash_frames", "handle_rapid_trim", "handle_fill_gaps", "handle_add_connected_clip", "handle_reformat_timeline", "handle_add_audio", "handle_create_compound_clip", "handle_flatten_compound_clip", "handle_auto_rough_cut", "handle_generate_montage", "handle_generate_ab_roll", "handle_list_templates", "handle_apply_template", "handle_import_beat_markers", "handle_snap_to_beats", "handle_import_srt_markers", "handle_import_transcript_markers", "handle_transcript_markers", "handle_assign_role", "handle_filter_by_role", "handle_export_role_stems", "handle_transcribe_media", "handle_edit_by_transcript", "handle_remove_filler_words", "handle_export_edl", "handle_export_csv", "handle_export_resolve_xml", "handle_export_fcp7_xml", "handle_relink_media", "handle_diarize_media", "handle_analyze_voice_features", "handle_build_voice_timeline", "handle_remove_speakers", "handle_refine_voice_timeline", "handle_apply_voice_actions", "handle_get_voice_analysis_config", "handle_save_voice_analysis_config", "handle_validate_subtitle_layout", "handle_generate_dynamic_subtitles", "handle_generate_plain_subtitles", "handle_push_to_fcp", "handle_list_fcp_libraries", ] __version__ = "0.13.1" server = Server("fcp-mcp-server", version=__version__) # The category modules composing this server's tool catalog, in the order # Engine/docs/03_SERVER_TOOLS.md lists them. _CATEGORY_MODULES = ( timeline, qc, editing, generation, markers_import, roles, transcript, export, voice, subtitles, live, ) @server.list_resources() async def list_resources() -> list[Resource]: """Expose discovered FCPXML files as MCP resources.""" files = find_fcpxml_files(PROJECTS_DIR) resources = [] for f in files: p = Path(f) resources.append(Resource( uri=f"file://{f}", name=p.stem, description=f"FCPXML project: {p.name} ({format_duration(0)})", mimeType="application/xml", )) return resources @server.read_resource() async def read_resource(uri: str) -> str: """Read an FCPXML file and return a summary.""" filepath = str(uri).replace("file://", "") try: filepath = _validate_filepath(filepath, ('.fcpxml', '.fcpxmld')) except (ValueError, FileNotFoundError) as e: return str(e) project, tl = _parse_project(filepath) if not tl: return f"No timelines found in {filepath}" return f"""FCPXML Project: {tl.name} Duration: {format_duration(tl.duration.seconds)} Resolution: {tl.width}x{tl.height} @ {tl.frame_rate}fps Clips: {tl.total_clips} Markers: {len(tl.markers)} Cuts/min: {tl.cuts_per_minute:.1f} Path: {filepath}""" # ============================================================================ # MCP PROMPTS — Pre-built workflows # ============================================================================ @server.list_prompts() async def list_prompts() -> list[Prompt]: return [ Prompt( name="qc-check", description="Run a full quality control check on your timeline — flash frames, gaps, duplicates, and health score", arguments=[ PromptArgument(name="filepath", description="Path to FCPXML file", required=True), ], ), Prompt( name="youtube-chapters", description="Extract chapter markers formatted for YouTube descriptions", arguments=[ PromptArgument(name="filepath", description="Path to FCPXML file", required=True), ], ), Prompt( name="rough-cut", description="Guided rough cut generation — choose keywords, pacing, and duration", arguments=[ PromptArgument(name="filepath", description="Path to source FCPXML with clips", required=True), PromptArgument(name="duration", description="Target duration (e.g., '3m', '90s')", required=True), ], ), Prompt( name="timeline-summary", description="Quick overview of a timeline — stats, pacing, and potential issues", arguments=[ PromptArgument(name="filepath", description="Path to FCPXML file", required=True), ], ), Prompt( name="cleanup", description="Find and fix common timeline issues — flash frames, gaps, and duplicates", arguments=[ PromptArgument(name="filepath", description="Path to FCPXML file", required=True), ], ), ] @server.get_prompt() async def get_prompt(name: str, arguments: dict[str, str] | None = None) -> GetPromptResult: args = arguments or {} filepath = args.get("filepath", "") if name == "qc-check": return GetPromptResult( description="Full QC check on timeline", messages=[PromptMessage( role="user", content=TextContent( type="text", text=f"""Run a complete quality control check on my timeline. File: {filepath} Please: 1. Use `validate_timeline` to get the health score 2. Use `detect_flash_frames` to find any ultra-short clips 3. Use `detect_gaps` to find unintentional gaps 4. Use `detect_duplicates` to find repeated source clips 5. Summarize all issues and recommend fixes If there are critical issues, offer to fix them automatically with `fix_flash_frames` and `fill_gaps`.""" ), )], ) elif name == "youtube-chapters": return GetPromptResult( description="Export YouTube chapter markers", messages=[PromptMessage( role="user", content=TextContent( type="text", text=f"""Extract chapter markers from my timeline and format them for YouTube. File: {filepath} Please: 1. Use `list_markers` with format "youtube" to get chapter timestamps 2. Format the output so I can copy-paste directly into a YouTube description 3. If there are no chapter markers, suggest good chapter points based on the timeline structure using `analyze_pacing`""" ), )], ) elif name == "rough-cut": duration = args.get("duration", "3m") return GetPromptResult( description="Guided rough cut generation", messages=[PromptMessage( role="user", content=TextContent( type="text", text=f"""Help me create a rough cut from my source clips. File: {filepath} Target duration: {duration} Please: 1. Use `list_library_clips` to show me what clips are available 2. Use `list_keywords` to show me the tags I can filter by 3. Suggest a structure (segments, pacing) based on what's available 4. Generate the rough cut with `auto_rough_cut` using my preferences 5. Show me a summary of what was created""" ), )], ) elif name == "timeline-summary": return GetPromptResult( description="Quick timeline overview", messages=[PromptMessage( role="user", content=TextContent( type="text", text=f"""Give me a quick overview of my timeline. File: {filepath} Please: 1. Use `analyze_timeline` for stats (duration, resolution, clip count) 2. Use `analyze_pacing` for pacing metrics and suggestions 3. Use `list_keywords` to show what tags are in use 4. Use `list_markers` to show any markers 5. Give me a brief assessment of the edit""" ), )], ) elif name == "cleanup": return GetPromptResult( description="Find and fix timeline issues", messages=[PromptMessage( role="user", content=TextContent( type="text", text=f"""Help me clean up my timeline by finding and fixing common issues. File: {filepath} Please: 1. Use `validate_timeline` to get the health score 2. If there are flash frames, use `fix_flash_frames` to remove them 3. If there are gaps, use `fill_gaps` to close them 4. Report what was fixed and the new health score""" ), )], ) raise ValueError(f"Unknown prompt: {name}") # ============================================================================ # TOOL CATALOG & DISPATCH # ============================================================================ @server.list_tools() async def list_tools() -> list: """Concatenate every category module's Tool() schemas.""" tools = [] for module in _CATEGORY_MODULES: tools.extend(module.TOOLS) return tools TOOL_HANDLERS = {} for _module in _CATEGORY_MODULES: TOOL_HANDLERS.update(_module.HANDLERS) del _module @server.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) -> Sequence[TextContent]: handler = TOOL_HANDLERS.get(name) if not handler: return _text_result(f"Unknown tool: {name}") try: return await handler(arguments) except _NoTimelineError: return _no_timeline() except FileNotFoundError as e: return _text_result(f"File not found: {e}") except ValueError as e: return _text_result(f"Validation error: {e}") except Exception as e: return _text_result(f"Error: {type(e).__name__}") # ============================================================================ # MAIN # ============================================================================ async def main(): async with stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream, server.create_initialization_options()) def main_sync(): """Synchronous entry point for use as a console script.""" import asyncio asyncio.run(main()) if __name__ == "__main__": main_sync()