"""
Servidor MCP personalizado con herramientas útiles.
"""
import asyncio
import json
import logging
import sys
from typing import Any, Dict, List, Optional
import click
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.server.stdio import stdio_server
from mcp.types import TextContent, Tool
from .tools import get_available_tools
# Configurar logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class NotificationOptions:
"""Clase simple para opciones de notificación."""
def __init__(self, tools_changed: bool = False):
self.tools_changed = tools_changed
class MCPCustomServer:
"""Servidor MCP personalizado con herramientas útiles."""
def __init__(self, name: str = "mcp-custom-tools", version: str = "0.1.0"):
self.name = name
self.version = version
self.server = Server(name)
self.tools = get_available_tools()
self._setup_handlers()
def _setup_handlers(self) -> None:
"""Configurar los handlers del servidor MCP."""
@self.server.list_tools()
async def handle_list_tools() -> List[Tool]:
"""Listar todas las herramientas disponibles."""
logger.info("Listando herramientas disponibles")
return [
Tool(
name=tool_name,
description=tool_config["description"],
inputSchema=tool_config.get("inputSchema", {}),
)
for tool_name, tool_config in self.tools.items()
]
@self.server.call_tool()
async def handle_call_tool(
name: str, arguments: Optional[Dict[str, Any]] = None
) -> List[TextContent]:
"""Ejecutar una herramienta específica."""
logger.info(f"Ejecutando herramienta: {name} con argumentos: {arguments}")
if name not in self.tools:
raise ValueError(f"Herramienta desconocida: {name}")
tool_config = self.tools[name]
handler = tool_config["handler"]
try:
# Ejecutar el handler de la herramienta
if asyncio.iscoroutinefunction(handler):
result = await handler(arguments or {})
else:
result = handler(arguments or {})
# Asegurar que el resultado sea una cadena
if isinstance(result, dict):
result = json.dumps(result, indent=2, ensure_ascii=False)
elif not isinstance(result, str):
result = str(result)
return [TextContent(type="text", text=result)]
except Exception as e:
logger.error(f"Error ejecutando {name}: {str(e)}")
error_msg = f"Error ejecutando {name}: {str(e)}"
return [TextContent(type="text", text=error_msg)]
async def run(self) -> None:
"""Ejecutar el servidor MCP."""
logger.info(f"Configurando servidor con {len(self.tools)} herramientas")
try:
# Ejecutar el servidor con stdio
async with stdio_server() as (read_stream, write_stream):
logger.info("Servidor MCP iniciado y esperando conexiones...")
await self.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name=self.name,
server_version=self.version,
capabilities=self.server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
except Exception as e:
logger.error(f"Error en el método run(): {e}")
logger.error("Detalles del error:", exc_info=True)
raise
@click.command()
@click.option(
"--log-level",
default="INFO",
help="Nivel de logging (DEBUG, INFO, WARNING, ERROR)",
)
@click.option(
"--name",
default="mcp-custom-tools",
help="Nombre del servidor MCP",
)
@click.option(
"--version",
default="0.1.0",
help="Versión del servidor",
)
def cli_main(log_level: str, name: str, version: str) -> None:
"""Comando CLI para el servidor MCP."""
run_server(log_level, name, version)
def main() -> None:
"""Punto de entrada para el script mcp-server."""
# Valores por defecto para el script
run_server("INFO", "mcp-custom-tools", "0.1.0")
def run_server(log_level: str, name: str, version: str) -> None:
"""Ejecutar el servidor MCP con la configuración especificada."""
# Configurar el nivel de logging
logging.getLogger().setLevel(getattr(logging, log_level.upper()))
logger.info(f"Iniciando servidor MCP: {name} v{version}")
# Crear y ejecutar el servidor
server = MCPCustomServer(name=name, version=version)
try:
asyncio.run(server.run())
except KeyboardInterrupt:
logger.info("Servidor detenido por el usuario")
except Exception as e:
logger.error(f"Error ejecutando el servidor: {e}")
logger.error("Traceback completo:", exc_info=True)
sys.exit(1)
if __name__ == "__main__":
cli_main()
