""" mcp_server.py — Servidor MCP para RPG Maker XP / Pokémon Essentials v21.1 Protocolo: JSON-RPC 2.0 sobre stdio Solo expone lo que Antigravity NO puede hacer nativamente: - Leer/escribir .rxdata (Marshal binario de Ruby) - Escribir PBS con validación de formato - Templates de eventos con estructura RGSS garantizada - Backup automático y undo Lo que Antigravity ya hace (y este servidor NO reimplementa): - Leer archivos de texto (PBS/*.txt, Plugins/*.rb) - Buscar en código fuente - Entender el codebase con contexto de 1M tokens """ import json import os import sys import traceback from pathlib import Path ROOT = Path(__file__).parent sys.path.insert(0, str(ROOT / "tools")) sys.path.insert(0, str(ROOT / "bridge")) from rxdata_tools import RxDataTools from rxdata_bridge import RxDataError # ============================================================================= # CONFIGURACIÓN # ============================================================================= def _load_config() -> dict: if os.environ.get("PROJECT_PATH"): return { "project_path": os.environ["PROJECT_PATH"], "ruby_path": os.environ.get("RUBY_PATH", "ruby"), } cfg_file = ROOT / "config.json" if cfg_file.exists(): return json.loads(cfg_file.read_text(encoding="utf-8")) return {} _cfg = _load_config() PROJECT_PATH = _cfg.get("project_path", "") RUBY_PATH = _cfg.get("ruby_path", "ruby") _tools: RxDataTools | None = None def _get() -> RxDataTools: global _tools if not PROJECT_PATH: raise ValueError( "PROJECT_PATH no configurado. " "Define la variable de entorno o crea config.json." ) if _tools is None: _tools = RxDataTools(PROJECT_PATH, RUBY_PATH) return _tools # ============================================================================= # SCHEMA HELPERS # ============================================================================= def _s(props: dict, req: list = None) -> dict: return {"type": "object", "properties": props, "required": req or []} def _str(d): return {"type": "string", "description": d} def _int(d): return {"type": "integer", "description": d} def _bool(d):return {"type": "boolean", "description": d} def _obj(d): return {"type": "object", "description": d} def _arr(d): return {"type": "array", "description": d} # ============================================================================= # TOOLS — solo las imprescindibles # ============================================================================= TOOLS = { # ── MAPAS — lectura ─────────────────────────────────────────────────────── # Nota: Antigravity lee PBS/*.txt directamente. # Estos son los únicos reads necesarios porque .rxdata es binario. "list_maps": { "description": ( "Lista todos los mapas del proyecto con ID y nombre. " "Usar para identificar el map_id antes de cualquier operación." ), "inputSchema": _s({}), "fn": lambda p, t: t.list_maps(), }, "get_map_info": { "description": ( "Devuelve dimensiones, tileset y lista de eventos de un mapa. " "Usar SIEMPRE antes de añadir eventos para conocer el estado actual." ), "inputSchema": _s({"map_id": _int("ID del mapa")}, ["map_id"]), "fn": lambda p, t: t.get_map_info(p["map_id"]), }, "get_map_section": { "description": ( "Devuelve los eventos dentro de un rectángulo de coordenadas. " "Útil cuando el usuario señala una zona concreta del mapa." ), "inputSchema": _s({ "map_id": _int("ID del mapa"), "x1": _int("X inicial"), "y1": _int("Y inicial"), "x2": _int("X final"), "y2": _int("Y final"), }, ["map_id", "x1", "y1", "x2", "y2"]), "fn": lambda p, t: t.get_map_section( p["map_id"], p["x1"], p["y1"], p["x2"], p["y2"] ), }, "find_free_positions": { "description": ( "Devuelve posiciones del mapa sin eventos. " "Usar para sugerir dónde colocar nuevos NPCs u objetos." ), "inputSchema": _s({ "map_id": _int("ID del mapa"), "count": _int("Número de posiciones a devolver (default 10)"), }, ["map_id"]), "fn": lambda p, t: t.find_free_positions(p["map_id"], p.get("count", 10)), }, # ── MAPAS — escritura ───────────────────────────────────────────────────── "create_map": { "description": "Crea un mapa vacío y lo registra en MapInfos.rxdata.", "inputSchema": _s({ "name": _str("Nombre del mapa"), "width": _int("Ancho en tiles"), "height": _int("Alto en tiles"), "tileset_id": _int("ID del tileset (default 1)"), }, ["name", "width", "height"]), "fn": lambda p, t: t.create_map( p["name"], p["width"], p["height"], p.get("tileset_id", 1) ), }, # ── EVENTOS — templates ─────────────────────────────────────────────────── "add_npc": { "description": ( "Añade un NPC con diálogo. " "Lee PBS/pokemon.txt y Plugins/ nativamente para el contexto; " "usa esta tool solo para crear el evento." ), "inputSchema": _s({ "map_id": _int("ID del mapa"), "name": _str("Nombre del evento, ej: NPC_ARQUEOLOGA"), "x": _int("Coordenada X"), "y": _int("Coordenada Y"), "graphic": _str("Nombre del charset sin extensión, ej: NPC 01"), "dialogue": _arr("Lista de líneas del diálogo principal"), "second_dialogue": _arr("Líneas para segunda conversación (opcional)"), "direction": _int("Dirección: 2=abajo 4=izq 6=der 8=arriba"), }, ["map_id", "name", "x", "y", "graphic", "dialogue"]), "fn": lambda p, t: t.add_npc( p["map_id"], p["name"], p["x"], p["y"], p["graphic"], p["dialogue"], p.get("second_dialogue"), p.get("direction", 2), ), }, "add_item_ball": { "description": "Añade un objeto recogible en el suelo. Desaparece tras recogerlo.", "inputSchema": _s({ "map_id": _int("ID del mapa"), "name": _str("Nombre del evento"), "x": _int("Coordenada X"), "y": _int("Coordenada Y"), "item": _str("Nombre interno del objeto, ej: POTION"), }, ["map_id", "name", "x", "y", "item"]), "fn": lambda p, t: t.add_item_ball( p["map_id"], p["name"], p["x"], p["y"], p["item"] ), }, "add_mart": { "description": ( "Añade un NPC vendedor de tienda Pokémon. " "Lee PBS/items.txt para confirmar nombres internos antes de llamar." ), "inputSchema": _s({ "map_id": _int("ID del mapa"), "name": _str("Nombre del evento"), "x": _int("Coordenada X"), "y": _int("Coordenada Y"), "graphic": _str("Nombre del charset"), "items": _arr("Lista de nombres internos de objetos a vender"), "intro": _arr("Líneas de introducción (opcional)"), }, ["map_id", "name", "x", "y", "graphic", "items"]), "fn": lambda p, t: t.add_mart( p["map_id"], p["name"], p["x"], p["y"], p["graphic"], p["items"], p.get("intro"), ), }, "add_warp": { "description": "Añade un warp que transfiere al jugador a otro mapa.", "inputSchema": _s({ "map_id": _int("ID del mapa origen"), "name": _str("Nombre del evento"), "x": _int("X en mapa origen"), "y": _int("Y en mapa origen"), "dest_map_id": _int("ID del mapa destino"), "dest_x": _int("X en mapa destino"), "dest_y": _int("Y en mapa destino"), "direction": _int("Dirección al llegar (default 2=abajo)"), }, ["map_id", "name", "x", "y", "dest_map_id", "dest_x", "dest_y"]), "fn": lambda p, t: t.add_warp( p["map_id"], p["name"], p["x"], p["y"], p["dest_map_id"], p["dest_x"], p["dest_y"], p.get("direction", 2), ), }, "add_script_event": { "description": ( "Añade un evento que ejecuta código Ruby de PE v21.1. " "IMPORTANTE: lee Plugins/ nativamente para confirmar que el método " "existe antes de llamar a esta tool." ), "inputSchema": _s({ "map_id": _int("ID del mapa"), "name": _str("Nombre del evento"), "x": _int("Coordenada X"), "y": _int("Coordenada Y"), "ruby_code": _str("Código Ruby, ej: pbAddPokemon(:PIKACHU, 5)"), "trigger": _int("0=acción 1=tocar 3=auto 4=paralelo"), "graphic": _str("Charset (opcional, vacío=invisible)"), "one_time": _bool("True = ejecutar solo una vez"), }, ["map_id", "name", "x", "y", "ruby_code"]), "fn": lambda p, t: t.add_script_event( p["map_id"], p["name"], p["x"], p["y"], p["ruby_code"], p.get("trigger", 0), p.get("graphic", ""), p.get("one_time", False), ), }, "add_multipage_event": { "description": ( "Añade un evento con múltiples páginas, donde cada página ejecuta su propio script o usa diferentes self-switches y gráficos. " "Usar para eventos que cambian de fase (ej. Antes de batalla, Durante, Después)." ), "inputSchema": _s({ "map_id": _int("ID del mapa"), "name": _str("Nombre del evento"), "x": _int("Coordenada X"), "y": _int("Coordenada Y"), "pages_config": _arr("Lista de diccionarios con la configuración de las páginas. Permite: ruby_code, graphic, trigger, through, direction, condition (ej. {'self_switch':'A'})."), }, ["map_id", "name", "x", "y", "pages_config"]), "fn": lambda p, t: t.add_multipage_event( p["map_id"], p["name"], p["x"], p["y"], p["pages_config"] ), }, "add_raw_event": { "description": ( "Añade un evento desde un dict completo (uso avanzado). " "Usar cuando los templates no cubren el caso necesario." ), "inputSchema": _s({ "map_id": _int("ID del mapa"), "event_data": _obj("Estructura completa del evento en formato RGSS"), }, ["map_id", "event_data"]), "fn": lambda p, t: t.add_raw_event(p["map_id"], p["event_data"]), }, # ── EVENTOS — edición ───────────────────────────────────────────────────── "update_dialogue": { "description": "Reemplaza el diálogo de un NPC existente.", "inputSchema": _s({ "map_id": _int("ID del mapa"), "event_id": _int("ID del evento"), "new_dialogue": _arr("Nueva lista de líneas de texto"), "page": _int("Índice de página (default 0)"), }, ["map_id", "event_id", "new_dialogue"]), "fn": lambda p, t: t.update_dialogue( p["map_id"], p["event_id"], p["new_dialogue"], p.get("page", 0) ), }, "edit_event_page": { "description": "Modifica campos concretos de una página de evento existente.", "inputSchema": _s({ "map_id": _int("ID del mapa"), "event_id": _int("ID del evento"), "page": _int("Índice de página"), "changes": _obj("Campos a modificar, ej: {trigger: 1, graphic: {...}}"), }, ["map_id", "event_id", "page", "changes"]), "fn": lambda p, t: t.edit_event_page( p["map_id"], p["event_id"], p["page"], p["changes"] ), }, "delete_event": { "description": "Elimina un evento de un mapa.", "inputSchema": _s({ "map_id": _int("ID del mapa"), "event_id": _int("ID del evento a eliminar"), }, ["map_id", "event_id"]), "fn": lambda p, t: t.delete_event(p["map_id"], p["event_id"]), }, # ── PBS — solo escritura con validación ─────────────────────────────────── # Antigravity lee PBS/*.txt directamente con su contexto nativo. # Estas tools solo escriben, validando el formato estrictamente. "write_pokemon": { "description": ( "Escribe o actualiza una especie en PBS/pokemon.txt con validación. " "Lee el archivo PBS directamente antes de llamar para conocer el estado actual. " "Campos: Name, Type1, Type2, BaseStats (6 nums), GenderRate, GrowthRate, " "BaseEXP, CatchRate, Happiness, Abilities, HiddenAbility, Moves, " "EggMoves, Evolutions, Height, Weight, Kind, Pokedex, etc." ), "inputSchema": _s({ "species": _str("Nombre interno en mayúsculas, ej: PIKACHU"), "fields": _obj("Campos a escribir o actualizar"), }, ["species", "fields"]), "fn": lambda p, t: t.write_pokemon(p["species"], p["fields"]), }, "write_move": { "description": ( "Escribe o actualiza un movimiento en PBS/moves.txt con validación. " "Campos: Name, Type, Category (Physical/Special/Status), " "Power, Accuracy, PP, Target, Priority, FunctionCode, etc." ), "inputSchema": _s({ "move": _str("Nombre interno en mayúsculas, ej: THUNDERBOLT"), "fields": _obj("Campos a escribir o actualizar"), }, ["move", "fields"]), "fn": lambda p, t: t.write_move(p["move"], p["fields"]), }, "write_item": { "description": ( "Escribe o actualiza un objeto en PBS/items.txt con validación. " "Campos: Name, NamePlural, Pocket, Price, Description, " "UseFromBag, UseInBattle, BattleUse, Flags, etc." ), "inputSchema": _s({ "item": _str("Nombre interno en mayúsculas, ej: POTION"), "fields": _obj("Campos a escribir o actualizar"), }, ["item", "fields"]), "fn": lambda p, t: t.write_item(p["item"], p["fields"]), }, # ── UTILIDADES ──────────────────────────────────────────────────────────── "undo": { "description": ( "Deshace la última escritura de rxdata restaurando el backup automático. " "Usar si algo salió mal tras una operación de escritura." ), "inputSchema": _s({}), "fn": lambda p, t: t.undo(), }, } # ============================================================================= # BUCLE MCP — JSON-RPC 2.0 sobre stdio # ============================================================================= def _ok(rid, result): return {"jsonrpc": "2.0", "id": rid, "result": result} def _err(rid, code, msg): return {"jsonrpc": "2.0", "id": rid, "error": {"code": code, "message": msg}} def _handle(req: dict) -> dict | None: method = req.get("method", "") rid = req.get("id") if method == "notifications/initialized": return None if method == "initialize": return _ok(rid, { "protocolVersion": "2024-11-05", "serverInfo": {"name": "rmxp-agent", "version": "2.0.0"}, "capabilities": {"tools": {}}, }) if method == "tools/list": return _ok(rid, {"tools": [ {"name": k, "description": v["description"], "inputSchema": v["inputSchema"]} for k, v in TOOLS.items() ]}) if method == "tools/call": name = req.get("params", {}).get("name", "") args = req.get("params", {}).get("arguments") or {} if name not in TOOLS: return _err(rid, -32601, f"Tool desconocida: '{name}'") try: result = TOOLS[name]["fn"](args, _get()) return _ok(rid, { "content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}] }) except (ValueError, RxDataError) as e: return _ok(rid, {"content": [{"type": "text", "text": f"Error: {e}"}], "isError": True}) except Exception as e: return _ok(rid, {"content": [ {"type": "text", "text": f"Error inesperado: {e}\n{traceback.format_exc()}"} ], "isError": True}) return _err(rid, -32601, f"Método desconocido: '{method}'") def main(): sys.stdout.reconfigure(encoding="utf-8") sys.stderr.reconfigure(encoding="utf-8") sys.stdout = open(sys.stdout.fileno(), mode="w", encoding="utf-8", buffering=1) print(f"[rmxp-agent v2] PROJECT_PATH={PROJECT_PATH or '(no configurado)'}", file=sys.stderr) for raw in sys.stdin: line = raw.strip() if not line: continue try: req = json.loads(line) except json.JSONDecodeError: print(json.dumps(_err(None, -32700, "JSON inválido")), flush=True) continue resp = _handle(req) if resp is not None: print(json.dumps(resp, ensure_ascii=False), flush=True) if __name__ == "__main__": main()