integration/tools/mcp-bridge.py

#!/usr/bin/env python3
"""
MCP stdio-to-TCP bridge for BlueprintMCP.

Reads JSON-RPC messages from stdin, forwards them to the BlueprintMCP TCP server,
and writes responses to stdout. If the editor isn't running, returns valid MCP
error responses instead of crashing.
"""

import sys
import json
import socket
import time

HOST = "localhost"
PORT = 9847
CONNECT_TIMEOUT = 2
READ_TIMEOUT = 120

sock = None


def connect():
    """Try to connect to the editor. Returns True on success."""
    global sock
    if sock is not None:
        return True
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(CONNECT_TIMEOUT)
        s.connect((HOST, PORT))
        s.settimeout(READ_TIMEOUT)
        sock = s
        return True
    except (ConnectionRefusedError, socket.timeout, OSError):
        return False


def disconnect():
    global sock
    if sock is not None:
        try:
            sock.close()
        except Exception:
            pass
        sock = None


def send_and_receive(message):
    """Send a JSON-RPC message to the editor and return the response."""
    data = json.dumps(message) + "\n"
    sock.sendall(data.encode())

    result = b""
    while True:
        chunk = sock.recv(65536)
        if not chunk:
            raise ConnectionError("Connection closed")
        result += chunk
        try:
            return json.loads(result)
        except json.JSONDecodeError:
            continue


def editor_down_response(msg):
    """Return a valid MCP error response indicating the editor is down."""
    msg_id = msg.get("id", 0)
    method = msg.get("method", "")

    if method == "initialize":
        return {
            "jsonrpc": "2.0",
            "id": msg_id,
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {}},
                "serverInfo": {"name": "blueprint-mcp", "version": "1.0.0"},
                "_notice": "Unreal Editor is not running. No tools are available. Start the editor and restart Claude Code.",
            },
        }

    if method == "notifications/initialized":
        return None  # No response needed for notifications

    if method == "tools/list":
        return {
            "jsonrpc": "2.0",
            "id": msg_id,
            "result": {
                "tools": [{
                    "name": "editor_not_running",
                    "description": "Unreal Editor is not running. Start the editor and restart Claude Code to get BlueprintMCP tools.",
                    "inputSchema": {"type": "object", "properties": {}},
                }],
            },
        }

    if method == "tools/call":
        return {
            "jsonrpc": "2.0",
            "id": msg_id,
            "result": {
                "content": [
                    {"type": "text", "text": json.dumps({"error": "Unreal Editor is down."})}
                ],
                "isError": True,
            },
        }

    return {
        "jsonrpc": "2.0",
        "id": msg_id,
        "error": {"code": -32000, "message": "Unreal Editor is down."},
    }


def handle_message(msg):
    """Handle one JSON-RPC message. Try the editor first, fall back to offline responses."""
    method = msg.get("method", "")

    # Notifications don't get responses
    if "id" not in msg:
        if connect():
            try:
                sock.sendall((json.dumps(msg) + "\n").encode())
            except Exception:
                disconnect()
        return None

    # Try connecting (or reconnecting) to the editor
    if not connect():
        return editor_down_response(msg)

    try:
        return send_and_receive(msg)
    except Exception:
        disconnect()
        # Retry once in case the connection was stale
        if connect():
            try:
                return send_and_receive(msg)
            except Exception:
                disconnect()
        return editor_down_response(msg)


def main():
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            msg = json.loads(line)
        except json.JSONDecodeError:
            continue

        response = handle_message(msg)
        if response is not None:
            sys.stdout.write(json.dumps(response) + "\n")
            sys.stdout.flush()


if __name__ == "__main__":
    main()