Spaces:
Sleeping
Sleeping
File size: 2,694 Bytes
8bf4d58 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 | """Base MCP server implementation."""
import logging
from typing import Any, Dict, List, Optional
import asyncio
# Try to import MCP SDK - adjust imports based on actual SDK version
try:
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
MCP_AVAILABLE = True
except ImportError:
# Fallback if MCP SDK structure is different
MCP_AVAILABLE = False
logger.warning("MCP SDK not available - MCP servers will not function")
logger = logging.getLogger(__name__)
class BaseMCPServer:
"""Base MCP server with common functionality."""
def __init__(self, name: str):
"""Initialize base MCP server."""
self.name = name
if not MCP_AVAILABLE:
logger.warning(f"MCP SDK not available - {name} server cannot be initialized")
self.server = None
self.tools: List[Any] = []
return
self.server = Server(name)
self.tools: List[Any] = []
self._setup_handlers()
def _setup_handlers(self):
"""Setup MCP server handlers."""
if not self.server:
return
@self.server.list_tools()
async def list_tools() -> List[Any]:
"""List available tools."""
return self.tools
@self.server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> List[Any]:
"""Call a tool by name."""
try:
result = await self._execute_tool(name, arguments)
return [TextContent(type="text", text=str(result))]
except Exception as e:
logger.error(f"Error executing tool {name}: {e}")
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def _execute_tool(self, name: str, arguments: Dict[str, Any]) -> Any:
"""Execute a tool - to be overridden by subclasses."""
raise NotImplementedError("Subclasses must implement _execute_tool")
def register_tool(self, tool: Any):
"""Register a tool with the server."""
self.tools.append(tool)
logger.info(f"Registered tool: {tool.name if hasattr(tool, 'name') else 'unknown'}")
async def run(self):
"""Run the MCP server."""
if not self.server or not MCP_AVAILABLE:
logger.error("Cannot run MCP server - SDK not available")
return
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
self.server.create_initialization_options(),
)
|