591 lines
17 KiB
Python
591 lines
17 KiB
Python
#!/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_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_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_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", "<path to your .fcpxml file>")
|
|
|
|
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()
|