start.py•9.44 kB
#!/usr/bin/env python3
"""
AnyDocs-MCP Startup Script
Supports multiple startup modes:
- MCP server mode
- Web management interface mode
- Hybrid mode (start both MCP server and web interface simultaneously)
"""
import argparse
import asyncio
import logging
import os
import subprocess
import sys
from pathlib import Path
from typing import Optional
# Add project root directory to Python path
project_root = Path(__file__).parent
sys.path.insert(0, str(project_root / "src"))
try:
from anydocs_mcp.config import Settings, ConfigManager
from anydocs_mcp.database.manager import DatabaseManager
from anydocs_mcp.server import AnyDocsMCPServer
from anydocs_mcp.web.app import create_app
except ImportError as e:
print(f"Error: Unable to import required modules: {e}")
print("Please ensure all dependencies are installed: uv install --dev")
sys.exit(1)
def check_dependencies() -> bool:
"""Check if required dependencies are installed"""
required_packages = [
"mcp",
"fastapi",
"uvicorn",
"sqlalchemy",
"pydantic",
"yaml",
"httpx",
"bs4",
"markdown",
"python-multipart",
"jose",
"passlib",
"bcrypt"
]
missing_packages = []
for package in required_packages:
try:
__import__(package.replace("-", "_"))
except ImportError:
missing_packages.append(package)
if missing_packages:
print(f"Error: Missing the following dependencies: {', '.join(missing_packages)}")
print("Please run: uv install --dev")
return False
return True
def kill_port(port: int) -> bool:
"""Kill process occupying the specified port"""
try:
if os.name == 'nt': # Windows
result = subprocess.run(
["netstat", "-ano", "-p", "TCP"],
capture_output=True,
text=True
)
for line in result.stdout.split('\n'):
if f":{port}" in line and "LISTENING" in line:
pid = line.strip().split()[-1]
subprocess.run(["taskkill", "/F", "/PID", pid],
capture_output=True)
print(f"Killed process occupying port {port} (PID: {pid})")
return True
else: # Unix/Linux/macOS
result = subprocess.run(
["lsof", "-ti", f":{port}"],
capture_output=True,
text=True
)
if result.stdout.strip():
pid = result.stdout.strip()
subprocess.run(["kill", "-9", pid])
print(f"Killed process occupying port {port} (PID: {pid})")
return True
except Exception as e:
print(f"Error killing process on port {port}: {e}")
return False
async def init_database(config: Settings) -> bool:
"""Initialize database"""
try:
db_manager = DatabaseManager(
database_url=config.database.url,
echo=config.database.echo
)
db_manager.initialize_database()
print("Database initialization successful")
return True
except Exception as e:
print(f"Database initialization failed: {e}")
return False
async def start_mcp_server(config: Settings) -> None:
"""Start MCP server"""
try:
server = AnyDocsMCPServer(config)
await server.initialize()
print(f"MCP server started successfully")
print(f"Server configuration: {config.server.host}:{config.server.port}")
# Keep server running
await server.run()
except Exception as e:
logging.error(f"MCP server startup failed: {e}")
raise
async def start_web_interface(config: Settings) -> None:
"""Start web management interface"""
try:
import uvicorn
# Check and release port
web_port = 8080 # Default web port
kill_port(web_port)
app = create_app(config)
print(f"Starting web management interface...")
print(f"Access URL: http://{config.server.host}:{web_port}")
# Start web server
uvicorn_config = uvicorn.Config(
app,
host=config.server.host,
port=web_port,
log_level="info" if config.server.debug else "warning",
reload=config.server.debug
)
server = uvicorn.Server(uvicorn_config)
await server.serve()
except Exception as e:
logging.error(f"Web interface startup failed: {e}")
raise
async def start_hybrid_mode(config: Settings) -> None:
"""Hybrid mode: start both MCP server and web interface simultaneously"""
try:
# Create tasks
mcp_task = asyncio.create_task(start_mcp_server(config))
web_task = asyncio.create_task(start_web_interface(config))
print("Starting hybrid mode...")
print("MCP server and web management interface will run simultaneously")
# Wait for any task to complete or fail
done, pending = await asyncio.wait(
[mcp_task, web_task],
return_when=asyncio.FIRST_COMPLETED
)
# Cancel uncompleted tasks
for task in pending:
task.cancel()
except Exception as e:
logging.error(f"Hybrid mode startup failed: {e}")
raise
def setup_logging(debug: bool = False) -> None:
"""Setup logging configuration"""
level = logging.DEBUG if debug else logging.INFO
logging.basicConfig(
level=level,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout),
logging.FileHandler('anydocs_mcp.log')
]
)
def main():
"""Main function"""
parser = argparse.ArgumentParser(
description="AnyDocs-MCP Startup Script",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Usage examples:
python start.py --mode mcp # Start MCP server only
python start.py --mode web # Start web management interface only
python start.py --mode hybrid # Start both simultaneously
python start.py --config custom_config.yaml # Use custom configuration file
python start.py --debug # Enable debug mode
"""
)
parser.add_argument(
"--mode",
choices=["mcp", "web", "hybrid"],
default="hybrid",
help="Startup mode (default: hybrid)"
)
parser.add_argument(
"--config",
type=str,
default="config.yaml",
help="Configuration file path (default: config.yaml)"
)
parser.add_argument(
"--debug",
action="store_true",
help="Enable debug mode"
)
parser.add_argument(
"--no-deps-check",
action="store_true",
help="Skip dependency check"
)
parser.add_argument(
"--no-db-init",
action="store_true",
help="Skip database initialization"
)
parser.add_argument(
"--kill-ports",
action="store_true",
help="Kill processes occupying ports before startup"
)
args = parser.parse_args()
# Setup logging
setup_logging(args.debug)
print("=" * 50)
print("AnyDocs-MCP Startup Script")
print("=" * 50)
# Check dependencies
if not args.no_deps_check:
print("Checking dependencies...")
if not check_dependencies():
sys.exit(1)
print("✓ Dependency check passed")
# Load configuration
try:
config_manager = ConfigManager()
config = config_manager.load_config(args.config)
# Apply debug mode
if args.debug:
config.debug = True
config.log_level = "DEBUG"
print(f"✓ Configuration loaded successfully: {args.config}")
except Exception as e:
print(f"Error: Configuration loading failed: {e}")
sys.exit(1)
# Kill processes occupying ports
if args.kill_ports:
print("Checking and releasing ports...")
kill_port(config.server.port)
if hasattr(config.server, 'web_port'):
kill_port(config.server.web_port)
# Run main program
try:
asyncio.run(main_async(args, config))
except KeyboardInterrupt:
print("\nProgram interrupted by user")
except Exception as e:
logging.error(f"Program execution failed: {e}")
sys.exit(1)
async def main_async(args, config: Settings):
"""Async main function"""
# Initialize database
if not args.no_db_init:
print("Initializing database...")
if not await init_database(config):
sys.exit(1)
print("✓ Database initialization completed")
print(f"Startup mode: {args.mode}")
print("-" * 30)
# Start corresponding services based on mode
if args.mode == "mcp":
await start_mcp_server(config)
elif args.mode == "web":
await start_web_interface(config)
elif args.mode == "hybrid":
await start_hybrid_mode(config)
if __name__ == "__main__":
main()