main_server.py•4.68 kB
"""
AgentKnowledgeMCP Main Server - FastMCP Server Composition
Modern server composition using FastMCP mounting architecture for modular design.
"""
import asyncio
from pathlib import Path
from fastmcp import FastMCP
# Import our existing modules for initialization
from src.config.config import load_config
from src.utils.security import init_security
from src.elasticsearch.elasticsearch_client import init_elasticsearch
from src.elasticsearch.elasticsearch_setup import auto_setup_elasticsearch
from src.confirmation.confirmation import initialize_confirmation_manager
# Import individual server modules for mounting
from src.admin.admin_server import app as admin_server_app
from src.elasticsearch.elasticsearch_server import app as elasticsearch_server_app
from src.prompts.prompt_server import app as prompt_server_app
# Import middleware
from src.middleware.confirmation_middleware import ConfirmationMiddleware
# Load configuration and initialize components
CONFIG = load_config()
init_security(CONFIG["security"]["allowed_base_directory"])
# Initialize confirmation manager
confirmation_manager = initialize_confirmation_manager(CONFIG)
print(f"✅ Confirmation system initialized (enabled: {CONFIG.get('confirmation', {}).get('enabled', True)})")
# Auto-setup Elasticsearch if needed
print("🔍 Checking Elasticsearch configuration...")
config_path = Path(__file__).parent / "config.json"
setup_result = auto_setup_elasticsearch(config_path, CONFIG)
if setup_result["status"] == "setup_completed":
# Reload config after setup
CONFIG = load_config()
print("✅ Elasticsearch auto-setup completed")
elif setup_result["status"] == "already_configured":
print("✅ Elasticsearch already configured")
elif setup_result["status"] == "setup_failed":
print(f"⚠️ Elasticsearch auto-setup failed: {setup_result.get('error', 'Unknown error')}")
print("📝 You can manually setup using the 'setup_elasticsearch' tool")
init_elasticsearch(CONFIG)
# Create main FastMCP server
app = FastMCP(
name=CONFIG["server"]["name"],
version=CONFIG["server"]["version"],
instructions="🏗️ AgentKnowledgeMCP - Modern FastMCP server with modular composition architecture for knowledge management, Elasticsearch operations, file management, and system administration"
)
# ================================
# MIDDLEWARE CONFIGURATION
# ================================
print("🔒 Adding confirmation middleware...")
# Add confirmation middleware to main server
app.add_middleware(ConfirmationMiddleware())
print("✅ Confirmation middleware added successfully!")
# ================================
# SERVER COMPOSITION - MOUNTING
# ================================
print("🏗️ Mounting individual servers into main server...")
# Mount Elasticsearch server with 'es' prefix
# This provides: es_search, es_index_document, es_create_index, etc.
app.mount(elasticsearch_server_app)
# Mount Administrative operations server with 'admin' prefix
# This provides: admin_get_config, admin_update_config, admin_server_status, etc.
app.mount(admin_server_app)
# Mount Prompt server for AgentKnowledgeMCP guidance
# This provides: usage_guide, help_request (prompts for LLM assistance)
app.mount(prompt_server_app)
print("🎉 Server composition completed successfully!")
# ================================
# BACKWARD COMPATIBILITY ALIASES
# ================================
# Add core tools without prefix for backward compatibility using static import
def cli_main():
"""CLI entry point for main FastMCP server."""
print("🚀 Starting AgentKnowledgeMCP Main FastMCP Server...")
print(f"📊 Server: {CONFIG['server']['name']}")
print(f"🔧 Version: {CONFIG['server']['version']}")
print("🌟 Architecture: Modern FastMCP with Server Mounting")
print()
print("📋 Available Servers (Mounted):")
print(" 🔍 Elasticsearch Server (es_*) - Document search, indexing, and management")
print(" └─ Tools: search, index_document, create_index, get_document, delete_document, list_indices, delete_index")
print(" ⚙️ Admin Server (admin_*) - Configuration and system management")
print(" └─ Tools: get_config, update_config, server_status, server_upgrade, setup_elasticsearch, elasticsearch_status, validate_config, reset_config, reload_config")
print(" 📝 Prompt Server - AgentKnowledgeMCP guidance and help")
print(" └─ Prompts: usage_guide, copilot_instructions")
print()
print("🔗 Compatibility: All tools also available without prefixes")
print()
# Start the FastMCP app (sync)
app.run()
if __name__ == "__main__":
cli_main()