264 lines
9.5 KiB
Python
264 lines
9.5 KiB
Python
#!/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()
|