server.py

Go to the documentation of this file.

"""
DaVinci Resolve MCP Server.
 
A clean, modern implementation of the Model Context Protocol server
for DaVinci Resolve integration.
"""
 
import logging
from typing import Any, Dict, List, Optional
 
from mcp.server import NotificationOptions, Server
from mcp.server.models import InitializationOptions
import mcp.server.stdio
import mcp.types as types
 
from .resolve_client import DaVinciResolveClient, DaVinciResolveError
from .tools import get_all_tools
from .resources import get_all_resources
 
 
logger = logging.getLogger(__name__)
 
 
class DaVinciMCPServer:
    """
    DaVinci Resolve MCP Server.
    
    Provides a clean interface between MCP clients and DaVinci Resolve
    through organized tools and resources.
    """
    
    def __init__(self) -> None:
        self.server = Server("davinci-resolve-mcp")
        self.resolve_client = DaVinciResolveClient()
        
        # Register handlers
        self._register_handlers()
        
        # Register tools and resources
        self._register_tools()
        self._register_resources()
    
    def _register_handlers(self) -> None:
        """Register MCP server handlers."""
        
        @self.server.list_tools()
        async def handle_list_tools() -> List[types.Tool]:
            """List available tools."""
            return get_all_tools()
        
        @self.server.call_tool()
        async def handle_call_tool(
            name: str, arguments: Optional[Dict[str, Any]] = None
        ) -> List[types.TextContent]:
            """Handle tool calls."""
            if arguments is None:
                arguments = {}
            
            try:
                # Ensure we're connected
                if not self.resolve_client.is_connected():
                    self.resolve_client.connect()
                
                # Call the appropriate tool
                result = await self._call_tool(name, arguments)
                
                return [types.TextContent(type="text", text=str(result))]
                
            except DaVinciResolveError as e:
                error_msg = f"DaVinci Resolve error: {e}"
                logger.error(error_msg)
                return [types.TextContent(type="text", text=error_msg)]
            except Exception as e:
                error_msg = f"Unexpected error: {e}"
                logger.error(error_msg)
                return [types.TextContent(type="text", text=error_msg)]
        
        @self.server.list_resources()
        async def handle_list_resources() -> List[types.Resource]:
            """List available resources."""
            return get_all_resources()
        
        @self.server.read_resource()
        async def handle_read_resource(uri: str) -> str:
            """Handle resource reads."""
            try:
                # Ensure we're connected
                if not self.resolve_client.is_connected():
                    self.resolve_client.connect()
                
                # Read the resource
                result = await self._read_resource(uri)
                return str(result)
                
            except DaVinciResolveError as e:
                error_msg = f"DaVinci Resolve error: {e}"
                logger.error(error_msg)
                return error_msg
            except Exception as e:
                error_msg = f"Unexpected error: {e}"
                logger.error(error_msg)
                return error_msg
    
    def _register_tools(self) -> None:
        """Register tool implementations."""
        # Tool implementations will be imported from the tools module
        pass
    
    def _register_resources(self) -> None:
        """Register resource implementations.""" 
        # Resource implementations will be imported from the resources module
        pass
    
    async def _call_tool(self, name: str, arguments: Dict[str, Any]) -> Any:
        """Call a specific tool."""
        # System tools
        if name == "get_version":
            return self.resolve_client.get_version()
        
        elif name == "get_current_page":
            return self.resolve_client.get_current_page()
        
        elif name == "switch_page":
            page = arguments.get("page", "")
            return self.resolve_client.switch_page(page)
        
        # Project tools
        elif name == "list_projects":
            return self.resolve_client.list_projects()
        
        elif name == "get_current_project":
            return self.resolve_client.get_current_project_name()
        
        elif name == "open_project":
            name_arg = arguments.get("name", "")
            result = self.resolve_client.open_project(name_arg)
            return f"Successfully opened project '{name_arg}'" if result else f"Failed to open project '{name_arg}'"
        
        elif name == "create_project":
            name_arg = arguments.get("name", "")
            result = self.resolve_client.create_project(name_arg)
            return f"Successfully created project '{name_arg}'" if result else f"Failed to create project '{name_arg}'"
        
        # Timeline tools
        elif name == "list_timelines":
            return self.resolve_client.list_timelines()
        
        elif name == "get_current_timeline":
            return self.resolve_client.get_current_timeline_name()
        
        elif name == "create_timeline":
            name_arg = arguments.get("name", "")
            result = self.resolve_client.create_timeline(name_arg)
            return f"Successfully created timeline '{name_arg}'" if result else f"Failed to create timeline '{name_arg}'"
        
        elif name == "switch_timeline":
            name_arg = arguments.get("name", "")
            result = self.resolve_client.switch_timeline(name_arg)
            return f"Successfully switched to timeline '{name_arg}'" if result else f"Failed to switch to timeline '{name_arg}'"
        
        # Media tools
        elif name == "list_media_clips":
            return self.resolve_client.list_media_clips()
        
        elif name == "import_media":
            file_path = arguments.get("file_path", "")
            result = self.resolve_client.import_media(file_path)
            return f"Successfully imported media '{file_path}'" if result else f"Failed to import media '{file_path}'"
        
        else:
            raise ValueError(f"Unknown tool: {name}")
    
    async def _read_resource(self, uri: str) -> Any:
        """Read a specific resource."""
        # System resources
        if uri == "resolve://version":
            return self.resolve_client.get_version()
        
        elif uri == "resolve://current-page":
            return self.resolve_client.get_current_page()
        
        # Project resources
        elif uri == "resolve://projects":
            return self.resolve_client.list_projects()
        
        elif uri == "resolve://current-project":
            name = self.resolve_client.get_current_project_name()
            return name if name else "No project open"
        
        # Timeline resources
        elif uri == "resolve://timelines":
            return self.resolve_client.list_timelines()
        
        elif uri == "resolve://current-timeline":
            name = self.resolve_client.get_current_timeline_name()
            return name if name else "No timeline active"
        
        # Media resources
        elif uri == "resolve://media-clips":
            return self.resolve_client.list_media_clips()
        
        else:
            raise ValueError(f"Unknown resource: {uri}")
    
    async def run(self) -> None:
        """Run the MCP server."""
        logger.info("Starting DaVinci Resolve MCP Server...")
        
        async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
            await self.server.run(
                read_stream,
                write_stream,
                InitializationOptions(
                    server_name="davinci-mcp-professional",
                    server_version="2.1.0",
                    capabilities=self.server.get_capabilities(
                        notification_options=NotificationOptions(),
                        experimental_capabilities={},
                    ),
                ),
            )