MCP Project Orchestrator

MIT License
OverviewInspectSchema Related Servers Reviews Score
src
mcp_project_orchestrator
#!/usr/bin/env python3
"""
FastMCP server implementation for the MCP Project Orchestrator.

This module provides a lightweight MCP server that handles communication
with MCP clients like Claude Desktop, exposing project orchestration
capabilities through the Model Context Protocol.
"""
import os
import sys
import signal
import logging
import json
import time
from typing import Dict, Any, Optional, Callable

# Set up logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("mcp-project-orchestrator")

class FastMCP:
    """
    FastMCP server implementation for project orchestration.
    
    This class provides a lightweight MCP server that handles communication
    with MCP clients, exposing project orchestration capabilities through
    registered tools and appropriate error handling.
    """
    
    def __init__(self, name: str):
        """
        Initialize the MCP server with the given name.
        
        Args:
            name: Name of the MCP server
        """
        self.name = name
        self.config = {}
        self.tools = {}
        self.resources = {}
        
        # Set up signal handlers
        signal.signal(signal.SIGINT, self._handle_signal)
        signal.signal(signal.SIGTERM, self._handle_signal)
        
        logger.info(f"Initialized FastMCP server '{name}'")
        
        # Try to load configuration if it exists
        self._load_config()
    
    def _load_config(self) -> None:
        """Load configuration from project_orchestration.json if it exists."""
        try:
            config_path = os.path.join(os.getcwd(), "project_orchestration.json")
            if os.path.exists(config_path):
                with open(config_path, 'r') as f:
                    self.config = json.load(f)
                logger.info(f"Loaded configuration from {config_path}")
            else:
                logger.warning(f"Configuration file {config_path} not found, using defaults")
        except Exception as e:
            logger.error(f"Error loading configuration: {str(e)}")
    
    def tool(self, func: Optional[Callable] = None, 
             name: Optional[str] = None, 
             description: Optional[str] = None):
        """
        Decorator to register a function as an MCP tool.
        
        Args:
            func: The function to register
            name: Optional name for the tool (defaults to function name)
            description: Optional description of the tool
        
        Returns:
            The decorated function
        """
        def decorator(fn):
            tool_name = name or fn.__name__
            tool_desc = description or fn.__doc__ or f"Tool {tool_name}"
            
            self.tools[tool_name] = {
                "function": fn,
                "description": tool_desc,
                "parameters": {}  # In a real implementation, extract from function signature
            }
            
            logger.info(f"Registered tool '{tool_name}'")
            return fn
        
        if func is None:
            return decorator
        return decorator(func)
    
    def resource(self, name: str, content: Any) -> None:
        """
        Register a resource with the MCP server.
        
        Args:
            name: Name of the resource
            content: Content of the resource
        """
        self.resources[name] = content
        logger.info(f"Registered resource '{name}'")
    
    def _handle_signal(self, signum: int, frame) -> None:
        """
        Handle shutdown signals gracefully.
        
        Args:
            signum: Signal number
            frame: Current stack frame
        """
        signal_name = signal.Signals(signum).name
        logger.info(f"Received {signal_name}, shutting down...")
        sys.exit(0)
    
    def _handle_client_connection(self) -> None:
        """Handle client connections and MCP protocol messages."""
        logger.info("Handling client connection")
        # In a real implementation, this would use proper MCP protocol
        # to handle client requests, parse JSON messages, etc.
    
    def run(self) -> None:
        """Run the MCP server and handle client connections."""
        logger.info(f"FastMCP server '{self.name}' running with configuration: {self.config}")
        
        try:
            # In a real implementation, this would start a proper server
            # that listens for connections and handles MCP protocol
            logger.info("Server running on port 8080")
            logger.info("Press Ctrl+C to stop")
            
            # For demonstration, just keep the process running
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            logger.info("Keyboard interrupt received, shutting down...")
        except Exception as e:
            logger.error(f"Error running server: {str(e)}")
        finally:
            logger.info("Server shutting down")

def main():
    """Main entry point for the FastMCP server."""
    server = FastMCP("Project Orchestrator")
    server.run()

if __name__ == "__main__":
    main() 