#!/usr/bin/env python3 """Cliente de consulta del brain de Carlos Moya (MemoryBrain, MCP remoto). Solo lectura: search / read / list. No hay comandos de escritura porque el token de invitado no los permite — el servidor ni los expone. python3 brain.py check python3 brain.py search "output outcome" [--limit 8] [--folder "Permanent Notes"] python3 brain.py read "Permanent Notes/Titulo de la nota.md" python3 brain.py list ["Literature Notes"] python3 brain.py conform borrador.md --cited "Permanent Notes/Nota.md" --own "que pongo yo" El token se busca, en este orden: 1. la variable de entorno MEMORYBRAIN_TOKEN 2. el archivo que apunte MEMORYBRAIN_TOKEN_FILE 3. los sitios habituales (~/.config/memorybrain/token, /tmp/memorybrain-token, y cualquier *memorybrain*token*.txt subido a la sesion) """ import glob import json import os import sys import urllib.error import urllib.request ENDPOINT = os.environ.get("MEMORYBRAIN_ENDPOINT", "https://mcp.memorybrain.ai/mcp") HEALTH = ENDPOINT.rsplit("/mcp", 1)[0] + "/health" # Cloudflare corta a python-urllib con un error 1010 antes de llegar al servidor: # el user-agent propio no es cosmetica, es lo que hace que la peticion pase. UA = "memorybrain-guest/1.0" CANDIDATES = [ "~/.config/memorybrain/token", "/tmp/memorybrain-token", "/mnt/user-data/uploads/**/*memorybrain*token*", "/mnt/user-data/uploads/**/*brain*token*", "~/**/*memorybrain*token*.txt", ] HELP_TOKEN = """No encuentro el token del brain de Carlos. Si lo tienes a mano, la forma mas rapida es esta, y vale para el resto de la sesion: printf %s 'EL-TOKEN' > /tmp/memorybrain-token && chmod 600 /tmp/memorybrain-token Si prefieres no pegarlo en el chat, guardalo en un archivo llamado memorybrain-token.txt dentro de una carpeta conectada y pideme que lo suba a la sesion. Detalles y diagnostico: references/setup.md de esta skill.""" def token(): t = os.environ.get("MEMORYBRAIN_TOKEN", "").strip() if t: return t paths = [] explicit = os.environ.get("MEMORYBRAIN_TOKEN_FILE") if explicit: paths.append(os.path.expanduser(explicit)) for pat in CANDIDATES: paths.extend(sorted(glob.glob(os.path.expanduser(pat), recursive=True))) for p in paths: try: if os.path.isfile(p): value = open(p, encoding="utf-8").read().strip() if value: return value except OSError: continue sys.exit(HELP_TOKEN) def rpc(method, params=None, auth=True): headers = {"content-type": "application/json", "accept": "application/json", "user-agent": UA} if auth: headers["authorization"] = "Bearer " + token() body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params or {}}) req = urllib.request.Request(ENDPOINT, data=body.encode(), headers=headers) try: with urllib.request.urlopen(req, timeout=45) as res: raw = res.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: detail = e.read().decode("utf-8", "replace")[:300] if e.code == 401: sys.exit("El servidor rechaza el token (401). Puede estar mal copiado o revocado: " "pideselo de nuevo a Carlos. El brain no se ha caido, /health responde igual.") if e.code == 403: sys.exit("403 de Cloudflare. Suele ser el user-agent: no llames al endpoint con curl " "pelado ni con urllib sin cabecera, usa este script.") sys.exit(f"HTTP {e.code}: {detail}") except urllib.error.URLError as e: sys.exit(f"No hay salida a internet hacia {ENDPOINT} ({e.reason}). Esta skill necesita red " "y un terminal: en el chat de claude.ai sin shell no puede funcionar.") if raw.lstrip().startswith("event:"): # el servidor puede responder SSE raw = raw.split("data: ", 1)[-1] try: data = json.loads(raw) except json.JSONDecodeError: sys.exit(f"Respuesta que no es JSON: {raw[:200]}") if "error" in data: sys.exit(f"{data['error'].get('message')} (codigo {data['error'].get('code')})") return data.get("result", {}) def call(name, args): r = rpc("tools/call", {"name": name, "arguments": args}) text = "\n".join(c.get("text", "") for c in r.get("content", [])) if r.get("isError"): # No es un fallo del cliente: la nota no existe, la ruta es ambigua o esta # fuera del alcance del token. El texto del servidor ya dice cual de las tres. print(text) sys.exit(2) return text def flag(rest, name, default=None): if name in rest: i = rest.index(name) value = rest[i + 1] if i + 1 < len(rest) else default del rest[i:i + 2] return value return default def main(): if len(sys.argv) < 2: sys.exit(__doc__) cmd, rest = sys.argv[1], sys.argv[2:] if cmd == "check": req = urllib.request.Request(HEALTH, headers={"user-agent": UA}) try: with urllib.request.urlopen(req, timeout=20) as res: print("servidor:", res.read().decode()) except Exception as e: # noqa: BLE001 sys.exit(f"El servidor no responde en {HEALTH}: {e}") tools = [t["name"] for t in rpc("tools/list").get("tools", [])] print("token: valido · tools disponibles:", ", ".join(tools) or "ninguna") return if cmd == "search": limit = flag(rest, "--limit", "12") folder = flag(rest, "--folder") query = " ".join(rest).strip() if not query: sys.exit('uso: search "palabras clave" [--limit N] [--folder "Permanent Notes"]') args = {"query": query, "limit": int(limit)} if folder: args["folder"] = folder print(call("search_brain", args)) elif cmd == "read": path = " ".join(rest).strip() if not path: sys.exit('uso: read "ruta o titulo de la nota"') print(call("read_note", {"path": path})) elif cmd == "list": limit = flag(rest, "--limit") args = {} if rest: args["folder"] = " ".join(rest).strip() if limit: args["limit"] = int(limit) print(call("list_notes", args)) elif cmd == "conform": # El gate de conformidad, antes de entregar nada apoyado en el brain: comprueba # huecos sin rellenar, wikilinks que aqui fuera no resuelven y rutas citadas que # no existen, y devuelve el criterio de cada nota citada para contrastarlo. own = flag(rest, "--own") cited = [] while "--cited" in rest: value = flag(rest, "--cited") cited.extend(p.strip() for p in str(value or "").split(";") if p.strip()) text = flag(rest, "--text") source = " ".join(rest).strip() draft = text if draft is None and source: if source == "-": draft = sys.stdin.read() else: try: with open(os.path.expanduser(source)) as fh: draft = fh.read() except OSError as err: sys.exit(f"no pude leer {source}: {err}") if not draft or not draft.strip(): sys.exit('uso: conform borrador.md [--cited "ruta"] [--own "que aporto yo"]\n' ' conform - --text "..." (- lee de stdin)') args = {"draft": draft} if cited: args["cited"] = cited if own: args["own_contribution"] = own print(call("check_conformity", args)) else: sys.exit(f"comando desconocido: {cmd}\n{__doc__}") if __name__ == "__main__": main()