becomingone/spatial_engine/mcp_evennia_admin.py

#!/usr/bin/env python3
import sys
import os
import argparse
import json
from mcp.server.fastmcp import FastMCP

# -----------------------------------------------------------------------------
# EVENNIA NATIVE SETUP
# -----------------------------------------------------------------------------
def init_evennia():
    # Only bootstrap if not already initialized
    import django
    from django.apps import apps
    if not apps.ready:
        sys.path.append("/home/becomingone/kairos/spatial_engine")
        os.environ.setdefault("DJANGO_SETTINGS_MODULE", "server.conf.settings")
        
        # Change cwd to the MUD root to avoid permission errors when settings_default.py calls os.listdir()
        os.chdir("/home/becomingone/kairos/spatial_engine")
        django.setup()

# -----------------------------------------------------------------------------
# MCP SERVER DEFINITION
# -----------------------------------------------------------------------------
mcp = FastMCP("evennia_admin")

@mcp.tool()
def db_query_character(name: str) -> str:
    """Queries the underlying Evennia database for raw character data."""
    init_evennia()
    from evennia.utils.search import search_object
    results = search_object(name)
    if not results:
        return json.dumps({"error": f"Character '{name}' not found."})
    
    char = results[0]
    data = {
        "id": char.id,
        "key": char.key,
        "location": char.location.key if char.location else None,
        "attributes": {attr.key: str(attr.value) for attr in char.attributes.all()},
        "typeclass": char.typeclass_path
    }
    return json.dumps(data, indent=2)

@mcp.tool()
def db_query_room(name: str) -> str:
    """Queries the underlying Evennia database for raw room data and topology."""
    init_evennia()
    from evennia.utils.search import search_object
    results = search_object(name)
    if not results:
        return json.dumps({"error": f"Room '{name}' not found."})
    
    room = results[0]
    contents = [{"key": obj.key, "typeclass": obj.typeclass_path} for obj in room.contents]
    exits = [{"key": ex.key, "destination": ex.destination.key if ex.destination else None} for ex in room.exits]
    
    data = {
        "id": room.id,
        "key": room.key,
        "desc": room.db.desc,
        "contents": contents,
        "exits": exits
    }
    return json.dumps(data, indent=2)

@mcp.tool()
def execute_as(character_name: str, command: str) -> str:
    """Forces a character to execute a command seamlessly at the root level."""
    init_evennia()
    from evennia.utils.search import search_object
    results = search_object(character_name)
    if not results:
        return json.dumps({"error": f"Character '{character_name}' not found."})
    
    char = results[0]
    
    # Intercept output by overriding the msg method temporarily
    output_buffer = []
    original_msg = char.msg
    
    def intercepted_msg(self, text=None, **kwargs):
        if text:
            if isinstance(text, tuple):
                output_buffer.append(str(text[0]))
            else:
                output_buffer.append(str(text))
        original_msg(text=text, **kwargs)
        
    char.msg = intercepted_msg.__get__(char, type(char))
    
    try:
        char.execute_cmd(command)
        # Wait a tiny bit for any async command processing
        import time
        time.sleep(0.5)
    except Exception as e:
        return json.dumps({"error": str(e)})
    finally:
        # Restore original msg method
        char.msg = original_msg

    # Clean ANSI codes
    import re
    ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
    cleaned = [ansi_escape.sub('', msg) for msg in output_buffer]
    
    return json.dumps({"character": character_name, "command": command, "output": "\n".join(cleaned)})

@mcp.tool()
def broadcast_event(message: str) -> str:
    """Broadcasts a global message to all active sessions in the MUD."""
    init_evennia()
    from evennia.utils.create import create_message
    from evennia.server.sessionhandler import SESSIONS
    try:
        SESSIONS.announce_all(message)
        return json.dumps({"status": "success", "message": message})
    except Exception as e:
        return json.dumps({"error": str(e)})

# -----------------------------------------------------------------------------
# CLI / MCP ROUTER
# -----------------------------------------------------------------------------
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Evennia MCP Admin Server / CLI Suite")
    parser.add_argument("--mcp", action="store_true", help="Run as MCP stdio server")
    
    # CLI Commands
    parser.add_argument("--query-char", type=str, help="Query character data")
    parser.add_argument("--query-room", type=str, help="Query room data")
    parser.add_argument("--execute", nargs=2, metavar=('CHARACTER', 'COMMAND'), help="Execute command as character")
    parser.add_argument("--broadcast", type=str, help="Broadcast a global event")
    
    args = parser.parse_args()
    
    if args.mcp:
        # Run standard MCP stdio loop
        mcp.run()
    else:
        # Run in CLI mode
        if args.query_char:
            print(db_query_character(args.query_char))
        elif args.query_room:
            print(db_query_room(args.query_room))
        elif args.execute:
            print(execute_as(args.execute[0], args.execute[1]))
        elif args.broadcast:
            print(broadcast_event(args.broadcast))
        else:
            parser.print_help()