MCP Project Orchestrator
by sparesparrow
- 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()