178 lines
11 KiB
Python
178 lines
11 KiB
Python
"""Read-only, project-bound UE MCP probe using its own short-lived session."""
|
|
import asyncio
|
|
import base64
|
|
import json
|
|
import logging
|
|
import re
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from mcp import ClientSession
|
|
from mcp.client.streamable_http import streamablehttp_client
|
|
from mcp.types import Implementation
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
PROJECT = ROOT / "ueRouge.uproject"
|
|
ENDPOINT = "http://127.0.0.1:8000/mcp"
|
|
ZONE = timezone(timedelta(hours=8))
|
|
CAPABILITIES = {
|
|
"scene": ("editor_toolset.toolsets.scene.SceneTools", ["get_current_level", "load_level"]),
|
|
"assets": ("editor_toolset.toolsets.asset.AssetTools", ["find_assets", "save_assets", "is_dirty"]),
|
|
"static_mesh_import": ("editor_toolset.toolsets.static_mesh.StaticMeshTools", ["import_file", "get_bounds"]),
|
|
"skeletal_mesh_import": ("editor_toolset.toolsets.skeletal_mesh.SkeletalMeshTools", ["import_file", "get_socket_names"]),
|
|
"texture_import": ("editor_toolset.toolsets.texture.TextureTools", ["import_file"]),
|
|
"blueprint": ("editor_toolset.toolsets.blueprint.BlueprintTools", ["create", "compile_blueprint"]),
|
|
"umg": ("UMGToolSet.UMGToolSet", ["CreateWidgetBlueprint", "AddWidget", "CompileWidgetBlueprint"]),
|
|
"data_assets": ("editor_toolset.toolsets.data_asset.DataAssetTools", ["create"]),
|
|
"automation": ("AutomationTestToolset.AutomationTestToolset", ["DiscoverTests", "RunTests", "GetTestResults"]),
|
|
"editor": ("EditorToolset.EditorAppToolset", ["IsPIERunning", "StartPIE", "StopPIE", "GetCameraTransform", "CaptureViewport"]),
|
|
}
|
|
|
|
|
|
def redact(text):
|
|
text = re.sub(r"session id '[^']+'", "session id '<redacted>'", text)
|
|
text = re.sub(r"(?i)(password|token|api_key|密码)(\s*[:=]\s*)[^,\s\"}]+", r"\1\2<redacted>", text)
|
|
return re.sub(r"https?://[^/\s:@]+:[^/\s@]+@", "<credential-url-redacted>", text)
|
|
|
|
|
|
def verify_owner():
|
|
command = r"""
|
|
$listener = Get-NetTCPConnection -LocalPort 8000 -State Listen -ErrorAction Stop | Where-Object LocalAddress -eq '127.0.0.1' | Select-Object -First 1
|
|
if (-not $listener) { throw 'No loopback listener on port 8000' }
|
|
$process = Get-CimInstance Win32_Process -Filter "ProcessId = $($listener.OwningProcess)"
|
|
$matches = [regex]::Matches($process.CommandLine, '(?i)"([^"\r\n]+\.uproject)"|([A-Z]:[\\/][^\s"]+\.uproject)')
|
|
$projects = @($matches | ForEach-Object { if ($_.Groups[1].Success) { $_.Groups[1].Value } else { $_.Groups[2].Value } })
|
|
[pscustomobject]@{ processId = $process.ProcessId; processName = $process.Name; executablePath = $process.ExecutablePath; projectArguments = $projects; port = $listener.LocalPort } | ConvertTo-Json -Compress
|
|
"""
|
|
result = subprocess.run(["powershell.exe", "-NoProfile", "-Command", command], capture_output=True, check=True)
|
|
owner = json.loads(result.stdout.decode("utf-8-sig"))
|
|
projects = owner["projectArguments"]
|
|
if isinstance(projects, str):
|
|
projects = [projects]
|
|
if owner["processName"].casefold() != "unrealeditor.exe" or len(projects) != 1 or Path(projects[0]).resolve() != PROJECT.resolve():
|
|
raise RuntimeError("Port 8000 does not belong to the exact target project")
|
|
version_path = Path(owner["executablePath"]).parents[2] / "Build/Build.version"
|
|
version = json.loads(version_path.read_text(encoding="utf-8-sig"))
|
|
owner["engineVersion"] = ".".join(str(version[k]) for k in ("MajorVersion", "MinorVersion", "PatchVersion"))
|
|
owner["projectFile"] = str(PROJECT).replace("\\", "/")
|
|
return owner
|
|
|
|
|
|
def unpack(result):
|
|
if result.isError:
|
|
raise RuntimeError("MCP call returned an error; raw payload omitted")
|
|
if result.structuredContent is not None:
|
|
return result.structuredContent
|
|
strings = [block.text for block in result.content if block.type == "text"]
|
|
text = "\n".join(strings)
|
|
try:
|
|
return json.loads(text)
|
|
except json.JSONDecodeError:
|
|
return {"text": text}
|
|
|
|
|
|
class Warnings(logging.Handler):
|
|
def __init__(self, output):
|
|
super().__init__(logging.WARNING)
|
|
self.output = output
|
|
|
|
def emit(self, record):
|
|
self.output.append(redact(record.getMessage()))
|
|
|
|
|
|
async def check():
|
|
owner = verify_owner()
|
|
now = datetime.now(ZONE)
|
|
folder = ROOT / "Docs/QA/GameFrameworkMcpReadiness" / now.strftime("%Y-%m-%d-%H%M%S")
|
|
folder.mkdir(parents=True, exist_ok=False)
|
|
warnings = []
|
|
logger = logging.getLogger("mcp.client.streamable_http")
|
|
handler = Warnings(warnings)
|
|
old_propagate = logger.propagate
|
|
logger.propagate = False
|
|
logger.addHandler(handler)
|
|
report = {"checkedAt": now.isoformat(), "endpoint": ENDPOINT, "project": owner,
|
|
"bindingVerification": "Loopback listener PID and exact .uproject process argument before and after native MCP calls",
|
|
"readOnly": True, "capabilities": {}, "ueBuildExecuted": False,
|
|
"gameFrameworkBuildExecuted": False, "packagedGameExecuted": False}
|
|
try:
|
|
async with streamablehttp_client(ENDPOINT, timeout=15, sse_read_timeout=30, terminate_on_close=True) as (reader, writer, _):
|
|
async with ClientSession(reader, writer, client_info=Implementation(name="irons-gameFramework-readiness", version="1.0")) as session:
|
|
init = await session.initialize()
|
|
report["protocolVersion"] = init.protocolVersion
|
|
tools = await session.list_tools()
|
|
report["routerTools"] = [tool.name for tool in tools.tools]
|
|
discovered = unpack(await session.call_tool("list_toolsets", {}))
|
|
text = discovered.get("text", "")
|
|
report["toolsets"] = re.findall(r"^- ([^:\n]+):", text, re.M)
|
|
report["toolsetCount"] = len(report["toolsets"])
|
|
|
|
async def call(toolset, name, arguments=None):
|
|
result = await session.call_tool("call_tool", {"toolset_name": toolset, "tool_name": name, "arguments": arguments or {}})
|
|
return unpack(result)
|
|
|
|
report["pieBefore"] = (await call("EditorToolset.EditorAppToolset", "IsPIERunning"))["returnValue"]
|
|
report["activeLevel"] = (await call("editor_toolset.toolsets.scene.SceneTools", "get_current_level"))["returnValue"]
|
|
for label, (toolset, expected) in CAPABILITIES.items():
|
|
schema = unpack(await session.call_tool("describe_toolset", {"toolset_name": toolset}))
|
|
names = [tool["name"].rsplit(".", 1)[-1] for tool in schema["tools"]]
|
|
report["capabilities"][label] = {"toolset": toolset, "requiredTools": expected,
|
|
"allAdvertised": all(name in names for name in expected),
|
|
"verification": "Schema discovery; no asset mutations or test execution"}
|
|
log_args = {"category": "", "pattern": "(?i)(error|warning)", "maxEntries": 20}
|
|
entries = (await call("EditorToolset.LogsToolset", "GetLogEntries", log_args))["returnValue"]
|
|
report["recentMatchingLogEntries"] = [redact(entry) for entry in entries]
|
|
report["logScope"] = "Last 20 matching entries from this editor session, including entries predating this check; not a zero-error certification"
|
|
|
|
# This UE build advertises optional struct parameters, but its
|
|
# RPC binder requires explicit values. Reuse the observed camera.
|
|
camera = (await call("EditorToolset.EditorAppToolset", "GetCameraTransform"))["returnValue"]
|
|
capture_arguments = {"captureTransform": camera, "bShowUI": False,
|
|
"annotations": {"gridSpacing": 0, "gridExtent": 0, "gridHeight": 0,
|
|
"maxLabelDistance": 0, "maxLabels": 0,
|
|
"classFilter": {"refPath": "/Script/Engine.Actor"}}}
|
|
capture_result = await session.call_tool("call_tool", {"toolset_name": "EditorToolset.EditorAppToolset", "tool_name": "CaptureViewport", "arguments": capture_arguments})
|
|
capture = unpack(capture_result).get("returnValue", {})
|
|
image = capture.get("image", {})
|
|
encoded = image.get("data")
|
|
if not encoded:
|
|
encoded = next((block.data for block in capture_result.content if block.type == "image"), None)
|
|
if not encoded:
|
|
raise RuntimeError("Viewport capture returned no image")
|
|
pixels = base64.b64decode(encoded)
|
|
if not pixels.startswith(b"\x89PNG\r\n\x1a\n"):
|
|
raise RuntimeError("Viewport capture is not PNG")
|
|
width, height = struct.unpack(">II", pixels[16:24])
|
|
(folder / "editor-viewport.png").write_bytes(pixels)
|
|
report["screenshot"] = {"file": "editor-viewport.png", "width": width, "height": height,
|
|
"environment": "UE editor level viewport", "level": report["activeLevel"]}
|
|
report["viewportCaptureBindingNote"] = "Explicit observed camera and disabled annotation values supplied because this build rejects omitted optional struct parameters"
|
|
report["pieAfter"] = (await call("EditorToolset.EditorAppToolset", "IsPIERunning"))["returnValue"]
|
|
report["activeLevelAfter"] = (await call("editor_toolset.toolsets.scene.SceneTools", "get_current_level"))["returnValue"]
|
|
if report["pieBefore"] != report["pieAfter"] or report["activeLevel"] != report["activeLevelAfter"]:
|
|
raise RuntimeError("Editor state changed during the check")
|
|
after = verify_owner()
|
|
if after["processId"] != owner["processId"]:
|
|
raise RuntimeError("Editor process changed during the check")
|
|
report["nativeQueriesSucceeded"] = True
|
|
report["clientSessionClosed"] = True
|
|
report["sharedServicePreserved"] = True
|
|
finally:
|
|
logger.removeHandler(handler)
|
|
logger.propagate = old_propagate
|
|
report["clientWarnings"] = warnings
|
|
report["serverSessionDeletionIndependentlyVerified"] = False
|
|
(folder / "verification.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
print(json.dumps({"recordDirectory": str(folder), "engineVersion": owner["engineVersion"],
|
|
"nativeQueriesSucceeded": report["nativeQueriesSucceeded"], "toolsetCount": report["toolsetCount"],
|
|
"pie": report["pieAfter"], "level": report["activeLevel"], "screenshot": report["screenshot"],
|
|
"clientWarnings": warnings, "sharedServicePreserved": True}, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
asyncio.run(check())
|