#!/usr/bin/env python3
"""
Godot MCP Server
Provides tools for Godot Engine development:
- Run GUT tests
- Check for parse/syntax errors
- Run the game
- Import assets
"""
import asyncio
import os
import subprocess
import json
from pathlib import Path
from typing import Any
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
# Environment variables for headless Godot
GODOT_ENV = {
**os.environ,
"QT_QPA_PLATFORM": "xcb",
"__GLX_VENDOR_LIBRARY_NAME": "mesa",
"LIBGL_ALWAYS_SOFTWARE": "0",
}
# Default timeout for operations (seconds)
DEFAULT_TIMEOUT = 60
TEST_TIMEOUT = 120
def find_godot_binary() -> str:
"""Find the Godot binary in PATH or common locations."""
# Check PATH first
result = subprocess.run(["which", "godot"], capture_output=True, text=True)
if result.returncode == 0:
return result.stdout.strip()
# Check common locations
common_paths = [
"/usr/bin/godot",
"/usr/local/bin/godot",
os.path.expanduser("~/.local/bin/godot"),
"/snap/bin/godot",
]
for path in common_paths:
if os.path.exists(path):
return path
return "godot" # Fallback, hope it's in PATH
def find_project_root(start_path: str) -> str | None:
"""Find the Godot project root by looking for project.godot."""
path = Path(start_path).resolve()
while path != path.parent:
if (path / "project.godot").exists():
return str(path)
path = path.parent
return None
async def run_godot_command(
args: list[str],
project_path: str,
timeout: int = DEFAULT_TIMEOUT,
headless: bool = True,
) -> tuple[int, str, str]:
"""Run a Godot command and return (returncode, stdout, stderr)."""
godot = find_godot_binary()
cmd = [godot]
if headless:
cmd.append("--headless")
cmd.extend(["--path", project_path])
cmd.extend(args)
try:
proc = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env=GODOT_ENV,
)
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=timeout
)
return proc.returncode or 0, stdout.decode(), stderr.decode()
except asyncio.TimeoutError:
proc.kill()
return -1, "", f"Command timed out after {timeout} seconds"
except Exception as e:
return -1, "", str(e)
def parse_gut_output(output: str) -> dict[str, Any]:
"""Parse GUT test output into structured results."""
results = {
"passed": 0,
"failed": 0,
"pending": 0,
"errors": [],
"failures": [],
"summary": "",
}
lines = output.split("\n")
for line in lines:
line_lower = line.lower()
if "passed" in line_lower and "failed" in line_lower:
# Summary line like "Passed: 10 Failed: 2"
results["summary"] = line.strip()
parts = line.split()
for i, part in enumerate(parts):
if part.lower() == "passed:" and i + 1 < len(parts):
try:
results["passed"] = int(parts[i + 1])
except ValueError:
pass
elif part.lower() == "failed:" and i + 1 < len(parts):
try:
results["failed"] = int(parts[i + 1])
except ValueError:
pass
elif "fail" in line_lower and ("test_" in line_lower or "assert" in line_lower):
results["failures"].append(line.strip())
elif "error" in line_lower:
results["errors"].append(line.strip())
return results
# Create the MCP server
server = Server("godot-mcp")
@server.list_tools()
async def list_tools() -> list[Tool]:
"""List available Godot tools."""
return [
Tool(
name="godot_run_tests",
description="Run GUT (Godot Unit Test) tests for a Godot project. Returns test results including pass/fail counts and any failures.",
inputSchema={
"type": "object",
"properties": {
"project_path": {
"type": "string",
"description": "Path to the Godot project directory (containing project.godot). If not provided, searches upward from current directory.",
},
"test_dir": {
"type": "string",
"description": "Directory containing tests (relative to project). Default: 'res://test/unit'",
"default": "res://test/unit",
},
"test_script": {
"type": "string",
"description": "Specific test script to run (e.g., 'test_combatant.gd'). If not provided, runs all tests.",
},
"timeout": {
"type": "integer",
"description": f"Timeout in seconds. Default: {TEST_TIMEOUT}",
"default": TEST_TIMEOUT,
},
},
"required": [],
},
),
Tool(
name="godot_check_errors",
description="Check a Godot project for parse errors and syntax issues without running the game.",
inputSchema={
"type": "object",
"properties": {
"project_path": {
"type": "string",
"description": "Path to the Godot project directory (containing project.godot).",
},
},
"required": [],
},
),
Tool(
name="godot_import",
description="Import/reimport assets in a Godot project. Useful after adding new files.",
inputSchema={
"type": "object",
"properties": {
"project_path": {
"type": "string",
"description": "Path to the Godot project directory.",
},
},
"required": [],
},
),
Tool(
name="godot_run_scene",
description="Run a specific scene in the Godot project (non-headless, will open a window).",
inputSchema={
"type": "object",
"properties": {
"project_path": {
"type": "string",
"description": "Path to the Godot project directory.",
},
"scene": {
"type": "string",
"description": "Scene path to run (e.g., 'res://src/GameLoop.tscn'). If not provided, runs main scene.",
},
"timeout": {
"type": "integer",
"description": f"Timeout in seconds. Default: {DEFAULT_TIMEOUT}",
"default": DEFAULT_TIMEOUT,
},
},
"required": [],
},
),
Tool(
name="godot_export",
description="Export a Godot project to a specific platform.",
inputSchema={
"type": "object",
"properties": {
"project_path": {
"type": "string",
"description": "Path to the Godot project directory.",
},
"preset": {
"type": "string",
"description": "Export preset name (as defined in export_presets.cfg).",
},
"output_path": {
"type": "string",
"description": "Output file path for the export.",
},
},
"required": ["preset", "output_path"],
},
),
]
@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Handle tool calls."""
# Get project path, default to current directory
project_path = arguments.get("project_path", os.getcwd())
# Try to find project root if not a valid project
if not (Path(project_path) / "project.godot").exists():
found_path = find_project_root(project_path)
if found_path:
project_path = found_path
else:
return [TextContent(
type="text",
text=f"Error: No Godot project found at {project_path} or in parent directories. "
f"Make sure project.godot exists."
)]
if name == "godot_run_tests":
return await handle_run_tests(project_path, arguments)
elif name == "godot_check_errors":
return await handle_check_errors(project_path, arguments)
elif name == "godot_import":
return await handle_import(project_path, arguments)
elif name == "godot_run_scene":
return await handle_run_scene(project_path, arguments)
elif name == "godot_export":
return await handle_export(project_path, arguments)
else:
return [TextContent(type="text", text=f"Unknown tool: {name}")]
async def handle_run_tests(project_path: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Run GUT tests."""
test_dir = arguments.get("test_dir", "res://test/unit")
test_script = arguments.get("test_script")
timeout = arguments.get("timeout", TEST_TIMEOUT)
# Build GUT command line arguments
args = ["-s", "addons/gut/gut_cmdln.gd"]
args.extend(["-gdir", test_dir])
args.append("-gexit")
if test_script:
args.extend(["-gtest", test_script])
returncode, stdout, stderr = await run_godot_command(
args, project_path, timeout=timeout
)
# Parse results
full_output = stdout + "\n" + stderr
results = parse_gut_output(full_output)
# Build response
status = "PASSED" if results["failed"] == 0 and returncode == 0 else "FAILED"
response = f"## Test Results: {status}\n\n"
response += f"**Passed:** {results['passed']}\n"
response += f"**Failed:** {results['failed']}\n"
if results["failures"]:
response += "\n### Failures:\n"
for failure in results["failures"][:10]: # Limit to 10
response += f"- {failure}\n"
if results["errors"]:
response += "\n### Errors:\n"
for error in results["errors"][:10]: # Limit to 10
response += f"- {error}\n"
if returncode != 0 and not results["failures"]:
response += f"\n### Raw Output:\n```\n{full_output[-2000:]}\n```"
return [TextContent(type="text", text=response)]
async def handle_check_errors(project_path: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Check for parse errors."""
# Run godot with --check-only to validate scripts
returncode, stdout, stderr = await run_godot_command(
["--check-only"], project_path, timeout=30
)
full_output = stdout + "\n" + stderr
# Look for errors
errors = []
for line in full_output.split("\n"):
line_lower = line.lower()
if "error" in line_lower or "parse error" in line_lower:
errors.append(line.strip())
if errors:
response = "## Parse Errors Found\n\n"
for error in errors:
response += f"- {error}\n"
else:
response = "## No Parse Errors\n\nAll scripts validated successfully."
return [TextContent(type="text", text=response)]
async def handle_import(project_path: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Import/reimport assets."""
returncode, stdout, stderr = await run_godot_command(
["--import"], project_path, timeout=60
)
if returncode == 0:
response = "## Import Complete\n\nAssets imported successfully."
else:
response = f"## Import Failed\n\n```\n{stderr[-1000:]}\n```"
return [TextContent(type="text", text=response)]
async def handle_run_scene(project_path: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Run a scene (non-headless)."""
scene = arguments.get("scene")
timeout = arguments.get("timeout", DEFAULT_TIMEOUT)
args = []
if scene:
args.extend(["--scene", scene])
returncode, stdout, stderr = await run_godot_command(
args, project_path, timeout=timeout, headless=False
)
if returncode == 0:
response = "## Scene Completed\n\nScene ran and exited normally."
elif returncode == -1:
response = f"## Scene Timed Out\n\nScene was terminated after {timeout} seconds."
else:
response = f"## Scene Error (code {returncode})\n\n```\n{stderr[-1000:]}\n```"
return [TextContent(type="text", text=response)]
async def handle_export(project_path: str, arguments: dict[str, Any]) -> list[TextContent]:
"""Export project."""
preset = arguments["preset"]
output_path = arguments["output_path"]
args = ["--export-release", preset, output_path]
returncode, stdout, stderr = await run_godot_command(
args, project_path, timeout=300 # Exports can take a while
)
if returncode == 0:
response = f"## Export Complete\n\nProject exported to: {output_path}"
else:
response = f"## Export Failed\n\n```\n{stderr[-1000:]}\n```"
return [TextContent(type="text", text=response)]
async def run_server():
"""Run the MCP server."""
async with stdio_server() as (read_stream, write_stream):
await server.run(read_stream, write_stream, server.create_initialization_options())
def main():
"""Entry point."""
asyncio.run(run_server())
if __name__ == "__main__":
main()
