#!/usr/bin/env python3 """mcp-pool-manager.py — Gestor de conexiones MCP para swl-ses. Conecta a servidores MCP configurados en .claude/settings.json y expone sus herramientas para uso por agentes y scripts de swl-ses. Lee la configuracion de mcpServers en orden de precedencia: 1. .claude/settings.local.json 2. .claude/settings.json 3. mcp-servers.json (config independiente opcional) Uso: python scripts/mcp-pool-manager.py list-servers python scripts/mcp-pool-manager.py list-tools [servidor] python scripts/mcp-pool-manager.py ping [servidor] python scripts/mcp-pool-manager.py call-tool [json-args] Flags: --json Salida en formato JSON (default: texto legible) --cwd Directorio raiz del proyecto (default: .) """ from __future__ import annotations import argparse import asyncio import json import os import sys import time from pathlib import Path from typing import Any # --------------------------------------------------------------------------- # Dependencias — raw mcp SDK (disponible en entornos con Claude Code) # --------------------------------------------------------------------------- try: from mcp import ClientSession from mcp.client.stdio import stdio_client, StdioServerParameters from mcp.client.sse import sse_client HAS_MCP = True except ImportError: HAS_MCP = False # --------------------------------------------------------------------------- # Constantes # --------------------------------------------------------------------------- SETTINGS_CANDIDATES = [ '.claude/settings.local.json', '.claude/settings.json', 'mcp-servers.json', ] TIMEOUT_CONNECT_S = 10 # segundos maximo para conectar a un servidor TIMEOUT_CALL_S = 30 # segundos maximo para llamar una herramienta # --------------------------------------------------------------------------- # Carga de configuracion # --------------------------------------------------------------------------- def _cargar_config(cwd: Path, config_path: str | None = None) -> dict: """Carga la seccion mcpServers del primer archivo de config encontrado.""" candidates = [Path(config_path)] if config_path else [cwd / p for p in SETTINGS_CANDIDATES] for p in candidates: if p.exists(): try: data = json.loads(p.read_text(encoding='utf-8')) servers = data.get('mcpServers', {}) if servers: return servers except Exception: continue return {} def _get_server(servers: dict, nombre: str) -> dict: """Retorna la config de un servidor o termina con error.""" if nombre not in servers: sys.stderr.write(f'[mcp-pool] Servidor no encontrado: {nombre}\n') sys.stderr.write(f'[mcp-pool] Servidores disponibles: {", ".join(servers.keys())}\n') sys.exit(1) return servers[nombre] # --------------------------------------------------------------------------- # Fabrica de transporte # --------------------------------------------------------------------------- def _build_env(cfg: dict) -> dict | None: """Combina el entorno del proceso con las variables de entorno del servidor.""" extra = cfg.get('env') or {} if not extra: return None merged = dict(os.environ) merged.update(extra) return merged def _make_stdio_params(cfg: dict) -> StdioServerParameters: return StdioServerParameters( command=cfg['command'], args=cfg.get('args', []), env=_build_env(cfg), cwd=cfg.get('cwd'), ) # --------------------------------------------------------------------------- # Operaciones MCP (async helpers) # --------------------------------------------------------------------------- async def _list_tools_server(nombre: str, cfg: dict) -> dict: """Conecta a un servidor MCP y lista sus herramientas. Tolera errores.""" resultado: dict = {'server': nombre, 'tools': [], 'error': None, 'duration_ms': 0} t0 = time.time() try: if 'url' in cfg: transport = sse_client(cfg['url']) else: transport = stdio_client(_make_stdio_params(cfg)) async with transport as (read, write): async with ClientSession(read, write) as session: await asyncio.wait_for(session.initialize(), timeout=TIMEOUT_CONNECT_S) resp = await asyncio.wait_for(session.list_tools(), timeout=TIMEOUT_CONNECT_S) resultado['tools'] = [ { 'name': t.name, 'description': t.description or '', 'schema': t.inputSchema or {}, } for t in (resp.tools or []) ] except asyncio.TimeoutError: resultado['error'] = f'Timeout al conectar ({TIMEOUT_CONNECT_S}s)' except Exception as exc: resultado['error'] = str(exc) finally: resultado['duration_ms'] = int((time.time() - t0) * 1000) return resultado async def _call_tool_server(nombre: str, cfg: dict, tool: str, args: dict) -> dict: """Llama una herramienta en un servidor MCP.""" resultado: dict = {'server': nombre, 'tool': tool, 'result': None, 'is_error': False, 'error': None, 'duration_ms': 0} t0 = time.time() try: if 'url' in cfg: transport = sse_client(cfg['url']) else: transport = stdio_client(_make_stdio_params(cfg)) async with transport as (read, write): async with ClientSession(read, write) as session: await asyncio.wait_for(session.initialize(), timeout=TIMEOUT_CONNECT_S) resp = await asyncio.wait_for( session.call_tool(tool, args), timeout=TIMEOUT_CALL_S ) resultado['is_error'] = bool(getattr(resp, 'isError', False)) resultado['result'] = [ c.text if hasattr(c, 'text') else str(c) for c in (resp.content or []) ] except asyncio.TimeoutError: resultado['error'] = f'Timeout al llamar herramienta ({TIMEOUT_CALL_S}s)' except Exception as exc: resultado['error'] = str(exc) finally: resultado['duration_ms'] = int((time.time() - t0) * 1000) return resultado # --------------------------------------------------------------------------- # Subcomandos # --------------------------------------------------------------------------- async def cmd_list_servers(servers: dict, as_json: bool) -> None: out = sys.stdout.write if as_json: payload = [ {'name': n, 'transport': 'http' if 'url' in c else 'stdio', 'command': c.get('url') or f"{c.get('command', '?')} {' '.join(c.get('args', []))}"} for n, c in servers.items() ] out(json.dumps(payload, ensure_ascii=False, indent=2) + '\n') return out(f'Servidores MCP configurados ({len(servers)}):\n') out('-' * 60 + '\n') for nombre, cfg in servers.items(): if 'url' in cfg: detalle = f"HTTP {cfg['url']}" else: detalle = f"stdio {cfg.get('command', '?')} {' '.join(cfg.get('args', []))}" out(f' {nombre:<28} {detalle}\n') async def cmd_list_tools(servers: dict, server_filter: str | None, as_json: bool) -> None: out = sys.stdout.write targets = {server_filter: servers[server_filter]} if server_filter else servers tareas = [_list_tools_server(n, c) for n, c in targets.items()] resultados = await asyncio.gather(*tareas) if as_json: out(json.dumps(resultados, ensure_ascii=False, indent=2) + '\n') return for r in resultados: if r['error']: out(f'\n[{r["server"]}] ERROR: {r["error"]}\n') continue out(f'\n[{r["server"]}] {len(r["tools"])} herramientas ({r["duration_ms"]}ms):\n') for t in r['tools']: desc = t['description'] if len(desc) > 80: desc = desc[:77] + '...' out(f' - {t["name"]:<32} {desc}\n') async def cmd_ping(servers: dict, server_filter: str | None, as_json: bool) -> None: out = sys.stdout.write targets = {server_filter: servers[server_filter]} if server_filter else servers tareas = [_list_tools_server(n, c) for n, c in targets.items()] resultados = await asyncio.gather(*tareas) pings = [ { 'server': r['server'], 'estado': 'OK' if not r['error'] else 'ERROR', 'herramientas': len(r['tools']), 'duration_ms': r['duration_ms'], 'error': r['error'], } for r in resultados ] if as_json: out(json.dumps(pings, ensure_ascii=False, indent=2) + '\n') return out('\nPing a servidores MCP:\n') out('-' * 64 + '\n') for p in pings: estado = p['estado'] ms = p['duration_ms'] tools = p['herramientas'] sufijo = f' ({p["error"]})' if p['error'] else '' out(f' {p["server"]:<28} {estado:<6} {tools:>3} tools {ms:>5}ms{sufijo}\n') async def cmd_call_tool(servers: dict, server_name: str, tool: str, tool_args: dict, as_json: bool) -> None: out = sys.stdout.write cfg = _get_server(servers, server_name) r = await _call_tool_server(server_name, cfg, tool, tool_args) if as_json: out(json.dumps(r, ensure_ascii=False, indent=2) + '\n') return if r['error']: out(f'ERROR ({r["duration_ms"]}ms): {r["error"]}\n') return if r['is_error']: out(f'Tool error ({r["duration_ms"]}ms):\n') else: out(f'Resultado ({r["duration_ms"]}ms):\n') for linea in (r['result'] or []): out(f'{linea}\n') # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main() -> None: if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8', errors='replace') if not HAS_MCP: sys.stderr.write( '[mcp-pool] ERROR: la libreria mcp no esta instalada.\n' '[mcp-pool] Ejecuta: pip install mcp\n' ) sys.exit(1) parser = argparse.ArgumentParser( description='Gestor de conexiones MCP para swl-ses' ) parser.add_argument('--json', action='store_true', help='Salida en formato JSON') parser.add_argument('--config', default=None, help='Ruta al archivo de config MCP') parser.add_argument('--cwd', default='.', help='Directorio raiz del proyecto') sub = parser.add_subparsers(dest='cmd') sub.add_parser('list-servers', help='Lista servidores MCP configurados') p_tools = sub.add_parser('list-tools', help='Lista herramientas de servidor(es)') p_tools.add_argument('servidor', nargs='?', help='Nombre del servidor (opcional)') p_ping = sub.add_parser('ping', help='Verifica conectividad de servidor(es)') p_ping.add_argument('servidor', nargs='?', help='Nombre del servidor (opcional)') p_call = sub.add_parser('call-tool', help='Llama una herramienta MCP') p_call.add_argument('servidor', help='Nombre del servidor') p_call.add_argument('herramienta', help='Nombre de la herramienta') p_call.add_argument('args', nargs='?', default='{}', help='Argumentos JSON de la herramienta (default: {})') args = parser.parse_args() cwd = Path(args.cwd).resolve() servers = _cargar_config(cwd, args.config) if not servers: sys.stderr.write( '[mcp-pool] No se encontraron servidores MCP configurados.\n' '[mcp-pool] Verifica la clave "mcpServers" en .claude/settings.json\n' ) sys.exit(1) as_json = bool(getattr(args, 'json', False)) if args.cmd == 'list-servers': asyncio.run(cmd_list_servers(servers, as_json)) elif args.cmd == 'list-tools': servidor = getattr(args, 'servidor', None) if servidor: _get_server(servers, servidor) # valida que exista asyncio.run(cmd_list_tools(servers, servidor, as_json)) elif args.cmd == 'ping': servidor = getattr(args, 'servidor', None) if servidor: _get_server(servers, servidor) asyncio.run(cmd_ping(servers, servidor, as_json)) elif args.cmd == 'call-tool': try: tool_args = json.loads(args.args) except json.JSONDecodeError: sys.stderr.write(f'[mcp-pool] Argumentos JSON invalidos: {args.args}\n') sys.exit(1) asyncio.run(cmd_call_tool(servers, args.servidor, args.herramienta, tool_args, as_json)) else: parser.print_help() if __name__ == '__main__': main()