#!/usr/bin/env python3 """ Cliente debug para el MCP server via stdio. Arranca el servidor como subprocess y te deja mandarle JSON-RPC interactivamente. Uso: python3 debug_client.py [proyecto_dir] Ejemplo: python3 debug_client.py /Users/jordandiaz/webs-locales/keepsailing.es Comandos especiales: init - Manda initialize + initialized automaticamente tools - Lista las tools disponibles call - Modo interactivo para llamar una tool prompts - Lista los prompts resources - Lista los resources quit/exit - Salir """ import subprocess import json import sys import os import threading import time # Colores para la terminal GREEN = "\033[92m" CYAN = "\033[96m" YELLOW = "\033[93m" RED = "\033[91m" DIM = "\033[2m" RESET = "\033[0m" class McpDebugClient: def __init__(self, project_dir=""): self.project_dir = project_dir self.msg_id = 0 self.proc = None self.responses = {} self._reader_thread = None def start(self): """Arranca el proceso MCP stdio.""" env = os.environ.copy() if self.project_dir: env["ACAI_PROJECT_DIR"] = self.project_dir mcp_file = os.path.join(self.project_dir, ".mcp.json") if os.path.exists(mcp_file): try: with open(mcp_file) as f: data = json.load(f) server = data.get("mcpServers", {}).get("acai-code", {}) server_env = server.get("env", {}) for key in ( "ACAI_WEBSITE", "ACAI_WEB_URL", "ACAI_API_WEB_URL", "ACAI_FORGE_HOST", "ACAI_TOKEN", "ACAI_TOKEN_HASH", "ACAI_PROJECT_DIR", ): if server_env.get(key): env.setdefault(key, server_env[key]) print( f"{GREEN}Leido .mcp.json:{RESET} " f"website={env.get('ACAI_WEBSITE')}, " f"web_url={env.get('ACAI_WEB_URL')}, " f"api_web_url={env.get('ACAI_API_WEB_URL')}" ) except Exception as e: print(f"{YELLOW}No se pudo leer .mcp.json: {e}{RESET}") # Fallback a .acai para sacar website, token y una URL basica si no hay .mcp.json acai_file = os.path.join(self.project_dir, ".acai") if os.path.exists(acai_file): try: with open(acai_file) as f: data = json.load(f) website = data.get("website") or data.get("domain") or "" web_url = data.get("web_url") or data.get("webUrl") or "" if not web_url and website: scheme = "https" if data.get("ssl", True) else "http" web_url = f"{scheme}://{website}" env.setdefault("ACAI_WEBSITE", website) env.setdefault("ACAI_WEB_URL", web_url) if data.get("token"): env.setdefault("ACAI_TOKEN", data["token"]) if data.get("tokenHash"): env.setdefault("ACAI_TOKEN_HASH", data["tokenHash"]) print(f"{GREEN}Leido .acai:{RESET} website={env.get('ACAI_WEBSITE')}, web_url={env.get('ACAI_WEB_URL')}") except Exception as e: print(f"{YELLOW}No se pudo leer .acai: {e}{RESET}") script_dir = os.path.dirname(os.path.abspath(__file__)) stdio_path = os.path.join(script_dir, "stdio.js") print(f"{DIM}Arrancando: node {stdio_path}{RESET}") self.proc = subprocess.Popen( ["node", stdio_path], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, cwd=script_dir, ) # Hilo para leer stderr (logs del server) self._reader_thread = threading.Thread(target=self._read_stderr, daemon=True) self._reader_thread.start() # Dar tiempo a que arranque time.sleep(1) def _read_stderr(self): """Lee stderr del server y lo muestra.""" for line in self.proc.stderr: text = line.decode("utf-8", errors="replace").rstrip() if text: print(f"{DIM}[server] {text}{RESET}") def send(self, obj): """Envia un mensaje JSON-RPC al server.""" raw = json.dumps(obj) msg = raw + "\n" print(f"\n{CYAN}>>> Enviando:{RESET}") print(json.dumps(obj, indent=2, ensure_ascii=False)) self.proc.stdin.write(msg.encode("utf-8")) self.proc.stdin.flush() def recv(self, timeout=10): """Lee una respuesta del server.""" self.proc.stdout.flush() line = self.proc.stdout.readline().decode("utf-8", errors="replace") if not line: return None obj = json.loads(line) print(f"\n{GREEN}<<< Respuesta:{RESET}") print(json.dumps(obj, indent=2, ensure_ascii=False)) return obj def request(self, method, params=None): """Envia un request y espera la respuesta.""" self.msg_id += 1 msg = {"jsonrpc": "2.0", "id": self.msg_id, "method": method} if params is not None: msg["params"] = params self.send(msg) return self.recv() def notify(self, method, params=None): """Envia una notificacion (sin id, no espera respuesta).""" msg = {"jsonrpc": "2.0", "method": method} if params is not None: msg["params"] = params self.send(msg) def initialize(self): """Handshake completo: initialize + initialized.""" print(f"\n{YELLOW}=== Inicializando ==={RESET}") resp = self.request("initialize", { "protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "debug_client", "version": "1.0.0"} }) self.notify("notifications/initialized") print(f"\n{GREEN}Inicializado OK{RESET}") if resp and "result" in resp: caps = resp["result"].get("capabilities", {}) if caps.get("tools"): print(f" Tools: disponibles") if caps.get("prompts"): print(f" Prompts: disponibles") if caps.get("resources"): print(f" Resources: disponibles") return resp def list_tools(self): """Lista las tools.""" resp = self.request("tools/list") if resp and "result" in resp: tools = resp["result"].get("tools", []) print(f"\n{YELLOW}=== {len(tools)} tools ==={RESET}") for t in tools: desc = t.get("description", "")[:60] print(f" {GREEN}{t['name']}{RESET} - {desc}") return resp def call_tool(self): """Modo interactivo para llamar una tool.""" name = input(f"{CYAN}Nombre de la tool: {RESET}").strip() if not name: return print(f"Argumentos como JSON (enter para {{}}):") args_str = input(f"{CYAN}> {RESET}").strip() args = json.loads(args_str) if args_str else {} return self.request("tools/call", {"name": name, "arguments": args}) def stop(self): if self.proc: self.proc.terminate() self.proc.wait() def main(): project_dir = sys.argv[1] if len(sys.argv) > 1 else "" print(f"{YELLOW}╔══════════════════════════════════╗{RESET}") print(f"{YELLOW}║ MCP Debug Client (stdio) ║{RESET}") print(f"{YELLOW}╚══════════════════════════════════╝{RESET}") print() print("Comandos: init, tools, call, prompts, resources, json, quit") print() client = McpDebugClient(project_dir) client.start() try: while True: try: cmd = input(f"\n{CYAN}mcp> {RESET}").strip().lower() except EOFError: break if not cmd: continue elif cmd in ("quit", "exit", "q"): break elif cmd == "init": client.initialize() elif cmd == "tools": client.list_tools() elif cmd == "call": client.call_tool() elif cmd == "prompts": client.request("prompts/list") elif cmd == "resources": client.request("resources/list") elif cmd == "json": print("Pega el JSON-RPC completo:") raw = input(f"{CYAN}> {RESET}").strip() try: obj = json.loads(raw) if "id" not in obj: client.notify(obj.get("method"), obj.get("params")) else: client.send(obj) client.recv() except json.JSONDecodeError as e: print(f"{RED}JSON invalido: {e}{RESET}") else: print(f"{YELLOW}Comando no reconocido. Usa: init, tools, call, prompts, resources, json, quit{RESET}") except KeyboardInterrupt: pass finally: print(f"\n{DIM}Cerrando...{RESET}") client.stop() if __name__ == "__main__": main()