OwusuBlessing
100 lines
3.5 KiB
Python
100 lines
3.5 KiB
Python
"""
|
|
Modular MCP Server Implementation
|
|
"""
|
|
from mcp.server.fastmcp import FastMCP
|
|
from typing import Optional, List
|
|
from .tools.tool_registry import ServerToolRegistry
|
|
from .prompts.prompt_registry import ServerPromptRegistry
|
|
from .resources.resource_registry import ServerResourceRegistry
|
|
|
|
|
|
class ModularMCPServer:
|
|
"""Modular MCP Server that automatically discovers and registers tools, prompts, and resources"""
|
|
|
|
def __init__(
|
|
self,
|
|
name: str,
|
|
host: str = "0.0.0.0",
|
|
port: int = 8050,
|
|
stateless_http: bool = True,
|
|
tools_directory: Optional[str] = None,
|
|
prompts_directory: Optional[str] = None,
|
|
resources_directory: Optional[str] = None
|
|
):
|
|
self.name = name
|
|
self.host = host
|
|
self.port = port
|
|
self.stateless_http = stateless_http
|
|
|
|
# Initialize registries
|
|
self.tool_registry = ServerToolRegistry(tools_directory)
|
|
self.prompt_registry = ServerPromptRegistry(prompts_directory)
|
|
self.resource_registry = ServerResourceRegistry(resources_directory)
|
|
|
|
# Create FastMCP server
|
|
self.mcp = FastMCP(
|
|
name=name,
|
|
host=host,
|
|
port=port,
|
|
stateless_http=stateless_http
|
|
)
|
|
|
|
self._initialized = False
|
|
|
|
async def initialize(self) -> None:
|
|
"""Initialize the server and register all components"""
|
|
if self._initialized:
|
|
return
|
|
|
|
print(f"Initializing {self.name} server...")
|
|
|
|
# Discover and register tools
|
|
print("Discovering tools...")
|
|
self.tool_registry.register_tools_with_server(self.mcp)
|
|
tool_count = len(self.tool_registry.get_all_tools())
|
|
print(f"Registered {tool_count} tools")
|
|
|
|
# Discover and register prompts
|
|
print("Discovering prompts...")
|
|
self.prompt_registry.register_prompts_with_server(self.mcp)
|
|
prompt_count = len(self.prompt_registry.get_all_prompts())
|
|
print(f"Registered {prompt_count} prompts")
|
|
|
|
# Discover and register resources
|
|
print("Discovering resources...")
|
|
self.resource_registry.register_resources_with_server(self.mcp)
|
|
resource_count = len(self.resource_registry.get_all_resources())
|
|
print(f"Registered {resource_count} resources")
|
|
|
|
self._initialized = True
|
|
print(f"Server initialization complete!")
|
|
|
|
def run(self, transport: str = "stdio") -> None:
|
|
"""Run the server with the specified transport"""
|
|
if not self._initialized:
|
|
import asyncio
|
|
asyncio.run(self.initialize())
|
|
|
|
print(f"Starting {self.name} server with {transport} transport...")
|
|
self.mcp.run(transport=transport)
|
|
|
|
def get_server_info(self) -> dict:
|
|
"""Get information about the server and its components"""
|
|
return {
|
|
"name": self.name,
|
|
"host": self.host,
|
|
"port": self.port,
|
|
"tools": {
|
|
"count": len(self.tool_registry.get_all_tools()),
|
|
"names": self.tool_registry.get_tool_names()
|
|
},
|
|
"prompts": {
|
|
"count": len(self.prompt_registry.get_all_prompts()),
|
|
"names": self.prompt_registry.get_prompt_names()
|
|
},
|
|
"resources": {
|
|
"count": len(self.resource_registry.get_all_resources()),
|
|
"uris": self.resource_registry.get_resource_uris()
|
|
}
|
|
}
|