codebuddy.py•6.78 kB
#!/usr/bin/env python3
"""
Codebuddy MCP Server - Main application entry point.
A lightweight MCP server providing task scaffolding and memory for AI agents.
Follows Clean Code principles:
- Single Responsibility: Main module handles only server startup
- Dependency Injection: All dependencies are properly injected
- Separation of Concerns: Configuration, logging, and business logic are separate
"""
import argparse
import logging
import signal
import sys
from pathlib import Path
from fastmcp import FastMCP
from storage import TaskStorage, TaskStorageError
from tools import TaskService, setup_mcp_tools
from error_handling import (
error_handler,
health_monitor,
setup_default_health_checks,
ErrorType,
ErrorSeverity
)
# Global server instance for graceful shutdown
server_instance = None
def setup_logging(level: str = "INFO") -> None:
"""
Configure application logging.
Follows SRP: Handles only logging configuration.
"""
logging.basicConfig(
level=getattr(logging, level.upper(), logging.INFO),
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout)
]
)
# Reduce noise from external libraries
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
def setup_signal_handlers() -> None:
"""
Setup graceful shutdown signal handlers.
Handles SIGINT and SIGTERM for proper cleanup.
"""
def signal_handler(signum, frame):
"""Handle shutdown signals gracefully."""
logger = logging.getLogger(__name__)
logger.info(f"Received signal {signum}, shutting down gracefully...")
if server_instance:
# FastMCP handles its own cleanup
logger.info("Server shutdown complete")
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
def create_argument_parser() -> argparse.ArgumentParser:
"""
Create command-line argument parser.
Follows Open/Closed Principle: Easy to add new arguments.
"""
parser = argparse.ArgumentParser(
description="Codebuddy MCP Server - Task scaffolding and memory for AI agents",
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
parser.add_argument(
"--host",
type=str,
default="localhost",
help="Host address to bind the server to"
)
parser.add_argument(
"--port",
type=int,
default=8000,
help="Port number to bind the server to"
)
parser.add_argument(
"--data-file",
type=str,
default="data/tasks.jsonl",
help="Path to the JSONL storage file"
)
parser.add_argument(
"--log-level",
type=str,
default="INFO",
choices=["DEBUG", "INFO", "WARNING", "ERROR"],
help="Logging level"
)
return parser
def validate_configuration(args: argparse.Namespace) -> None:
"""
Validate server configuration.
Raises appropriate exceptions for invalid configurations.
"""
# Validate port range
if not (1 <= args.port <= 65535):
raise ValueError(f"Port must be between 1 and 65535, got {args.port}")
# Validate data file path
data_path = Path(args.data_file)
if not data_path.parent.exists():
try:
data_path.parent.mkdir(parents=True, exist_ok=True)
except OSError as e:
raise ValueError(f"Cannot create data directory {data_path.parent}: {e}")
# Check if we can write to the data directory
try:
test_file = data_path.parent / ".write_test"
test_file.touch()
test_file.unlink()
except OSError as e:
raise ValueError(f"Cannot write to data directory {data_path.parent}: {e}")
def create_server_components(data_file: str) -> tuple[FastMCP, TaskStorage, TaskService]:
"""
Create and wire together server components.
Follows Dependency Injection principle for testability.
Returns wired components for the server.
"""
logger = logging.getLogger(__name__)
try:
# Create storage layer
storage = TaskStorage(data_file)
logger.info(f"Initialized storage with {storage.get_total_count()} existing tasks")
# Setup health monitoring
setup_default_health_checks(storage)
# Create service layer
service = TaskService(storage)
# Create MCP server
mcp = FastMCP("Codebuddy")
setup_mcp_tools(mcp, service)
logger.info("Server components initialized successfully")
return mcp, storage, service
except TaskStorageError as e:
error_handler.handle_error(e, ErrorType.STORAGE, ErrorSeverity.CRITICAL)
raise
except Exception as e:
error_handler.handle_error(e, ErrorType.INTERNAL, ErrorSeverity.CRITICAL)
raise
def main() -> None:
"""
Main application entry point.
Handles configuration, initialization, and server startup.
"""
global server_instance
# Parse command line arguments
parser = create_argument_parser()
args = parser.parse_args()
# Setup logging
setup_logging(args.log_level)
logger = logging.getLogger(__name__)
try:
# Validate configuration
validate_configuration(args)
logger.info(f"Starting Codebuddy MCP Server on {args.host}:{args.port}")
logger.info(f"Data file: {args.data_file}")
# Setup signal handlers for graceful shutdown
setup_signal_handlers()
# Create server components
mcp, storage, service = create_server_components(args.data_file)
server_instance = mcp
# Log storage statistics
stats = storage.get_storage_stats()
logger.info(f"Storage stats: {stats}")
# Start the server
logger.info("Starting HTTP server...")
mcp.run(
transport="http",
host=args.host,
port=args.port
)
except KeyboardInterrupt:
logger.info("Received keyboard interrupt, shutting down...")
except ValueError as e:
logger.error(f"Configuration error: {e}")
sys.exit(1)
except TaskStorageError as e:
logger.error(f"Storage error: {e}")
sys.exit(1)
except Exception as e:
logger.error(f"Unexpected error: {e}", exc_info=True)
sys.exit(1)
finally:
logger.info("Codebuddy MCP Server stopped")
if __name__ == "__main__":
main()