main.pyโข7.5 kB
#!/usr/bin/env python3
"""
AnyDocs MCP Server Main Entry Point
Starts the AnyDocs MCP server with proper initialization and configuration.
"""
import asyncio
import sys
import signal
import argparse
from pathlib import Path
from typing import Optional
# Add src to path for development
src_path = Path(__file__).parent / "src"
if src_path.exists():
sys.path.insert(0, str(src_path))
from anydocs_mcp.config import init_config_manager, get_config
from anydocs_mcp.config.settings import AuthMethod
from anydocs_mcp.database import initialize_database_manager, get_database_manager
from anydocs_mcp.auth import AuthManager
from anydocs_mcp.server import AnyDocsMCPServer
from anydocs_mcp.utils import get_logger, setup_logging
logger = get_logger(__name__)
class ServerManager:
"""Manages the lifecycle of the AnyDocs MCP server."""
def __init__(self):
self.mcp_server: Optional[AnyDocsMCPServer] = None
self.config_manager = None
self.db_manager = None
self.auth_manager = None
self.shutdown_event = asyncio.Event()
async def initialize(self, config_file: Optional[str] = None):
"""Initialize all components.
Args:
config_file: Path to configuration file
"""
try:
logger.info("Initializing AnyDocs MCP Server")
# Initialize configuration
self.config_manager = init_config_manager(
config_file=config_file,
auto_reload=True
)
config = get_config()
logger.info('ๅฎๆด้
็ฝฎ็ปๆ', config=config.model_dump())
logger.info('่ฎค่ฏ้
็ฝฎ่ฏฆๆ
', auth_config=config.auth.model_dump())
# Setup logging
setup_logging(
level=config.logging.level,
log_file=config.logging.file if config.logging.file else None,
json_logs=config.logging.json_logs
)
logger.info("Configuration loaded", config_file=config_file)
# Initialize database
self.db_manager = initialize_database_manager(
database_url=config.database.url,
echo=config.database.echo
)
self.db_manager.initialize_database()
logger.info("Database initialized")
# Initialize authentication
self.auth_manager = AuthManager(
jwt_secret=config.auth.jwt_secret,
token_expire_minutes=config.auth.token_expire_minutes
)
logger.info("Authentication manager initialized")
# Initialize MCP server
self.mcp_server = AnyDocsMCPServer(config=config)
await self.mcp_server.initialize()
logger.info("MCP server initialized")
logger.info("AnyDocs MCP Server initialization complete")
except Exception as e:
logger.error("Failed to initialize server", error=str(e))
raise
async def start(self):
"""Start the MCP server."""
if not self.mcp_server:
raise RuntimeError("Server not initialized")
try:
logger.info("Starting AnyDocs MCP Server")
# Start MCP server
await self.mcp_server.run()
logger.info("AnyDocs MCP Server started successfully")
# Wait for shutdown signal
await self.shutdown_event.wait()
except Exception as e:
logger.error("Server error", error=str(e))
raise
async def stop(self):
"""Stop the server gracefully."""
logger.info("Stopping AnyDocs MCP Server")
try:
# MCP server will be stopped when the run task is cancelled
pass
if self.db_manager:
self.db_manager.close()
logger.info("AnyDocs MCP Server stopped")
except Exception as e:
logger.error("Error during shutdown", error=str(e))
finally:
self.shutdown_event.set()
def signal_handler(self, signum, frame):
"""Handle shutdown signals."""
logger.info("Received shutdown signal", signal=signum)
asyncio.create_task(self.stop())
async def run_server(config_file: Optional[str] = None):
"""Run the AnyDocs MCP server.
Args:
config_file: Path to configuration file
"""
server_manager = ServerManager()
# Setup signal handlers
for sig in [signal.SIGINT, signal.SIGTERM]:
signal.signal(sig, server_manager.signal_handler)
try:
await server_manager.initialize(config_file)
await server_manager.start()
except KeyboardInterrupt:
logger.info("Received keyboard interrupt")
except Exception as e:
logger.error("Server failed", error=str(e))
sys.exit(1)
finally:
await server_manager.stop()
def main():
"""Main entry point."""
parser = argparse.ArgumentParser(
description="AnyDocs MCP Server - Transform documentation into MCP-compatible server"
)
parser.add_argument(
"-c", "--config",
help="Path to configuration file",
type=str
)
parser.add_argument(
"--web",
help="Start web administration interface",
action="store_true"
)
parser.add_argument(
"--web-port",
help="Web interface port (default: 8000)",
type=int,
default=8000
)
parser.add_argument(
"--debug",
help="Enable debug mode",
action="store_true"
)
parser.add_argument(
"--mode",
help="Server mode: mcp, web, or hybrid (default: mcp)",
choices=["mcp", "web", "hybrid"],
default="mcp"
)
args = parser.parse_args()
if args.mode == "web" or args.web:
# Start web interface
import uvicorn
from anydocs_mcp.web import create_app
# Initialize components for web app
config_manager = init_config_manager(config_file=args.config)
config = get_config()
if args.debug:
config.server.debug = True
config.logging.level = "DEBUG"
setup_logging(
level=config.logging.level,
log_file=config.logging.file if config.logging.file else None,
json_logs=config.logging.json_logs
)
# Initialize database
db_manager = initialize_database_manager(
database_url=config.database.url,
echo=config.database.echo
)
db_manager.initialize_database()
# Create web app
app = create_app()
# Run web server
uvicorn.run(
app,
host="0.0.0.0",
port=args.web_port,
log_level="info" if not args.debug else "debug",
reload=args.debug
)
elif args.mode == "mcp":
# Run MCP server only
asyncio.run(run_server(args.config))
elif args.mode == "hybrid":
# Run both MCP and web (not implemented yet)
logger.warning("Hybrid mode not yet implemented, running MCP server only")
asyncio.run(run_server(args.config))
if __name__ == "__main__":
main()