#!/usr/bin/env python3
"""
MCP Tools Server v2.0 - Modular FastMCP Implementation
š PRODUCTION VERSION with modular architecture for better maintainability
A production-ready Model Context Protocol (MCP) server built with FastMCP Python framework,
featuring HTTP Streaming transport and modular tool architecture.
Architecture:
- Modular tool system with dynamic registration
- FastMCP Python framework with auto-generated schemas
- HTTP Streaming transport (MCP Guidelines compliant)
- Shell OAuth authentication for headless environments
- uvicorn ASGI server for production deployment
"""
import logging
import sys
from datetime import datetime
from pathlib import Path
# Add current directory to path for common module imports
sys.path.insert(0, str(Path(__file__).parent))
import uvicorn
from fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route
# Import our modular components
from src.config.settings import Config
from src.auth.oauth_shell import (
oauth_authorization_server,
oauth_authorization_server_mcp,
oauth_protected_resource,
register_client,
authorize,
token_endpoint
)
from src.tools import register_all_tools, get_all_tool_descriptions
# Configure logging
logging.basicConfig(
level=getattr(logging, Config.LOG_LEVEL),
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
# Initialize FastMCP server
mcp = FastMCP("MCP Tools Server")
def main():
"""Main function to start the MCP Tools server"""
logger.info("Starting MCP Tools Server (Modular Architecture)...")
logger.info(f"Server configuration: Port {Config.DEFAULT_PORT}, Log Level {Config.LOG_LEVEL}")
try:
# Register all tools using the modular system
logger.info("Registering tools using modular architecture...")
registration_result = register_all_tools(mcp)
# Log registration results
if registration_result["status"] == "completed":
logger.info(f"All tools registered successfully: {registration_result['successful_registrations']} modules")
else:
logger.warning(f"Tool registration completed with errors: {registration_result['failed_registrations']} failed")
# Get the FastAPI app from FastMCP
app = mcp.http_app()
# Health endpoint for container health checks
async def health_check(request: Request):
tool_descriptions = get_all_tool_descriptions()
return JSONResponse({
"status": "healthy",
"service": "mcp-tools",
"version": "2.0.1",
"architecture": "modular",
"transport": "FastMCP HTTP Streaming",
"timestamp": datetime.now().isoformat(),
"port": Config.DEFAULT_PORT,
"tools": {
"available": list(tool_descriptions.keys()),
"count": len(tool_descriptions),
"descriptions": tool_descriptions
},
"registration": {
"status": registration_result["status"],
"successful": registration_result["successful_registrations"],
"failed": registration_result["failed_registrations"],
"total": registration_result["total_modules"]
}
})
# Add shell authentication routes for Claude Code compatibility
auth_routes = [
Route('/health', health_check, methods=['GET']),
Route('/.well-known/oauth-authorization-server', oauth_authorization_server, methods=['GET']),
Route('/.well-known/oauth-authorization-server-mcp', oauth_authorization_server_mcp, methods=['GET']),
Route('/.well-known/oauth-protected-resource', oauth_protected_resource, methods=['GET']),
Route('/register', register_client, methods=['POST']),
Route('/auth', authorize, methods=['GET']),
Route('/token', token_endpoint, methods=['POST']),
]
for route in auth_routes:
app.routes.append(route)
logger.info("FastMCP HTTP Streaming server initialized with shell authentication")
logger.info(f"Available tools: {', '.join(get_all_tool_descriptions().keys())}")
logger.info("Authentication endpoints: /.well-known/oauth-authorization-server, /register, /auth, /token")
# Run with uvicorn
uvicorn.run(
app,
host="0.0.0.0",
port=Config.DEFAULT_PORT,
log_level=Config.LOG_LEVEL.lower()
)
except Exception as e:
logger.error(f"Failed to start MCP Tools server: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
