89 lines
4.6 KiB
Python
89 lines
4.6 KiB
Python
|
"""Call the installed Unity MCP server through the standard MCP SDK.
|
||
|
|
|
||
|
|
Run with the Python environment belonging to the installed MCP for Unity server.
|
||
|
|
Scene mutations and builds stay inside the Unity MCP tools.
|
||
|
|
"""
|
||
|
|
import argparse
|
||
|
|
import asyncio
|
||
|
|
import json
|
||
|
|
import sys
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
from mcp import ClientSession, StdioServerParameters
|
||
|
|
from mcp.client.stdio import stdio_client
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[2]
|
||
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
|
|
||
|
|
|
||
|
|
def unpack(result):
|
||
|
|
data = getattr(result, "structuredContent", None)
|
||
|
|
if data is None:
|
||
|
|
texts = [x.text for x in result.content if x.type == "text"]
|
||
|
|
try:
|
||
|
|
data = json.loads("\n".join(texts))
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
data = {"text": "\n".join(texts)}
|
||
|
|
payload = data.get("result", data) if isinstance(data, dict) else data
|
||
|
|
failed = isinstance(payload, dict) and payload.get("success") is False
|
||
|
|
return {"isError": bool(result.isError or failed), "result": data}
|
||
|
|
|
||
|
|
|
||
|
|
async def run(args):
|
||
|
|
executable = Path.home() / ".local/share/unity-mcp/.venv/Scripts/mcp-for-unity.exe"
|
||
|
|
logs = ROOT / "Logs/MCP"
|
||
|
|
logs.mkdir(parents=True, exist_ok=True)
|
||
|
|
params = StdioServerParameters(command=str(executable), args=["--transport", "stdio", "--default-instance", ROOT.name])
|
||
|
|
with (logs / "server.log").open("w", encoding="utf-8") as errlog:
|
||
|
|
async with stdio_client(params, errlog=errlog) as (read, write):
|
||
|
|
async with ClientSession(read, write) as session:
|
||
|
|
await session.initialize()
|
||
|
|
project = await session.read_resource("mcpforunity://project/info")
|
||
|
|
project_text = "\n".join(x.text for x in project.contents if hasattr(x, "text"))
|
||
|
|
if str(ROOT).replace("\\", "/").lower() not in project_text.replace("\\\\", "/").replace("\\", "/").lower():
|
||
|
|
raise RuntimeError("MCP target is not the workspace: " + project_text)
|
||
|
|
if args.list:
|
||
|
|
available = await session.list_tools()
|
||
|
|
wanted = args.list.split(",")
|
||
|
|
out = [t.model_dump() for t in available.tools if t.name in wanted or args.list == "all"]
|
||
|
|
elif args.resource:
|
||
|
|
res = await session.read_resource(args.resource)
|
||
|
|
out = [x.text for x in res.contents if hasattr(x, "text")]
|
||
|
|
else:
|
||
|
|
commands = json.loads(Path(args.file).read_text(encoding="utf-8-sig"))
|
||
|
|
if isinstance(commands, dict):
|
||
|
|
commands = [commands]
|
||
|
|
out = []
|
||
|
|
jobs_path = logs / "jobs.json"
|
||
|
|
jobs = json.loads(jobs_path.read_text(encoding="utf-8")) if jobs_path.exists() else {}
|
||
|
|
for command in commands:
|
||
|
|
arguments = command.get("arguments", {})
|
||
|
|
job_reference = arguments.get("job_id", "")
|
||
|
|
if job_reference.startswith("$"):
|
||
|
|
arguments["job_id"] = jobs[job_reference[1:]]
|
||
|
|
result = unpack(await session.call_tool(command["tool"], arguments))
|
||
|
|
out.append(result)
|
||
|
|
print(json.dumps({"tool": command["tool"], **result}, ensure_ascii=False), flush=True)
|
||
|
|
payload = result["result"]
|
||
|
|
while isinstance(payload, dict) and isinstance(payload.get("result"), dict) and "data" not in payload:
|
||
|
|
payload = payload["result"]
|
||
|
|
job_id = payload.get("data", {}).get("job_id") if isinstance(payload, dict) and isinstance(payload.get("data"), dict) else None
|
||
|
|
if job_id:
|
||
|
|
category = "test" if command["tool"] in ("run_tests", "get_test_job") else "build" if command["tool"] == "manage_build" else "package"
|
||
|
|
jobs[category] = job_id
|
||
|
|
jobs_path.write_text(json.dumps(jobs), encoding="utf-8")
|
||
|
|
if result["isError"]:
|
||
|
|
break
|
||
|
|
(logs / "last-result.json").write_text(json.dumps(out, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
|
if args.list or args.resource:
|
||
|
|
print(json.dumps(out, ensure_ascii=False, indent=2), flush=True)
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
parser = argparse.ArgumentParser()
|
||
|
|
parser.add_argument("--list")
|
||
|
|
parser.add_argument("--resource")
|
||
|
|
parser.add_argument("--file")
|
||
|
|
args = parser.parse_args()
|
||
|
|
asyncio.run(asyncio.wait_for(run(args), timeout=58))
|