Skip to main content
Glama

MCP Inspector Server

by ACNet-AI
server.py8.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()

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/ACNet-AI/mcp-inspector-server'

If you have feedback or need assistance with the MCP directory API, please join our Discord server