RCiBot
92623a1bc8
Track Unity source, scenes, editable assets, licenses, tools and QA evidence. Exclude generated caches, Windows builds and raw runtime logs.
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))
|