server.py•8.5 kB
#!/usr/bin/env python3
"""
MCP server: mcp-inspector-server
Generated by MCP Factory
This server automatically discovers and registers components from:
- tools/ directory: @server.tool() functions
- resources/ directory: @server.resource() functions
- prompts/ directory: @server.prompt() functions
"""
import sys
from pathlib import Path
import yaml
from mcp_factory.server import ManagedServer
class ConfigurationError(Exception):
"""Configuration error"""
pass
def load_config():
"""Load and validate configuration"""
config_path = Path(__file__).parent / "config.yaml"
try:
with open(config_path, encoding="utf-8") as f:
config = yaml.safe_load(f)
# Validate required configuration
if not config or "server" not in config:
raise ConfigurationError("Invalid configuration: missing 'server' section")
server_config = config["server"]
if "name" not in server_config:
raise ConfigurationError("Invalid configuration: missing 'server.name'")
return config
except FileNotFoundError:
print(f"Error: Configuration file not found: {config_path}")
sys.exit(1)
except ConfigurationError as e:
print(f"Error: {e}")
sys.exit(1)
def create_server(config):
"""Create and configure the ManagedServer"""
server_config = config["server"]
management_config = config.get("management", {})
# Create server with basic configuration first
server_params = {
"name": server_config["name"],
"instructions": server_config.get("instructions", ""),
"expose_management_tools": management_config.get(
"expose_management_tools", True
),
}
# Load middleware from configuration
try:
from mcp_factory.middleware import load_middleware_from_config
middleware_instances = load_middleware_from_config(config)
if middleware_instances:
server_params["middleware"] = middleware_instances
print(f"✅ Loaded {len(middleware_instances)} middleware(s)")
except ImportError:
print("⚠️ Middleware support not available, skipping middleware loading")
except Exception as e:
print(f"❌ Failed to load middleware: {e}")
server = ManagedServer(**server_params)
# Configure external servers if present
if "mcpServers" in config:
try:
from mcp_factory.mounting import ServerRegistry
registry = ServerRegistry(server)
server_configs = registry.parse_external_servers_config(config)
registry.register_servers(server_configs)
lifespan = registry.create_lifespan({"auto_start": True})
if lifespan:
server._lifespan = lifespan
print("🔗 External servers configured")
except ImportError:
print("⚠️ ServerRegistry not available, skipping external servers")
except Exception as e:
print(f"⚠️ External servers configuration failed: {e}")
# Store full config for component registration
server._config = config
return server
def register_components(server):
"""Register project components using ComponentManager"""
try:
from mcp_factory.project.components import ComponentManager
project_path = Path(__file__).parent
# Check if auto-discovery is enabled
config = getattr(server, "_config", {})
auto_discovery_config = config.get("components", {}).get("auto_discovery", {})
if auto_discovery_config.get("enabled", False):
# Auto-discover components and update server config
print("🔍 Auto-discovering project components...")
discovered_components = ComponentManager.discover_project_components(
project_path
)
if discovered_components:
# Smart merge: preserve manual configuration, add newly discovered components
if "components" not in config:
config["components"] = {}
config_updated = False
for comp_type, discovered_items in discovered_components.items():
existing_items = config["components"].get(comp_type, [])
existing_modules = {item.get("module") for item in existing_items}
# Ensure all existing components have enabled attribute
for existing_item in existing_items:
if "enabled" not in existing_item:
existing_item["enabled"] = True
config_updated = True
# Only add non-existing components, deduplicate based on module path
for discovered_item in discovered_items:
if discovered_item.get("module") not in existing_modules:
# Newly discovered components are enabled by default
discovered_item["enabled"] = True
existing_items.append(discovered_item)
config_updated = True
config["components"][comp_type] = existing_items
server._config = config
# Write updated configuration back to disk - this is a required step for component registration
if config_updated:
config_path = project_path / "config.yaml"
try:
import yaml
with open(config_path, "w", encoding="utf-8") as f:
yaml.dump(
config,
f,
default_flow_style=False,
allow_unicode=True,
sort_keys=False,
)
print("💾 Configuration file updated")
except Exception as e:
print(f"⚠️ Configuration file write failed: {e}")
else:
print(
"💾 Configuration file does not need update (no new components)"
)
total_discovered = sum(
len(comps) for comps in discovered_components.values()
)
print(
f"📦 Discovered {total_discovered} components: {', '.join(f'{len(comps)} {name}' for name, comps in discovered_components.items())}"
)
else:
print("📂 No components found in project directories")
# Register components (both declared and discovered)
ComponentManager.register_components(server, project_path)
print("✅ Components registered successfully")
except ImportError:
print("⚠️ ComponentManager not available, skipping auto-discovery")
except Exception as e:
print(f"⚠️ Component registration failed: {e}")
print("💡 You can still add tools manually using @server.tool()")
def main():
"""Main entry point"""
print("🚀 Starting MCP Server...")
# Load configuration
config = load_config()
print(f"📋 Loaded configuration for: {config['server']['name']}")
# Create server
server = create_server(config)
print(f"🏗️ Created server: {server.name}")
# Register components
register_components(server)
# Start server
print(f"🎯 Instructions: {server.instructions}")
print("🔄 Starting server (Press Ctrl+C to stop)...")
try:
# Get transport configuration
transport_config = config.get("transport", {})
transport = transport_config.get("transport", "stdio")
if transport in ["sse", "streamable-http"]:
host = transport_config.get("host", "127.0.0.1")
port = transport_config.get("port", 8000)
print(f"🌐 Server will start on: {transport}://{host}:{port}")
server.run(transport=transport, host=host, port=port)
else:
print(f"🔗 Server will start with: {transport} transport")
server.run(transport=transport)
except KeyboardInterrupt:
print("\n👋 Server stopped by user")
except Exception as e:
print(f"❌ Server error: {e}")
sys.exit(1)
if __name__ == "__main__":
main()