#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Unity Agent CLI — Interactive colored game development agent.
A zero-dependency interactive REPL for driving the Unity game agent.
Uses raw ANSI escape codes for color, box-drawing characters for panels,
and an animated spinner while the LLM is "thinking".
Features
--------
* Colored output (cyan/magenta/green/red/yellow/blue/white/dim)
* Box-drawing panels and tables
* Animated "thinking" spinner (background thread)
* Progress bars during generation
* C# syntax-highlighted code boxes
* Full slash-command suite:
/help /tools /presets /generate /open /scene /clear /config
/history /export /test /stats /new /add /script /run /exit
* OpenAI-compatible Z.ai LLM client (urllib, no external deps)
* Tool-call parser & executor with graceful fallback tools
Usage
-----
python -m unity_agent.cli
python run.py
Environment
-----------
ZAI_API_KEY — Z.ai API key (https://z.ai)
ZAI_MODEL — model name (default: glm-4.5-flash)
ZAI_BASE_URL — OpenAI-compatible endpoint
Requirements
------------
Standard library only. No rich / click / prompt_toolkit / httpx needed.
"""
from __future__ import annotations
import os
import sys
import json
import re
import time
import shutil
import zipfile
import threading
import subprocess
import urllib.request
import urllib.error
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional
# ---------------------------------------------------------------------------
# Optional integration with the rest of the unity_agent package.
# The CLI degrades gracefully if sibling modules are absent, so it can run
# standalone with built-in fallback tools.
# ---------------------------------------------------------------------------
_PKG_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _PKG_ROOT not in sys.path:
sys.path.insert(0, _PKG_ROOT)
_TOOL_REGISTRY = None
_ToolResult = None
_SystemSettings = None
_SYSTEM_PROMPT = ""
try: # pragma: no cover - exercised only when the full package exists
from unity_agent.config import Settings as _Settings # type: ignore
_SystemSettings = _Settings
except Exception:
_SystemSettings = None
try: # pragma: no cover
from unity_agent.orchestrator.prompts import SYSTEM_PROMPT as _SP # type: ignore
_SYSTEM_PROMPT = _SP or ""
except Exception:
_SYSTEM_PROMPT = ""
try: # pragma: no cover
from unity_agent.tools.base import _GLOBAL_REGISTRY, ToolResult # type: ignore
_TOOL_REGISTRY = _GLOBAL_REGISTRY
_ToolResult = ToolResult
except Exception:
_TOOL_REGISTRY = None
_ToolResult = None
# ===========================================================================
# ANSI COLOR CODES
# ===========================================================================
class C:
"""Raw ANSI escape sequences for terminal styling.
Every constant is a plain string so it can be concatenated freely.
Use :py:meth:`strip` to remove escape codes when measuring string width
or writing to a non-TTY.
"""
RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
ITALIC = "\033[3m"
UNDERLINE = "\033[4m"
BLINK = "\033[5m"
REVERSE = "\033[7m"
HIDDEN = "\033[8m"
# Foreground (normal)
BLACK = "\033[30m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
BLUE = "\033[34m"
MAGENTA = "\033[35m"
CYAN = "\033[36m"
WHITE = "\033[37m"
# Foreground (bright)
B_BLACK = "\033[90m" # dim / secondary
B_RED = "\033[91m"
B_GREEN = "\033[92m"
B_YELLOW = "\033[93m"
B_BLUE = "\033[94m"
B_MAGENTA = "\033[95m"
B_CYAN = "\033[96m"
B_WHITE = "\033[97m"
# Background (normal)
BG_BLACK = "\033[40m"
BG_RED = "\033[41m"
BG_GREEN = "\033[42m"
BG_YELLOW = "\033[43m"
BG_BLUE = "\033[44m"
BG_MAGENTA = "\033[45m"
BG_CYAN = "\033[46m"
BG_WHITE = "\033[47m"
# Cursor / screen control
CLEAR = "\033[2J"
CLEAR_LINE = "\033[2K"
CURSOR_HOME = "\033[H"
CURSOR_UP = "\033[1A"
CURSOR_DOWN = "\033[1B"
CURSOR_RIGHT = "\033[1C"
CURSOR_LEFT = "\033[1D"
SAVE_CURSOR = "\033[s"
RESTORE_CURSOR = "\033[u"
HIDE_CURSOR = "\033[?25l"
SHOW_CURSOR = "\033[?25h"
_ANSI_RE = re.compile(r"\033\[[0-9;]*[A-Za-z]")
@classmethod
def strip(cls, text: str) -> str:
"""Return ``text`` with all ANSI escape sequences removed."""
return cls._ANSI_RE.sub("", text)
@classmethod
def width(cls, text: str) -> int:
"""Visible width of ``text`` (ignoring escape codes)."""
return len(cls.strip(text))
@classmethod
def wrap(cls, text: str, color: str, bold: bool = False) -> str:
"""Wrap ``text`` in ``color`` (optionally bold) and reset."""
prefix = (cls.BOLD if bold else "") + color
return f"{prefix}{text}{cls.RESET}"
@classmethod
def available(cls) -> bool:
"""Return True if the current stream looks color-capable."""
if os.environ.get("NO_COLOR"):
return False
if os.environ.get("FORCE_COLOR"):
return True
try:
return sys.stdout.isatty()
except Exception:
return False
# A friendly alias used throughout the module for "dim" text.
DIM = C.B_BLACK
# ===========================================================================
# BOX-DRAWING CHARACTERS
# ===========================================================================
class Box:
"""Box-drawing glyphs (single-line and double-line sets)."""
# Single line
TL, TR, BL, BR = "┌", "┐", "└", "┘"
H, V = "─", "│"
LT, RT, TT, BT, CROSS = "├", "┤", "┬", "┴", "┼"
# Double line
DTL, DTR, DBL, DBR = "╔", "╗", "╚", "╝"
DH, DV = "═", "║"
DLT, DRT, DTT, DBT, DCROSS = "╠", "╣", "╦", "╩", "╬"
# Rounded single line corners
RTL, RTR, RBL, RBR = "╭", "╮", "╰", "╯"
# Misc
BULLET = "•"
ARROW = "→"
CHECK = "✓"
CROSS_X = "✗"
STAR = "✦"
DIAMOND = "◆"
DOT = "·"
# ===========================================================================
# TERMINAL HELPERS
# ===========================================================================
def term_width(default: int = 80) -> int:
"""Best-effort terminal column width."""
try:
return shutil.get_terminal_size((default, 24)).columns
except Exception:
return default
def term_height(default: int = 24) -> int:
"""Best-effort terminal row height."""
try:
return shutil.get_terminal_size((80, default)).lines
except Exception:
return default
def visible_len(s: str) -> int:
"""Length of ``s`` ignoring ANSI escape codes."""
return C.width(s)
def pad_to(s: str, width: int, fill: str = " ", align: str = "left") -> str:
"""Pad ``s`` (which may contain ANSI codes) to a visible ``width``."""
pad = max(0, width - visible_len(s))
if align == "right":
return fill * pad + s
if align == "center":
left = pad // 2
right = pad - left
return fill * left + s + fill * right
return s + fill * pad
# ===========================================================================
# PANEL / BORDER RENDERING
# ===========================================================================
def hline(width: Optional[int] = None, char: str = Box.H, color: str = DIM) -> str:
"""Return a single horizontal line."""
width = width or term_width()
return color + (char * width) + C.RESET
def panel(
lines: List[str],
title: Optional[str] = None,
border_color: str = C.CYAN,
title_color: str = C.B_CYAN,
pad: int = 1,
width: Optional[int] = None,
rounded: bool = False,
) -> str:
"""Render ``lines`` inside a box-drawing panel.
``lines`` may already contain ANSI styling; widths are computed from the
visible characters only.
"""
width = width or min(term_width(), 90)
inner = width - 2 - pad * 2 # borders + padding
tl = Box.RTL if rounded else Box.TL
tr = Box.RTR if rounded else Box.TR
bl = Box.RBL if rounded else Box.BL
br = Box.RBR if rounded else Box.RBR
out: List[str] = []
if title:
title_str = f" {title} "
title_visible = len(title_str)
left_dash = (inner + pad * 2 - title_visible) // 2
right_dash = inner + pad * 2 - title_visible - left_dash
out.append(
border_color + tl + Box.H * left_dash
+ title_color + title_str
+ border_color + Box.H * right_dash + tr + C.RESET
)
else:
out.append(border_color + tl + Box.H * (width - 2) + tr + C.RESET)
def inner_row(content: str) -> str:
vis = visible_len(content)
overflow = max(0, vis - inner)
if overflow:
# Truncate visually (strip ansi, cut, re-add reset) — simple approach.
stripped = C.strip(content)
content = stripped[: inner - 1] + "…"
return (
border_color + Box.V + C.RESET
+ " " * pad + content + " " * pad
+ border_color + Box.V + C.RESET
)
if not lines:
out.append(inner_row(""))
for ln in lines:
if ln == "":
out.append(inner_row(""))
else:
for sub in _wrap_visible(ln, inner):
out.append(inner_row(sub))
out.append(border_color + bl + Box.H * (width - 2) + br + C.RESET)
return "\n".join(out)
def _wrap_visible(text: str, width: int) -> List[str]:
"""Word-wrap ``text`` to ``width`` visible columns."""
stripped = C.strip(text)
if not stripped:
return [""]
words = stripped.split(" ")
rows: List[str] = []
cur = ""
for w in words:
if not cur:
cur = w
elif len(cur) + 1 + len(w) <= width:
cur += " " + w
else:
rows.append(cur)
cur = w
if cur:
rows.append(cur)
return rows or [""]
def divider(title: Optional[str] = None, color: str = C.B_CYAN) -> str:
"""A thin divider line, optionally with a centered title."""
w = term_width()
if not title:
return color + Box.H * w + C.RESET
title_str = f" {title} "
pad = max(0, (w - len(title_str)) // 2)
right = max(0, w - len(title_str) - pad)
return color + Box.H * pad + title_str + Box.H * right + C.RESET
# ===========================================================================
# TABLE RENDERING
# ===========================================================================
def table(
headers: List[str],
rows: List[List[str]],
colors: Optional[List[str]] = None,
header_color: str = C.B_CYAN,
border_color: str = DIM,
min_widths: Optional[List[int]] = None,
) -> str:
"""Render a colored box-drawing table.
``colors`` optionally provides a per-column color applied to row cells.
"""
ncol = len(headers)
colors = colors or ([C.WHITE] * ncol)
min_widths = min_widths or ([0] * ncol)
# Compute column widths from visible text.
widths = list(min_widths)
for i, h in enumerate(headers):
widths[i] = max(widths[i], len(h))
for row in rows:
for i, cell in enumerate(row):
if i < ncol:
widths[i] = max(widths[i], len(C.strip(cell)))
def sep(left: str, mid: str, right: str, fill: str = Box.H) -> str:
parts = [left]
for i, w in enumerate(widths):
parts.append(fill * (w + 2))
parts.append(mid if i < ncol - 1 else right)
return border_color + "".join(parts) + C.RESET
out: List[str] = []
out.append(sep(Box.TL, Box.TT, Box.TR))
# Header row
cells = []
for i, h in enumerate(headers):
cells.append(" " + header_color + pad_to(h, widths[i]) + C.RESET + " ")
out.append(border_color + Box.V + C.RESET + (border_color + Box.V + C.RESET).join(cells))
out.append(sep(Box.LT, Box.CROSS, Box.RT))
# Data rows
for r_i, row in enumerate(rows):
cells = []
for i in range(ncol):
cell = row[i] if i < len(row) else ""
col = colors[i] if i < len(colors) else C.WHITE
cells.append(" " + col + pad_to(C.strip(cell), widths[i]) + C.RESET + " ")
out.append(border_color + Box.V + C.RESET + (border_color + Box.V + C.RESET).join(cells))
if r_i < len(rows) - 1:
out.append(sep(Box.LT, Box.CROSS, Box.RT))
out.append(sep(Box.BL, Box.BT, Box.BR))
return "\n".join(out)
# ===========================================================================
# C# SYNTAX HIGHLIGHTING
# ===========================================================================
CSHARP_KEYWORDS = {
"abstract", "as", "base", "bool", "break", "byte", "case", "catch",
"char", "checked", "class", "const", "continue", "decimal", "default",
"delegate", "do", "double", "else", "enum", "event", "explicit",
"extern", "false", "finally", "fixed", "float", "for", "foreach",
"goto", "if", "implicit", "in", "int", "interface", "internal", "is",
"lock", "long", "namespace", "new", "null", "object", "operator",
"out", "override", "params", "private", "protected", "public",
"readonly", "ref", "return", "sbyte", "sealed", "short", "sizeof",
"stackalloc", "static", "string", "struct", "switch", "this", "throw",
"true", "try", "typeof", "uint", "ulong", "unchecked", "unsafe",
"ushort", "using", "var", "virtual", "void", "volatile", "while",
"yield", "async", "await", "get", "set", "value", "where", "select",
"from", "group", "into", "orderby", "join", "let", "on", "equals",
"by", "ascending", "descending", "nameof", "when", "global", "partial",
}
CSHARP_TYPES = {
"Vector2", "Vector3", "Vector4", "Quaternion", "Transform", "GameObject",
"MonoBehaviour", "Rigidbody", "Rigidbody2D", "Collider", "Collider2D",
"AudioSource", "AudioClip", "Material", "Mesh", "MeshRenderer",
"MeshFilter", "Camera", "Light", "Canvas", "Button", "Text", "Image",
"RectTransform", "SceneManager", "Scene", "Input", "Time", "Physics",
"Physics2D", "Debug", "Color", "Ray", "RaycastHit", "List", "Dictionary",
"HashSet", "Queue", "Stack", "Action", "Func", "Predicate", "Tuple",
"IEnumerator", "WaitForSeconds", "WaitForEndOfFrame", "Coroutine",
"ScriptableObject", "Object", "Component", "Behaviour", "Animator",
"Animation", "Sprite", "Texture2D", "Shader", "Renderer", "TrailRenderer",
"LineRenderer", "ParticleSystem", "BoxCollider", "SphereCollider",
"CapsuleCollider", "CharacterController", "NavMeshAgent", "RaycastHit2D",
"KeyCode", "Touch", "TouchPhase", "PlayerPrefs", "Application",
"Screen", "Cursor", "Random", "Mathf", "Vector3Int", "Vector2Int",
"Gradient", "AnimationCurve", "GradientColorKey", "GradientAlphaKey",
"SyncVar", "Command", "ClientRpc", "SerializeField", "HideInInspector",
"Range", "Header", "Tooltip", "TextArea", "ContextMenu", "RequireComponent",
}
def highlight_csharp(code: str) -> str:
"""Apply simple regex-based C# syntax highlighting.
Order matters: comments and strings are matched first so that keywords
inside them are not re-colored.
"""
token_spec = [
("COMMENT", r"//[^\n]*|/\*[\s\S]*?\*/"),
("STRING", r"@?\"(?:\"\"|\\.|[^\"\\])*\"|'(?:\\.|[^'\\])*'"),
("NUMBER", r"\b\d+\.?\d*[fdmFDM]?\b|\b0x[0-9a-fA-F]+\b"),
("IDENT", r"[A-Za-z_][A-Za-z0-9_]*"),
("OP", r"[{}\[\]().,;:?=+\-*/%&|<>!^~]+"),
("WS", r"\s+"),
("ANY", r"."),
]
# Build a master alternation, naming each group so re.finditer exposes
# m.lastgroup. Use %-formatting to avoid f-string brace conflicts with
# the regex patterns (which contain literal { } ).
master = "|".join("(?P<%s>%s)" % (name, pat) for name, pat in token_spec)
out: List[str] = []
for m in re.finditer(master, code):
kind = m.lastgroup
val = m.group()
if kind == "COMMENT":
out.append(C.DIM + C.ITALIC + val + C.RESET)
elif kind == "STRING":
out.append(C.GREEN + val + C.RESET)
elif kind == "NUMBER":
out.append(C.YELLOW + val + C.RESET)
elif kind == "IDENT":
if val in CSHARP_KEYWORDS:
out.append(C.MAGENTA + val + C.RESET)
elif val in CSHARP_TYPES:
out.append(C.CYAN + val + C.RESET)
elif val[0].isupper():
out.append(C.B_CYAN + val + C.RESET)
else:
out.append(C.WHITE + val + C.RESET)
elif kind == "OP":
out.append(C.B_BLACK + val + C.RESET)
else:
out.append(val)
return "".join(out)
def code_block(code: str, language: str = "csharp", title: Optional[str] = None) -> str:
"""Render source code inside a bordered, syntax-highlighted panel."""
highlighted = highlight_csharp(code) if language.lower() in ("csharp", "c#", "cs") else C.WHITE + code + C.RESET
raw_lines = code.splitlines() or [""]
width = min(max(60, max((len(l) for l in raw_lines), default=0) + 8), term_width())
header = title or language.upper()
lines = []
for ln in highlighted.splitlines() or [""]:
lines.append(ln if ln else " ")
return panel(lines, title=header, border_color=C.B_BLUE, title_color=C.BLUE, pad=1, width=width)
# ===========================================================================
# ANIMATED SPINNER
# ===========================================================================
class Spinner:
"""Animated "thinking" spinner running on a background thread.
Usage::
s = Spinner("Thinking")
s.start()
... long operation ...
s.stop()
"""
FRAMES = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]
ALT_FRAMES = ["|", "/", "-", "\\"]
def __init__(self, message: str = "Thinking", color: str = C.MAGENTA, interval: float = 0.08):
self.message = message
self.color = color
self.interval = interval
self._stop = threading.Event()
self._thread: Optional[threading.Thread] = None
self._frame_idx = 0
# Use braille frames if stdout supports UTF-8; fall back to ASCII.
try:
if sys.stdout.encoding and "utf" in sys.stdout.encoding.lower():
self.frames = self.FRAMES
else:
self.frames = self.ALT_FRAMES
except Exception:
self.frames = self.ALT_FRAMES
def _spin(self) -> None:
sys.stdout.write(C.HIDE_CURSOR)
sys.stdout.flush()
while not self._stop.is_set():
frame = self.frames[self._frame_idx % len(self.frames)]
self._frame_idx += 1
line = f" {self.color}{frame}{C.RESET} {self.color}{self.message}{C.RESET}{C.B_BLACK}...{C.RESET}"
sys.stdout.write("\r" + C.CLEAR_LINE + "\r")
sys.stdout.write(line)
sys.stdout.flush()
time.sleep(self.interval)
sys.stdout.write("\r" + C.CLEAR_LINE + "\r")
sys.stdout.write(C.SHOW_CURSOR)
sys.stdout.flush()
def start(self) -> "Spinner":
self._stop.clear()
if C.available():
self._thread = threading.Thread(target=self._spin, daemon=True)
self._thread.start()
else:
sys.stdout.write(f" {self.message}...\n")
sys.stdout.flush()
return self
def stop(self, final_message: Optional[str] = None) -> None:
self._stop.set()
if self._thread and self._thread.is_alive():
self._thread.join(timeout=1.0)
self._thread = None
if final_message:
sys.stdout.write(final_message + "\n")
sys.stdout.flush()
def update(self, message: str) -> None:
self.message = message
# ===========================================================================
# PROGRESS BAR
# ===========================================================================
class ProgressBar:
"""A single-line animated progress bar."""
def __init__(self, total: int, label: str = "Working", width: int = 30,
color: str = C.CYAN, fill_color: str = C.B_CYAN):
self.total = max(1, total)
self.label = label
self.width = width
self.color = color
self.fill_color = fill_color
self.value = 0
def _render(self) -> str:
pct = self.value / self.total
filled = int(self.width * pct)
bar = self.fill_color + "█" * filled + C.B_BLACK + "░" * (self.width - filled) + C.RESET
return f" {self.color}{self.label}{C.RESET} [{bar}] {C.BOLD}{int(pct * 100):3d}%{C.RESET}"
def update(self, value: int, label: Optional[str] = None) -> None:
self.value = min(self.total, max(0, value))
if label:
self.label = label
sys.stdout.write("\r" + C.CLEAR_LINE + "\r")
sys.stdout.write(self._render())
sys.stdout.flush()
def finish(self, message: Optional[str] = None) -> None:
self.update(self.total)
sys.stdout.write("\n")
if message:
sys.stdout.write(message + "\n")
sys.stdout.flush()
def __enter__(self) -> "ProgressBar":
sys.stdout.write(self._render() + "\r")
sys.stdout.flush()
return self
def __exit__(self, exc_type, exc, tb) -> None:
if exc_type is None:
self.finish()
else:
sys.stdout.write("\n")
sys.stdout.flush()
# ===========================================================================
# LLM CLIENT (OpenAI-compatible, urllib only)
# ===========================================================================
class LLMError(Exception):
"""Raised when the LLM endpoint returns an error."""
class LLMClient:
"""Z.ai OpenAI-compatible LLM client (standard library only).
Talks to ``{base_url}/chat/completions`` using :mod:`urllib.request`.
Supports plain completion as well as a simple token stream callback.
"""
def __init__(
self,
api_key: str,
model: str = "glm-4.5-flash",
base_url: str = "https://open.bigmodel.cn/api/paas/v4",
timeout: float = 120.0,
):
self.api_key = api_key
self.model = model
self.base_url = base_url.rstrip("/")
self.timeout = timeout
# -- low-level HTTP -----------------------------------------------------
def _post(self, path: str, payload: Dict[str, Any], stream: bool = False):
url = f"{self.base_url}{path}"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
"Accept": "text/event-stream" if stream else "application/json",
}
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
try:
return urllib.request.urlopen(req, timeout=self.timeout)
except urllib.error.HTTPError as e:
body = ""
try:
body = e.read().decode("utf-8", errors="replace")
except Exception:
pass
raise LLMError(f"HTTP {e.code} {e.reason}: {body[:400]}") from None
except urllib.error.URLError as e:
raise LLMError(f"Network error: {e.reason}") from None
# -- public API ---------------------------------------------------------
def chat(
self,
messages: List[Dict[str, str]],
temperature: float = 0.2,
max_tokens: int = 4096,
on_token: Optional[Callable[[str], None]] = None,
) -> str:
"""Send a chat completion request and return the assistant message.
If ``on_token`` is provided, the response is streamed and the
callback receives each delta token; the concatenated content is
still returned.
"""
payload: Dict[str, Any] = {
"model": self.model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
}
if on_token is not None:
payload["stream"] = True
if on_token is None:
resp = self._post("/chat/completions", payload, stream=False)
try:
raw = resp.read().decode("utf-8")
data = json.loads(raw)
finally:
resp.close()
if "choices" not in data:
raise LLMError(f"Unexpected response: {raw[:400]}")
return data["choices"][0]["message"]["content"]
# Streaming path
resp = self._post("/chat/completions", payload, stream=True)
full: List[str] = []
try:
buf = ""
for chunk in resp:
buf += chunk.decode("utf-8", errors="replace")
while "\n" in buf:
line, buf = buf.split("\n", 1)
line = line.strip()
if not line or not line.startswith("data:"):
continue
payload_str = line[len("data:"):].strip()
if payload_str == "[DONE]":
break
try:
obj = json.loads(payload_str)
except json.JSONDecodeError:
continue
try:
delta = obj["choices"][0]["delta"].get("content", "")
except (KeyError, IndexError):
delta = ""
if delta:
full.append(delta)
on_token(delta)
finally:
resp.close()
return "".join(full)
def chat_with_tools(
self,
messages: List[Dict[str, str]],
temperature: float = 0.2,
max_tokens: int = 4096,
) -> str:
"""Convenience wrapper identical to :meth:`chat` for readability."""
return self.chat(messages, temperature=temperature, max_tokens=max_tokens)
# ===========================================================================
# TOOL-CALL PARSING
# ===========================================================================
_TOOL_CALL_RE = re.compile(
r"```(?:tool_call|tool|json)\s*\n(?P.*?)```",
re.DOTALL | re.IGNORECASE,
)
_TAG_RE = re.compile(r"(.*?)", re.DOTALL | re.IGNORECASE)
def parse_tool_call(text: str) -> Optional[Dict[str, Any]]:
"""Extract a tool call from an LLM response.
Recognized formats::
```tool_call
{"name": "write_file", "arguments": {...}}
```
```json
{"name": "write_file", "arguments": {...}}
```
{"name": "...", "arguments": {...}}
{"name": "...", "arguments": {...}} (inline JSON)
Returns a dict with at least ``name`` and ``arguments`` keys, or ``None``
if no tool call was found.
"""
if not text:
return None
# Fenced code blocks first (most reliable).
for m in _TOOL_CALL_RE.finditer(text):
body = m.group("body").strip()
parsed = _try_json(body)
if parsed and isinstance(parsed, dict) and "name" in parsed:
return _normalize_tool_call(parsed)
# ... tags.
for m in _TAG_RE.finditer(text):
body = m.group(1).strip()
parsed = _try_json(body)
if parsed and isinstance(parsed, dict) and "name" in parsed:
return _normalize_tool_call(parsed)
# Inline JSON object containing "name" and "arguments".
for m in re.finditer(r"\{[\s\S]*\}", text):
candidate = m.group(0)
parsed = _try_json(candidate)
if parsed and isinstance(parsed, dict) and "name" in parsed and "arguments" in parsed:
return _normalize_tool_call(parsed)
return None
def parse_all_tool_calls(text: str) -> List[Dict[str, Any]]:
"""Return every tool call found in ``text`` (in order)."""
out: List[Dict[str, Any]] = []
if not text:
return out
for m in _TOOL_CALL_RE.finditer(text):
parsed = _try_json(m.group("body").strip())
if parsed and isinstance(parsed, dict) and "name" in parsed:
out.append(_normalize_tool_call(parsed))
for m in _TAG_RE.finditer(text):
parsed = _try_json(m.group(1).strip())
if parsed and isinstance(parsed, dict) and "name" in parsed:
out.append(_normalize_tool_call(parsed))
if not out:
single = parse_tool_call(text)
if single:
out.append(single)
return out
def _try_json(text: str) -> Optional[Any]:
try:
return json.loads(text)
except Exception:
# Try to repair common LLM quirks: trailing commas.
try:
cleaned = re.sub(r",\s*([}\]])", r"\1", text)
return json.loads(cleaned)
except Exception:
return None
def _normalize_tool_call(obj: Dict[str, Any]) -> Dict[str, Any]:
name = obj.get("name") or obj.get("tool") or obj.get("function", {}).get("name")
arguments = obj.get("arguments")
if arguments is None:
arguments = obj.get("parameters") or obj.get("args") or obj.get("input") or {}
if isinstance(arguments, str):
parsed = _try_json(arguments)
arguments = parsed if parsed is not None else {"_raw": arguments}
if not isinstance(arguments, dict):
arguments = {"value": arguments}
return {"name": str(name), "arguments": arguments}
# ===========================================================================
# TOOL RESULT + FALLBACK TOOL REGISTRY
# ===========================================================================
@dataclass
class ToolResult:
"""Standard tool execution result (matches the package ToolResult)."""
success: bool = True
message: str = ""
data: Any = None
files: List[str] = field(default_factory=list)
error: Optional[str] = None
def as_dict(self) -> Dict[str, Any]:
return {
"success": self.success,
"message": self.message,
"data": self.data,
"files": self.files,
"error": self.error,
}
class _FallbackTool:
"""A simple builtin tool used when the full unity_agent package is absent."""
def __init__(self, name: str, description: str, schema: Dict[str, Any],
fn: Callable[..., ToolResult]):
self.name = name
self.__doc__ = description
self.description = description
self.input_schema = schema
self._fn = fn
def run(self, **kwargs: Any) -> ToolResult:
try:
return self._fn(**kwargs)
except Exception as e: # pragma: no cover - defensive
return ToolResult(success=False, message=str(e), error=str(e))
class _FallbackRegistry:
"""Minimal registry mimicking ``_GLOBAL_REGISTRY``."""
def __init__(self) -> None:
self._tools: Dict[str, _FallbackTool] = {}
def register(self, name: str, description: str, schema: Dict[str, Any],
fn: Callable[..., ToolResult]) -> None:
self._tools[name] = _FallbackTool(name, description, schema, fn)
def get(self, name: str) -> Optional[_FallbackTool]:
return self._tools.get(name)
def all_names(self) -> List[str]:
return sorted(self._tools.keys())
def all(self) -> List[_FallbackTool]:
return [self._tools[n] for n in self.all_names()]
def _build_fallback_registry() -> _FallbackRegistry:
"""Populate a small but useful set of builtin tools."""
reg = _FallbackRegistry()
def _safe_path(base: str, *parts: str) -> str:
root = os.path.abspath(base)
full = os.path.abspath(os.path.join(root, *parts))
if not full.startswith(root):
raise ValueError("Path escapes output directory")
return full
def t_write_file(path: str = "", content: str = "", output_dir: str = "./unity_projects",
settings: Any = None, transport: Any = None, **_: Any) -> ToolResult:
base = getattr(settings, "output_dir", output_dir) if settings else output_dir
full = _safe_path(base, path)
os.makedirs(os.path.dirname(full), exist_ok=True)
with open(full, "w", encoding="utf-8") as fh:
fh.write(content)
return ToolResult(success=True, message=f"Wrote {len(content)} bytes to {path}",
files=[full])
def t_read_file(path: str = "", output_dir: str = "./unity_projects",
settings: Any = None, **_: Any) -> ToolResult:
base = getattr(settings, "output_dir", output_dir) if settings else output_dir
full = _safe_path(base, path)
if not os.path.exists(full):
return ToolResult(success=False, message=f"File not found: {path}", error="not found")
with open(full, "r", encoding="utf-8", errors="replace") as fh:
content = fh.read()
return ToolResult(success=True, message=f"Read {len(content)} bytes from {path}",
data=content, files=[full])
def t_list_files(path: str = "", output_dir: str = "./unity_projects",
settings: Any = None, **_: Any) -> ToolResult:
base = getattr(settings, "output_dir", output_dir) if settings else output_dir
full = _safe_path(base, path) if path else base
if not os.path.isdir(full):
return ToolResult(success=False, message=f"Directory not found: {path}")
files: List[str] = []
for r, _d, fs in os.walk(full):
for f in fs:
files.append(os.path.relpath(os.path.join(r, f), base))
files.sort()
return ToolResult(success=True, message=f"{len(files)} files", data=files, files=files)
def t_setup_unity_project(project_name: str = "UnityGame",
game_type: str = "minimal", output_dir: str = "./unity_projects",
settings: Any = None, **_: Any) -> ToolResult:
base = getattr(settings, "output_dir", output_dir) if settings else output_dir
root = _safe_path(base, project_name)
layout = [
("Assets/Scripts", ""),
("Assets/Scenes", ""),
("Assets/Prefabs", ""),
("Assets/Materials", ""),
("Assets/Resources", ""),
("ProjectSettings", ""),
("Packages", ""),
]
for sub, _ in layout:
os.makedirs(os.path.join(root, sub), exist_ok=True)
# Minimal Unity marker files so the folder looks like a project.
_write(os.path.join(root, "Assets", ".gitkeep"), "")
_write(os.path.join(root, "ProjectSettings", "ProjectVersion.txt"),
"m_EditorVersion: 2022.3.0f1\n")
_write(os.path.join(root, "Packages", "manifest.json"),
json.dumps({"dependencies": {"com.unity.3d.core": "1.0.0"}}, indent=2))
return ToolResult(success=True,
message=f"Created Unity project skeleton at {project_name}",
files=[root], data={"root": root})
def t_generate_complete_game(game_name: str = "Game", game_type: str = "minimal",
output_dir: str = "./unity_projects",
settings: Any = None, **_: Any) -> ToolResult:
base = getattr(settings, "output_dir", output_dir) if settings else output_dir
proj = _safe_path(base, game_name)
# Reuse setup then drop a sample script + scene.
setup = t_setup_unity_project(game_name, game_type, base, settings)
files = list(setup.files)
sample_cs = _csharp_player_controller(game_name, game_type)
cs_path = os.path.join(proj, "Assets", "Scripts", "PlayerController.cs")
_write(cs_path, sample_cs)
files.append(cs_path)
readme = os.path.join(proj, "README.md")
_write(readme, f"# {game_name}\n\nGenerated by Unity Agent.\n\nType: {game_type}\n")
files.append(readme)
return ToolResult(success=True,
message=f"Generated {game_type} game '{game_name}' with {len(files)} files.",
files=files, data={"root": proj, "type": game_type})
def t_create_script(script_name: str = "NewScript", code: str = "",
output_dir: str = "./unity_projects",
settings: Any = None, **_: Any) -> ToolResult:
base = getattr(settings, "output_dir", output_dir) if settings else output_dir
if not script_name.endswith(".cs"):
script_name = script_name + ".cs"
full = _safe_path(base, "Assets", "Scripts", script_name)
os.makedirs(os.path.dirname(full), exist_ok=True)
if not code.strip():
code = _csharp_template(script_name[:-3])
_write(full, code)
return ToolResult(success=True, message=f"Created script {script_name}", files=[full])
def t_create_scene(scene_name: str = "Main",
output_dir: str = "./unity_projects",
settings: Any = None, **_: Any) -> ToolResult:
base = getattr(settings, "output_dir", output_dir) if settings else output_dir
if not scene_name.endswith(".unity"):
scene_name = scene_name + ".unity"
full = _safe_path(base, "Assets", "Scenes", scene_name)
os.makedirs(os.path.dirname(full), exist_ok=True)
_write(full, "%YAML 1.1\n%TAG !u! tag:unity3d.com,2011:\n---\n# Empty scene stub\n")
return ToolResult(success=True, message=f"Created scene {scene_name}", files=[full])
def t_create_prefab(prefab_name: str = "NewPrefab",
output_dir: str = "./unity_projects",
settings: Any = None, **_: Any) -> ToolResult:
base = getattr(settings, "output_dir", output_dir) if settings else output_dir
if not prefab_name.endswith(".prefab"):
prefab_name = prefab_name + ".prefab"
full = _safe_path(base, "Assets", "Prefabs", prefab_name)
os.makedirs(os.path.dirname(full), exist_ok=True)
_write(full, "%YAML 1.1\n%TAG !u! tag:unity3d.com,2011:\n---\n# Empty prefab stub\n")
return ToolResult(success=True, message=f"Created prefab {prefab_name}", files=[full])
def t_run_qa_checks(output_dir: str = "./unity_projects",
settings: Any = None, **_: Any) -> ToolResult:
base = getattr(settings, "output_dir", output_dir) if settings else output_dir
if not os.path.isdir(base):
return ToolResult(success=False, message="No project directory to check")
issues: List[str] = []
cs_count = 0
scene_count = 0
for r, _d, fs in os.walk(base):
for f in fs:
if f.endswith(".cs"):
cs_count += 1
p = os.path.join(r, f)
try:
with open(p, encoding="utf-8", errors="replace") as fh:
src = fh.read()
if "MonoBehaviour" in src and "void Start" not in src and "void Awake" not in src:
issues.append(f"{f}: MonoBehaviour without Start/Awake")
# brace balance
if src.count("{") != src.count("}"):
issues.append(f"{f}: unbalanced braces")
except Exception as e:
issues.append(f"{f}: read error {e}")
elif f.endswith(".unity"):
scene_count += 1
if cs_count == 0:
issues.append("No C# scripts found")
if scene_count == 0:
issues.append("No scenes found")
ok = not issues
msg = f"QA {'PASSED' if ok else 'FAILED'}: {cs_count} scripts, {scene_count} scenes"
return ToolResult(success=ok, message=msg, data={"issues": issues},
files=[base], error=None if ok else "; ".join(issues))
reg.register("write_file", "Write a file to the project output directory.",
{"path": "str", "content": "str"}, t_write_file)
reg.register("read_file", "Read a file from the project output directory.",
{"path": "str"}, t_read_file)
reg.register("list_files", "List files in the project (optionally under a subpath).",
{"path": "str"}, t_list_files)
reg.register("setup_unity_project", "Create a Unity project folder skeleton.",
{"project_name": "str", "game_type": "str"}, t_setup_unity_project)
reg.register("generate_complete_game", "Generate a complete playable game from a preset.",
{"game_name": "str", "game_type": "str"}, t_generate_complete_game)
reg.register("create_script", "Create a C# script under Assets/Scripts.",
{"script_name": "str", "code": "str"}, t_create_script)
reg.register("create_scene", "Create a Unity scene stub.",
{"scene_name": "str"}, t_create_scene)
reg.register("create_prefab", "Create a Unity prefab stub.",
{"prefab_name": "str"}, t_create_prefab)
reg.register("run_qa_checks", "Run QA / lint checks on the current project.",
{}, t_run_qa_checks)
return reg
def _write(path: str, content: str) -> None:
os.makedirs(os.path.dirname(path), exist_ok=True)
with open(path, "w", encoding="utf-8") as fh:
fh.write(content)
def _csharp_template(name: str) -> str:
return (
"using UnityEngine;\n\n"
f"public class {name} : MonoBehaviour\n"
"{{\n"
" void Start()\n"
" {{\n"
" Debug.Log(\"{name} ready\");\n"
" }}\n\n"
" void Update()\n"
" {{\n"
" }}\n"
"}}\n"
).format(name=name)
def _csharp_player_controller(game_name: str, game_type: str) -> str:
return (
"using UnityEngine;\n\n"
f"/// \n"
f"/// Auto-generated PlayerController for {game_name} ({game_type}).\n"
"/// \n"
"public class PlayerController : MonoBehaviour\n"
"{\n"
" public float moveSpeed = 5f;\n"
" public float rotateSpeed = 720f;\n"
" private Rigidbody rb;\n\n"
" void Awake()\n"
" {\n"
" rb = GetComponent();\n"
" }\n\n"
" void Start()\n"
" {\n"
" Debug.Log(\"PlayerController ready\");\n"
" }\n\n"
" void Update()\n"
" {\n"
" float h = Input.GetAxisRaw(\"Horizontal\");\n"
" float v = Input.GetAxisRaw(\"Vertical\");\n"
" Vector3 dir = new Vector3(h, 0f, v).normalized;\n"
" if (dir.magnitude > 0.01f)\n"
" {\n"
" Quaternion toRot = Quaternion.LookRotation(dir, Vector3.up);\n"
" transform.rotation = Quaternion.RotateTowards(\n"
" transform.rotation, toRot, rotateSpeed * Time.deltaTime);\n"
" }\n"
" if (rb != null)\n"
" {\n"
" rb.MovePosition(transform.position + dir * moveSpeed * Time.deltaTime);\n"
" }\n"
" }\n"
"}\n"
)
def get_registry():
"""Return the real tool registry if available, else the fallback."""
if _TOOL_REGISTRY is not None:
return _TOOL_REGISTRY
global _FALLBACK_REGISTRY
try:
return _FALLBACK_REGISTRY # type: ignore[name-defined]
except NameError:
_FALLBACK_REGISTRY = _build_fallback_registry()
return _FALLBACK_REGISTRY
def get_tool_result_class():
"""Return the real ToolResult class if available, else our fallback."""
return _ToolResult or ToolResult
def build_tool_catalog() -> str:
"""Return a plain-text catalog of all tools (for the system prompt)."""
reg = get_registry()
lines: List[str] = []
for name in reg.all_names():
try:
tool = reg.get(name)
schema = getattr(tool, "input_schema", {}) or {}
params = ", ".join(schema.keys()) if schema else "none"
doc = (getattr(tool, "__doc__", "") or "").strip().split("\n")[0][:80]
except Exception:
params, doc = "?", ""
lines.append(f" - {name}({params}): {doc}")
return "\n".join(lines)
def run_tool(tool_name: str, arguments: Dict[str, Any], settings: Any, transport: Any) -> ToolResult:
"""Execute a registered tool by name with ``arguments``.
``settings`` and ``transport`` are passed through to the tool constructor
/ function signature when supported. Unknown tools return a failed
:class:`ToolResult`.
"""
reg = get_registry()
tool = reg.get(tool_name)
if tool is None:
return ToolResult(success=False, message=f"Unknown tool: {tool_name}",
error="unknown tool")
try:
# Real registry tools are classes instantiated with (settings, transport).
if _TOOL_REGISTRY is not None and isinstance(tool, type):
try:
instance = tool(settings, transport)
return instance.run(**arguments)
except TypeError:
# Fallback: callable signature.
return tool(settings=settings, transport=transport, **arguments) # type: ignore[misc]
# Fallback registry tools expose a .run(**kwargs) bound method.
if hasattr(tool, "run"):
return tool.run(settings=settings, transport=transport, **arguments)
# Plain callable.
return tool(settings=settings, transport=transport, **arguments) # type: ignore[misc]
except TypeError as e:
# Try without settings/transport kwargs (some tools don't accept them).
try:
if hasattr(tool, "run"):
return tool.run(**arguments)
return tool(**arguments) # type: ignore[misc]
except Exception as e2:
return ToolResult(success=False, message=f"{e2}", error=str(e2))
except Exception as e:
return ToolResult(success=False, message=str(e), error=str(e))
# ===========================================================================
# SETTINGS (configurable, env-aware)
# ===========================================================================
class Settings:
"""Lightweight settings container; mirrors the package Settings if present."""
def __init__(self, output_dir: Optional[str] = None):
self.output_dir = os.path.abspath(output_dir or "./unity_projects")
os.makedirs(self.output_dir, exist_ok=True)
self.model = os.environ.get("ZAI_MODEL", "glm-4.5-flash")
self.base_url = os.environ.get("ZAI_BASE_URL", "https://open.bigmodel.cn/api/paas/v4")
self.temperature = float(os.environ.get("ZAI_TEMPERATURE", "0.2"))
self.max_tokens = int(os.environ.get("ZAI_MAX_TOKENS", "4096"))
def to_dict(self) -> Dict[str, Any]:
return {
"output_dir": self.output_dir,
"model": self.model,
"base_url": self.base_url,
"temperature": self.temperature,
"max_tokens": self.max_tokens,
}
# ===========================================================================
# PRESETS
# ===========================================================================
PRESETS: Dict[str, Dict[str, Any]] = {
"minimal": {
"title": "Minimal Cube Demo",
"description": "A single cube you can move with WASD. Perfect starter.",
"complexity": 1,
"estimated_files": 4,
},
"fps_arena": {
"title": "FPS Arena",
"description": "First-person shooter in a small arena with targets.",
"complexity": 4,
"estimated_files": 12,
},
"open_world_city": {
"title": "Open World City",
"description": "Procedural city blocks with a drivable car and traffic.",
"complexity": 5,
"estimated_files": 30,
},
"racing": {
"title": "Arcade Racing",
"description": "Lap-based racing with checkpoints and a speedometer.",
"complexity": 3,
"estimated_files": 10,
},
"platformer": {
"title": "2.5D Platformer",
"description": "Side-scrolling platformer with collectibles and enemies.",
"complexity": 3,
"estimated_files": 11,
},
"puzzle": {
"title": "Match-3 Puzzle",
"description": "Grid-based match-3 puzzle with score and timers.",
"complexity": 2,
"estimated_files": 7,
},
"tower_defense": {
"title": "Tower Defense",
"description": "Place towers along a path to stop waves of enemies.",
"complexity": 4,
"estimated_files": 16,
},
"endless_runner": {
"title": "Endless Runner",
"description": "Auto-runner with obstacles and increasing difficulty.",
"complexity": 2,
"estimated_files": 8,
},
}
# ===========================================================================
# BANNER / LOGO
# ===========================================================================
LOGO = r"""
__ _ _ _ __ _ _ _
\ \|| || || |___ / /_ | || | __ _ | |__ _____ __
\ | _ || / -_) _|| __ | / _` | | '_ \ / _ \ \/ /
\|_||_||_|_\___|\__||_||_| \__,_| |_.__/ \___/> <
/_/\_\
"""
def print_banner() -> None:
"""Print the animated startup banner."""
width = term_width()
# Top rule
print(C.B_MAGENTA + Box.DH * width + C.RESET)
for line in LOGO.splitlines():
print(C.B_CYAN + line.center(width) + C.RESET)
sub = f"{C.BOLD}{C.B_WHITE}Unity Game Agent{C.RESET} {DIM}·{C.RESET} {C.B_YELLOW}Interactive CLI{C.RESET}"
print(sub.center(width + 6))
info = (f"{C.GREEN}{Box.STAR}{C.RESET} Zero-dependency "
f"{C.GREEN}{Box.STAR}{C.RESET} Standard-library only "
f"{C.GREEN}{Box.STAR}{C.RESET} Z.ai powered")
print(info.center(width + 6))
print(C.B_MAGENTA + Box.DH * width + C.RESET)
# ===========================================================================
# DISPLAY HELPERS
# ===========================================================================
def ok(msg: str) -> None:
print(f" {C.BOLD}{C.GREEN}{Box.CHECK}{C.RESET} {C.GREEN}{msg}{C.RESET}")
def err(msg: str) -> None:
print(f" {C.BOLD}{C.RED}{Box.CROSS_X}{C.RESET} {C.RED}{msg}{C.RESET}")
def warn(msg: str) -> None:
print(f" {C.BOLD}{C.YELLOW}!{C.RESET} {C.YELLOW}{msg}{C.RESET}")
def info(msg: str) -> None:
print(f" {C.BLUE}{Box.BULLET}{C.RESET} {C.BLUE}{msg}{C.RESET}")
def dim(msg: str) -> None:
print(f" {DIM}{msg}{C.RESET}")
def user_echo(msg: str) -> None:
print(f" {C.CYAN}{Box.ARROW}{C.RESET} {C.CYAN}{msg}{C.RESET}")
def agent_echo(msg: str) -> None:
print(f" {C.MAGENTA}{Box.DIAMOND}{C.RESET} {C.MAGENTA}{msg}{C.RESET}")
def section(title: str, color: str = C.B_CYAN) -> None:
print()
print(divider(title, color=color))
def prompt_label() -> str:
"""Return the colored ``game>`` prompt string."""
return f"{C.BOLD}{C.CYAN}game{C.RESET}{C.B_MAGENTA}>{C.RESET} "
# ===========================================================================
# FILE-TREE RENDERING
# ===========================================================================
def render_file_tree(root: str, max_depth: int = 5) -> str:
"""Render an ASCII file tree under ``root``."""
if not os.path.isdir(root):
return f"{C.RED}No such directory: {root}{C.RESET}"
lines: List[str] = []
base_name = os.path.basename(os.path.normpath(root)) or root
lines.append(f"{C.B_BLUE}{Box.DIAMOND} {base_name}/{C.RESET}")
def walk(path: str, prefix: str, depth: int) -> None:
if depth > max_depth:
return
try:
entries = sorted(os.listdir(path))
except OSError:
return
dirs = [e for e in entries if os.path.isdir(os.path.join(path, e))]
files = [e for e in entries if not os.path.isdir(os.path.join(path, e))]
# Skip noisy / irrelevant dirs
skip = {".git", "__pycache__", ".idea", "Library", "Temp", "Obj", "Build"}
dirs = [d for d in dirs if d not in skip]
all_items = [(d, True) for d in dirs] + [(f, False) for f in files]
for i, (name, is_dir) in enumerate(all_items):
last = i == len(all_items) - 1
connector = Box.BT if last else Box.TT
child = os.path.join(path, name)
if is_dir:
lines.append(f"{prefix}{DIM}{connector}{C.RESET} {C.B_BLUE}{name}/{C.RESET}")
extension = " " if last else f"{DIM}{Box.V}{C.RESET} "
walk(child, prefix + extension, depth + 1)
else:
color = _file_color(name)
lines.append(f"{prefix}{DIM}{connector}{C.RESET} {color}{name}{C.RESET}")
walk(root, "", 1)
return "\n".join(lines)
def _file_color(name: str) -> str:
lower = name.lower()
if lower.endswith(".cs"):
return C.B_CYAN
if lower.endswith((".unity",)):
return C.B_GREEN
if lower.endswith((".prefab",)):
return C.B_YELLOW
if lower.endswith((".mat", ".asset", ".shader")):
return C.MAGENTA
if lower.endswith((".json",)):
return C.YELLOW
if lower.endswith((".md", ".txt")):
return C.WHITE
if lower.endswith((".png", ".jpg", ".jpeg", ".tga", ".psd")):
return C.B_MAGENTA
if lower.endswith((".wav", ".mp3", ".ogg")):
return C.BLUE
return C.WHITE
# ===========================================================================
# STATISTICS
# ===========================================================================
def collect_stats(root: str) -> Dict[str, Any]:
"""Walk ``root`` and return aggregate statistics."""
stats: Dict[str, Any] = {
"total_files": 0,
"total_dirs": 0,
"total_bytes": 0,
"by_ext": {},
"scripts": 0,
"scenes": 0,
"prefabs": 0,
"code_lines": 0,
}
if not os.path.isdir(root):
return stats
for r, d, fs in os.walk(root):
# skip noisy dirs
d[:] = [x for x in d if x not in {".git", "__pycache__", "Library", "Temp", "Obj"}]
stats["total_dirs"] += len(d)
for f in fs:
fp = os.path.join(r, f)
try:
size = os.path.getsize(fp)
except OSError:
size = 0
stats["total_files"] += 1
stats["total_bytes"] += size
ext = os.path.splitext(f)[1].lower() or "(none)"
stats["by_ext"][ext] = stats["by_ext"].get(ext, 0) + 1
if ext == ".cs":
stats["scripts"] += 1
try:
with open(fp, encoding="utf-8", errors="replace") as fh:
stats["code_lines"] += sum(1 for _ in fh)
except OSError:
pass
elif ext == ".unity":
stats["scenes"] += 1
elif ext == ".prefab":
stats["prefabs"] += 1
return stats
def human_size(n: int) -> str:
for unit in ("B", "KB", "MB", "GB", "TB"):
if n < 1024 or unit == "TB":
return f"{n:.1f} {unit}" if unit != "B" else f"{n} B"
n /= 1024
return f"{n} TB"
# ===========================================================================
# MAIN CLI
# ===========================================================================
class UnityCLI:
"""The interactive REPL driving the Unity game agent."""
HISTORY_FILE = ".unity_agent_history.json"
CONFIG_FILE = ".unity_agent_config.json"
def __init__(self, settings: Optional[Settings] = None):
self.settings = settings or Settings()
self.transport = None # Set if the real transport is available.
try:
from unity_agent.transport.unity_transport import UnityTransport # type: ignore
self.transport = UnityTransport(self.settings)
except Exception:
self.transport = None
self.api_key = self._resolve_api_key()
self.llm: Optional[LLMClient] = None
if self.api_key:
self.llm = LLMClient(self.api_key, self.settings.model, self.settings.base_url)
self.messages: List[Dict[str, str]] = []
self.history: List[Dict[str, str]] = [] # {role, content, ts}
self.current_project: Optional[str] = None
self.running = True
self.tool_call_count = 0
self.start_time = time.time()
self._init_system_prompt()
# -- setup --------------------------------------------------------------
def _resolve_api_key(self) -> str:
key = os.environ.get("ZAI_API_KEY", "").strip()
if key:
return key
env_path = os.path.join(os.getcwd(), ".env")
if os.path.exists(env_path):
try:
with open(env_path, encoding="utf-8") as fh:
for line in fh:
if line.strip().startswith("ZAI_API_KEY="):
return line.split("=", 1)[1].strip().strip('"').strip("'")
except OSError:
pass
return ""
def _init_system_prompt(self) -> None:
catalog = build_tool_catalog()
preset_lines = "\n".join(f" - {k}: {v['title']}" for k, v in PRESETS.items())
base = _SYSTEM_PROMPT or _DEFAULT_SYSTEM_PROMPT
full = (
base.strip()
+ "\n\n# AVAILABLE TOOLS\n" + catalog
+ "\n\n# AVAILABLE PRESETS\n" + preset_lines
+ "\n\n# OUTPUT FORMAT\n"
+ "When you need to use a tool, emit a fenced block:\n"
+ "```tool_call\n"
+ '{"name": "", "arguments": { ... }}\n'
+ "```\n"
+ "Then briefly explain what you did in plain prose.\n"
+ "When the task is fully complete, include the literal token TASK COMPLETE.\n"
)
self.messages = [{"role": "system", "content": full}]
self.system_prompt = full
# -- entry --------------------------------------------------------------
def run(self) -> None:
print_banner()
self._print_status()
self._maybe_prompt_api_key()
self._repl()
def _print_status(self) -> None:
lines = [
f"{C.B_CYAN}Model{C.RESET} : {C.WHITE}{self.settings.model}{C.RESET}",
f"{C.B_CYAN}Endpoint{C.RESET} : {C.WHITE}{self.settings.base_url}{C.RESET}",
f"{C.B_CYAN}Output{C.RESET} : {C.GREEN}{self.settings.output_dir}{C.RESET}",
f"{C.B_CYAN}Tools{C.RESET} : {C.WHITE}{len(get_registry().all_names())} registered{C.RESET}",
f"{C.B_CYAN}API Key{C.RESET} : "
+ (f"{C.GREEN}set ({self.api_key[:6]}...{self.api_key[-4:]}){C.RESET}"
if self.api_key else f"{C.RED}missing{C.RESET}"),
]
print(panel(lines, title="STATUS", border_color=C.CYAN, title_color=C.B_CYAN, rounded=True))
def _maybe_prompt_api_key(self) -> None:
if self.api_key:
return
print()
warn("No ZAI_API_KEY found in env or .env file.")
info("Get a free key at https://z.ai")
try:
key = input(f" {C.YELLOW}Paste your Z.ai API key (or blank to skip): {C.RESET}").strip()
except (EOFError, KeyboardInterrupt):
print()
return
if key:
self.api_key = key
self.llm = LLMClient(key, self.settings.model, self.settings.base_url)
ok("API key accepted for this session.")
save = input(f" {DIM}Save to .env for future sessions? [y/N]: {C.RESET}").strip().lower()
if save in ("y", "yes"):
try:
with open(os.path.join(os.getcwd(), ".env"), "a", encoding="utf-8") as fh:
fh.write(f"\nZAI_API_KEY={key}\n")
ok("Saved to .env")
except OSError as e:
err(f"Could not save .env: {e}")
# -- REPL ---------------------------------------------------------------
def _repl(self) -> None:
print()
info("Type a game idea, or " + C.BOLD + "/help" + C.RESET + " for commands.")
print()
while self.running:
try:
raw = input(prompt_label())
except (EOFError, KeyboardInterrupt):
print()
self.cmd_exit()
break
line = raw.strip()
if not line:
continue
if line.startswith("/"):
self._dispatch_slash(line)
continue
self._handle_chat(line)
def _dispatch_slash(self, line: str) -> None:
parts = line.split(None, 1)
cmd = parts[0].lower()
args = parts[1].strip() if len(parts) > 1 else ""
handler = {
"/help": self.cmd_help,
"/?": self.cmd_help,
"/tools": self.cmd_tools,
"/presets": self.cmd_presets,
"/generate": self.cmd_generate,
"/open": self.cmd_open,
"/scene": self.cmd_scene,
"/clear": self.cmd_clear,
"/config": self.cmd_config,
"/history": self.cmd_history,
"/export": self.cmd_export,
"/test": self.cmd_test,
"/stats": self.cmd_stats,
"/new": self.cmd_new,
"/add": self.cmd_add,
"/script": self.cmd_script,
"/run": self.cmd_run,
"/exit": self.cmd_exit,
"/quit": self.cmd_exit,
}.get(cmd)
if handler is None:
err(f"Unknown command: {cmd}. Type /help for the command list.")
return
try:
handler(args)
except Exception as e: # pragma: no cover
err(f"Command failed: {e}")
# -- slash commands -----------------------------------------------------
def cmd_help(self, args: str) -> None:
section("HELP — COMMANDS", color=C.B_CYAN)
groups = [
("Getting Around", C.B_CYAN, [
("/help", "Show this help with colored categories"),
("/tools", "List all available tools in a colored table"),
("/presets", "List built-in game presets"),
("/clear", "Clear the conversation history"),
("/exit", "Quit the Unity Agent CLI"),
]),
("Project Management", C.B_GREEN, [
("/new ", "Start a new game project (preset or custom)"),
("/generate ", "Quick-generate a game from a preset"),
("/open ", "Set the project output directory"),
("/scene", "Show the current project file tree"),
("/stats", "Show project statistics"),
("/export", "Export the current project as a .zip"),
("/run", "Show instructions to open the project in Unity"),
]),
("Development", C.B_YELLOW, [
("/add ", "Call a specific tool manually"),
("/script ", "Write / edit a custom C# script"),
("/test", "Run QA checks on the current project"),
]),
("Session", C.B_MAGENTA, [
("/config", "Show / edit CLI configuration"),
("/history", "Show conversation history"),
]),
]
for title, color, cmds in groups:
print(f"\n {color}{C.BOLD}{title}{C.RESET}")
for name, desc in cmds:
print(f" {C.BOLD}{C.WHITE}{name:<22}{C.RESET} {DIM}{desc}{C.RESET}")
print()
info("Or just type a request in plain English, e.g.:")
print(f" {DIM}> create a 3d platformer with double-jump and collectibles{C.RESET}")
print()
def cmd_tools(self, args: str) -> None:
section("TOOLS", color=C.B_GREEN)
reg = get_registry()
names = reg.all_names()
if not names:
warn("No tools registered.")
return
headers = ["#", "Tool", "Parameters", "Description"]
rows: List[List[str]] = []
for i, name in enumerate(names, 1):
tool = reg.get(name)
schema = getattr(tool, "input_schema", {}) or {}
params = ", ".join(schema.keys()) if schema else "—"
doc = (getattr(tool, "__doc__", "") or "").strip().split("\n")[0]
if not doc:
doc = "(no description)"
rows.append([
C.B_BLACK + str(i) + C.RESET,
C.B_CYAN + name + C.RESET,
C.YELLOW + params + C.RESET,
C.WHITE + doc + C.RESET,
])
colors = [C.B_BLACK, C.B_CYAN, C.YELLOW, C.WHITE]
print(table(headers, rows, colors=colors, header_color=C.B_GREEN, border_color=DIM))
print()
info(f"{C.BOLD}{len(names)}{C.RESET} tools available. Use " + C.BOLD + "/add " + C.RESET + " to call one.")
def cmd_presets(self, args: str) -> None:
section("GAME PRESETS", color=C.B_YELLOW)
headers = ["Preset", "Title", "Complexity", "Files", "Description"]
rows: List[List[str]] = []
for key, meta in PRESETS.items():
stars = C.B_YELLOW + "★" * meta["complexity"] + C.RESET + DIM + "★" * (5 - meta["complexity"]) + C.RESET
rows.append([
C.B_CYAN + key + C.RESET,
C.B_WHITE + meta["title"] + C.RESET,
stars,
C.YELLOW + str(meta["estimated_files"]) + C.RESET,
C.WHITE + meta["description"] + C.RESET,
])
colors = [C.B_CYAN, C.B_WHITE, C.B_YELLOW, C.YELLOW, C.WHITE]
print(table(headers, rows, colors=colors, header_color=C.B_YELLOW, border_color=DIM))
print()
info("Use " + C.BOLD + "/generate " + C.RESET + " to scaffold one.")
def cmd_generate(self, args: str) -> None:
preset = args.strip()
if not preset:
err("Usage: /generate (e.g. /generate open_world_city)")
info("Available presets: " + ", ".join(C.B_CYAN + k + C.RESET for k in PRESETS))
return
if preset not in PRESETS:
err(f"Unknown preset: {preset}")
info("Available: " + ", ".join(PRESETS.keys()))
return
meta = PRESETS[preset]
section(f"GENERATE — {meta['title']}", color=C.B_GREEN)
info(f"Preset: {C.BOLD}{preset}{C.RESET}")
info(f"Detail: {meta['description']}")
info(f"Expect: ~{meta['estimated_files']} files, complexity {meta['complexity']}/5")
game_name = f"Game_{preset}"
with ProgressBar(meta["estimated_files"] + 2, label="Generating", color=C.B_GREEN, fill_color=C.B_GREEN) as bar:
bar.update(1, "Setting up project")
r = run_tool("setup_unity_project",
{"project_name": game_name, "game_type": preset},
self.settings, self.transport)
bar.update(2, "Generating scripts")
if r.success:
ok("Project skeleton created.")
self.current_project = os.path.join(self.settings.output_dir, game_name)
else:
err(f"Setup failed: {r.message}")
bar.finish()
return
bar.update(3, "Generating game content")
r2 = run_tool("generate_complete_game",
{"game_name": game_name, "game_type": preset},
self.settings, self.transport)
for i in range(4, meta["estimated_files"] + 2):
bar.update(i, f"Writing asset {i - 3}")
time.sleep(0.05)
bar.update(meta["estimated_files"] + 2, "Finalizing")
if r2.success:
ok(f"Game generated: {self.current_project}")
self._record_history("system", f"Generated {preset} game at {self.current_project}")
print()
print(panel(
[
f"{C.GREEN}{Box.CHECK}{C.RESET} {C.BOLD}{meta['title']}{C.RESET}",
f"{DIM}Location:{C.RESET} {self.current_project}",
f"{DIM}Files:{C.RESET} {len(r2.files)}",
"",
f"{C.B_YELLOW}Next:{C.RESET} type " + C.BOLD + "/run" + C.RESET + " for Unity instructions,",
" or " + C.BOLD + "/scene" + C.RESET + " to browse the file tree.",
],
title="GENERATION COMPLETE",
border_color=C.B_GREEN,
title_color=C.B_GREEN,
rounded=True,
))
else:
err(f"Generation failed: {r2.message}")
def cmd_open(self, args: str) -> None:
if not args.strip():
err("Usage: /open ")
info(f"Current output dir: {self.settings.output_dir}")
return
path = os.path.abspath(os.path.expanduser(args.strip()))
try:
os.makedirs(path, exist_ok=True)
except OSError as e:
err(f"Cannot create/use directory {path}: {e}")
return
self.settings.output_dir = path
if self.transport is not None:
try:
from unity_agent.transport.unity_transport import UnityTransport # type: ignore
self.transport = UnityTransport(self.settings)
except Exception:
pass
ok(f"Output directory set to {path}")
self.current_project = None
def cmd_scene(self, args: str) -> None:
section("PROJECT FILE TREE", color=C.B_BLUE)
target = self.current_project or self.settings.output_dir
if not os.path.isdir(target):
err(f"No project directory yet: {target}")
info("Try " + C.BOLD + "/generate " + C.RESET + " or " + C.BOLD + "/new " + C.RESET + " first.")
return
print(render_file_tree(target))
def cmd_clear(self, args: str) -> None:
self.messages = [{"role": "system", "content": self.system_prompt}]
ok("Conversation cleared. History retained — use " + C.BOLD + "/history" + C.RESET + " to view it.")
def cmd_config(self, args: str) -> None:
section("CONFIGURATION", color=C.B_MAGENTA)
d = self.settings.to_dict()
rows = [
[C.B_CYAN + "output_dir" + C.RESET, C.GREEN + d["output_dir"] + C.RESET],
[C.B_CYAN + "model" + C.RESET, C.WHITE + d["model"] + C.RESET],
[C.B_CYAN + "base_url" + C.RESET, C.WHITE + d["base_url"] + C.RESET],
[C.B_CYAN + "temperature" + C.RESET, C.YELLOW + str(d["temperature"]) + C.RESET],
[C.B_CYAN + "max_tokens" + C.RESET, C.YELLOW + str(d["max_tokens"]) + C.RESET],
[C.B_CYAN + "api_key" + C.RESET,
(C.GREEN + f"{self.api_key[:6]}...{self.api_key[-4:]}" + C.RESET)
if self.api_key else C.RED + "(not set)" + C.RESET],
]
print(table(["Key", "Value"], rows, colors=[C.B_CYAN, C.WHITE],
header_color=C.B_MAGENTA, border_color=DIM))
print()
info("Edit by setting environment variables (ZAI_MODEL, ZAI_BASE_URL, ...)")
info("or run: " + C.BOLD + "/config set " + C.RESET)
if args.strip().lower().startswith("set"):
self._config_set(args.strip()[3:].strip())
def _config_set(self, args: str) -> None:
parts = args.split(None, 1)
if len(parts) != 2:
err("Usage: /config set ")
return
key, value = parts
key = key.lower()
if key in ("output_dir", "output"):
self.settings.output_dir = os.path.abspath(os.path.expanduser(value))
os.makedirs(self.settings.output_dir, exist_ok=True)
ok(f"output_dir = {self.settings.output_dir}")
elif key in ("model",):
self.settings.model = value
if self.llm:
self.llm.model = value
ok(f"model = {value}")
elif key in ("base_url", "baseurl", "endpoint"):
self.settings.base_url = value.rstrip("/")
if self.llm:
self.llm.base_url = self.settings.base_url
ok(f"base_url = {self.settings.base_url}")
elif key in ("temperature", "temp"):
try:
self.settings.temperature = float(value)
ok(f"temperature = {self.settings.temperature}")
except ValueError:
err("temperature must be a number")
elif key in ("max_tokens", "maxtokens"):
try:
self.settings.max_tokens = int(value)
ok(f"max_tokens = {self.settings.max_tokens}")
except ValueError:
err("max_tokens must be an integer")
else:
err(f"Unknown config key: {key}")
def cmd_history(self, args: str) -> None:
section("CONVERSATION HISTORY", color=C.B_MAGENTA)
if not self.history:
warn("No conversation history yet.")
return
headers = ["#", "Role", "Time", "Snippet"]
rows: List[List[str]] = []
for i, entry in enumerate(self.history, 1):
role = entry.get("role", "?")
ts = entry.get("ts", "")
content = entry.get("content", "")
snippet = content.replace("\n", " ")[:60]
if len(content) > 60:
snippet += "…"
role_color = {
"user": C.CYAN,
"assistant": C.MAGENTA,
"system": C.B_BLUE,
"tool": C.GREEN,
}.get(role, C.WHITE)
rows.append([
C.B_BLACK + str(i) + C.RESET,
role_color + role + C.RESET,
DIM + ts + C.RESET,
C.WHITE + snippet + C.RESET,
])
print(table(headers, rows, colors=[C.B_BLACK, C.WHITE, DIM, C.WHITE],
header_color=C.B_MAGENTA, border_color=DIM))
print()
info(f"{C.BOLD}{len(self.history)}{C.RESET} entries. Use " + C.BOLD + "/clear" + C.RESET + " to reset the live conversation.")
def cmd_export(self, args: str) -> None:
section("EXPORT PROJECT", color=C.B_YELLOW)
target = self.current_project or self.settings.output_dir
if not os.path.isdir(target):
err(f"No project to export: {target}")
return
base_name = os.path.basename(os.path.normpath(target)) or "unity_project"
zip_name = args.strip() or base_name + ".zip"
if not zip_name.endswith(".zip"):
zip_name += ".zip"
zip_path = os.path.abspath(zip_name)
info(f"Zipping {target} → {zip_path}")
skip_dirs = {".git", "__pycache__", "Library", "Temp", "Obj", "Build"}
file_count = 0
total_bytes = 0
with ProgressBar(100, label="Exporting", color=C.B_YELLOW, fill_color=C.B_YELLOW) as bar:
with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
for r, d, fs in os.walk(target):
d[:] = [x for x in d if x not in skip_dirs]
for f in fs:
full = os.path.join(r, f)
arc = os.path.relpath(full, target)
zf.write(full, arc)
file_count += 1
total_bytes += os.path.getsize(full)
bar.update(100)
ok(f"Exported {file_count} files ({human_size(total_bytes)}) → {zip_path}")
def cmd_test(self, args: str) -> None:
section("QA CHECKS", color=C.B_RED)
target = self.current_project or self.settings.output_dir
if not os.path.isdir(target):
err(f"No project to test: {target}")
return
info(f"Running QA checks on {target}...")
spinner = Spinner("Analyzing", color=C.RED, interval=0.07)
spinner.start()
time.sleep(0.4)
result = run_tool("run_qa_checks", {}, self.settings, self.transport)
spinner.stop()
if result.success:
ok(result.message)
else:
err(result.message)
issues = []
if isinstance(result.data, dict):
issues = result.data.get("issues", []) or []
if issues:
warn(f"{len(issues)} issue(s) found:")
for i, issue in enumerate(issues, 1):
print(f" {C.YELLOW}{i}.{C.RESET} {issue}")
else:
ok("No issues detected.")
# Extra heuristic checks
self._extra_qa(target)
def _extra_qa(self, target: str) -> None:
extra: List[str] = []
has_readme = os.path.exists(os.path.join(target, "README.md"))
has_manifest = os.path.exists(os.path.join(target, "Packages", "manifest.json"))
has_projectversion = os.path.exists(os.path.join(target, "ProjectSettings", "ProjectVersion.txt"))
if not has_readme:
extra.append("missing README.md")
if not has_manifest:
extra.append("missing Packages/manifest.json")
if not has_projectversion:
extra.append("missing ProjectSettings/ProjectVersion.txt")
if extra:
warn("Project hygiene:")
for e in extra:
print(f" {C.YELLOW}-{C.RESET} {e}")
else:
ok("Project hygiene: all marker files present.")
def cmd_stats(self, args: str) -> None:
section("PROJECT STATISTICS", color=C.B_BLUE)
target = self.current_project or self.settings.output_dir
if not os.path.isdir(target):
err(f"No project: {target}")
return
s = collect_stats(target)
rows = [
[C.B_CYAN + "Path" + C.RESET, C.GREEN + target + C.RESET],
[C.B_CYAN + "Files" + C.RESET, C.WHITE + str(s["total_files"]) + C.RESET],
[C.B_CYAN + "Directories" + C.RESET, C.WHITE + str(s["total_dirs"]) + C.RESET],
[C.B_CYAN + "Size" + C.RESET, C.YELLOW + human_size(s["total_bytes"]) + C.RESET],
[C.B_CYAN + "C# Scripts" + C.RESET, C.B_CYAN + str(s["scripts"]) + C.RESET],
[C.B_CYAN + "Scenes" + C.RESET, C.B_GREEN + str(s["scenes"]) + C.RESET],
[C.B_CYAN + "Prefabs" + C.RESET, C.B_YELLOW + str(s["prefabs"]) + C.RESET],
[C.B_CYAN + "Code Lines" + C.RESET, C.MAGENTA + str(s["code_lines"]) + C.RESET],
]
print(table(["Metric", "Value"], rows, colors=[C.B_CYAN, C.WHITE],
header_color=C.B_BLUE, border_color=DIM))
# Extension breakdown
if s["by_ext"]:
print()
info("Files by extension:")
ext_rows = sorted(s["by_ext"].items(), key=lambda kv: -kv[1])
for ext, count in ext_rows[:12]:
bar_len = min(40, count)
bar = C.B_BLUE + "█" * bar_len + C.RESET + DIM + "·" * max(0, 40 - bar_len) + C.RESET
print(f" {C.YELLOW}{ext:<10}{C.RESET} {C.WHITE}{count:>4}{C.RESET} {bar}")
print()
info(f"Session uptime: {self._uptime()}")
def cmd_new(self, args: str) -> None:
game_type = args.strip()
if not game_type:
err("Usage: /new (e.g. /new minimal, /new fps_arena)")
info("Presets: " + ", ".join(PRESETS.keys()))
return
section(f"NEW PROJECT — {game_type}", color=C.B_GREEN)
name = f"Game_{game_type}"
info(f"Creating new {game_type} project as '{name}'...")
r = run_tool("generate_complete_game",
{"game_name": name, "game_type": game_type},
self.settings, self.transport)
if r.success:
self.current_project = os.path.join(self.settings.output_dir, name)
ok(f"Project created at {self.current_project}")
self._record_history("system", f"New {game_type} project at {self.current_project}")
else:
err(f"Failed: {r.message}")
def cmd_add(self, args: str) -> None:
spec = args.strip()
if not spec:
err("Usage: /add [key=value key=value ...]")
info("Example: /add create_script script_name=Enemy")
return
parts = spec.split()
tool_name = parts[0]
rest = parts[1:]
arguments: Dict[str, Any] = {}
for tok in rest:
if "=" in tok:
k, v = tok.split("=", 1)
arguments[k.strip()] = _coerce(v.strip())
else:
arguments[tok] = True
reg = get_registry()
if reg.get(tool_name) is None:
err(f"Unknown tool: {tool_name}")
info("Use " + C.BOLD + "/tools" + C.RESET + " to list available tools.")
return
self._invoke_tool(tool_name, arguments)
def cmd_script(self, args: str) -> None:
name = args.strip()
if not name:
err("Usage: /script ")
return
clean = os.path.basename(name)
if clean.endswith(".cs"):
clean = clean[:-3]
if not clean or not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", clean):
err("Invalid script name (must be a valid C# identifier).")
return
section(f"SCRIPT EDITOR — {clean}.cs", color=C.B_CYAN)
# Try to load existing
existing = ""
if self.current_project:
cand = os.path.join(self.current_project, "Assets", "Scripts", clean + ".cs")
if os.path.exists(cand):
try:
with open(cand, encoding="utf-8") as fh:
existing = fh.read()
except OSError:
pass
if existing:
info(f"Existing script found. Re-displaying; type your new content.")
print(code_block(existing, "csharp", title=clean + ".cs"))
print()
info("Paste your C# code. End with a line containing only " + C.BOLD + "EOF" + C.RESET + ".")
info("Or type " + C.BOLD + "template" + C.RESET + " to insert a starter, " + C.BOLD + "cancel" + C.RESET + " to abort.")
lines: List[str] = []
try:
while True:
ln = input(f" {C.B_BLUE}{Box.V}{C.RESET} ")
if ln.strip() == "EOF":
break
if ln.strip().lower() == "cancel":
warn("Script creation cancelled.")
return
if ln.strip().lower() == "template" and not lines:
lines.extend(_csharp_template(clean).splitlines())
print(code_block("\n".join(lines), "csharp", title=clean + ".cs (template)"))
continue
lines.append(ln)
except (EOFError, KeyboardInterrupt):
print()
warn("Aborted.")
return
code = "\n".join(lines)
if not code.strip():
warn("Empty script; nothing written.")
return
print()
print(code_block(code, "csharp", title=clean + ".cs"))
result = run_tool("create_script", {"script_name": clean, "code": code},
self.settings, self.transport)
if result.success:
ok(f"Script written: {clean}.cs")
if result.files:
info(f"Path: {result.files[0]}")
else:
err(f"Failed to write script: {result.message}")
def cmd_run(self, args: str) -> None:
section("RUN IN UNITY", color=C.B_GREEN)
target = self.current_project or self.settings.output_dir
if not os.path.isdir(target):
err(f"No project to run: {target}")
return
steps = [
f"Open {C.BOLD}Unity Hub{C.RESET}.",
f"Click {C.BOLD}Add{C.RESET} → browse to:",
f" {C.GREEN}{target}{C.RESET}",
f"Select the project, then click {C.BOLD}Open{C.RESET}.",
f"Unity will import assets; press {C.BOLD}Play{C.RESET} ▶ to run.",
]
if sys.platform.startswith("win"):
steps.append(f"Tip: also run {C.CYAN}start \"\" \"{target}\"{C.RESET} to reveal in Explorer.")
elif sys.platform == "darwin":
steps.append(f"Tip: also run {C.CYAN}open \"{target}\"{C.RESET} to reveal in Finder.")
else:
steps.append(f"Tip: also run {C.CYAN}xdg-open \"{target}\"{C.RESET} to reveal in your file manager.")
print(panel([f"{C.GREEN}{i}.{C.RESET} {s}" for i, s in enumerate(steps, 1)],
title="OPEN IN UNITY", border_color=C.B_GREEN, title_color=C.B_GREEN,
rounded=True))
# Try to open the folder
if args.strip().lower() in ("open", "reveal", "folder"):
try:
if sys.platform.startswith("win"):
subprocess.Popen(["explorer", target])
elif sys.platform == "darwin":
subprocess.Popen(["open", target])
else:
subprocess.Popen(["xdg-open", target])
ok("Opened file manager.")
except Exception as e:
warn(f"Could not open file manager: {e}")
def cmd_exit(self, args: str = "") -> None:
self.running = False
print()
uptime = self._uptime()
summary = [
f"{C.BOLD}Session summary{C.RESET}",
f"{DIM}Uptime:{C.RESET} {uptime}",
f"{DIM}Tool calls:{C.RESET} {self.tool_call_count}",
f"{DIM}History entries:{C.RESET} {len(self.history)}",
f"{DIM}Project:{C.RESET} {self.current_project or '(none)'}",
]
print(panel(summary, title="GOODBYE", border_color=C.B_MAGENTA,
title_color=C.B_MAGENTA, rounded=True))
print()
print(f" {C.B_MAGENTA}{Box.STAR} Thanks for using Unity Agent. {Box.STAR}{C.RESET}")
print()
# -- chat ---------------------------------------------------------------
def _handle_chat(self, user_input: str) -> None:
if not self.llm:
err("No LLM client configured. Set ZAI_API_KEY or restart and paste a key.")
return
user_echo(user_input)
self.messages.append({"role": "user", "content": user_input})
self._record_history("user", user_input)
spinner = Spinner("Agent is thinking", color=C.MAGENTA, interval=0.08)
spinner.start()
try:
response = self.llm.chat(self.messages,
temperature=self.settings.temperature,
max_tokens=self.settings.max_tokens)
except LLMError as e:
spinner.stop()
err(f"LLM error: {e}")
return
except Exception as e: # pragma: no cover
spinner.stop()
err(f"Unexpected error: {e}")
return
spinner.stop()
self.messages.append({"role": "assistant", "content": response})
self._record_history("assistant", response)
# Display prose (strip fenced tool_call blocks).
tool_calls = parse_all_tool_calls(response)
prose = response
if tool_calls:
prose = _TOOL_CALL_RE.sub("", prose)
prose = _TAG_RE.sub("", prose)
prose = prose.strip()
if prose:
for block in _split_paragraphs(prose):
self._print_agent_block(block)
# Execute tool calls.
for call in tool_calls:
self._invoke_tool(call["name"], call["arguments"], from_llm=True)
if "TASK COMPLETE" in response.upper():
print()
print(divider("TASK COMPLETE", color=C.B_GREEN))
if self.current_project:
info(f"Project ready: {C.GREEN}{self.current_project}{C.RESET}")
info("Use " + C.BOLD + "/run" + C.RESET + " for Unity instructions.")
def _print_agent_block(self, block: str) -> None:
# Detect C# code fences for syntax highlighting.
fence_re = re.compile(r"```(?:csharp|c#|cs)?\s*\n(.*?)```", re.DOTALL | re.IGNORECASE)
last = 0
found_code = False
for m in fence_re.finditer(block):
before = block[last:m.start()].strip()
if before:
print(f" {C.MAGENTA}{Box.DIAMOND}{C.RESET} {before}")
code = m.group(1).rstrip()
print()
print(code_block(code, "csharp"))
print()
last = m.end()
found_code = True
tail = block[last:].strip()
if tail:
for ln in tail.splitlines():
print(f" {C.MAGENTA}{Box.DIAMOND}{C.RESET} {C.MAGENTA}{ln}{C.RESET}")
if not found_code and not tail:
for ln in block.splitlines():
print(f" {C.MAGENTA}{Box.DIAMOND}{C.RESET} {C.MAGENTA}{ln}{C.RESET}")
def _invoke_tool(self, name: str, arguments: Dict[str, Any], from_llm: bool = False) -> None:
self.tool_call_count += 1
arg_preview = json.dumps(arguments, default=str)
if len(arg_preview) > 140:
arg_preview = arg_preview[:137] + "..."
print()
print(f" {C.BOLD}{C.BLUE}{Box.ARROW}{C.RESET} {C.BOLD}{C.B_CYAN}tool_call{C.RESET} "
f"{C.WHITE}{name}{C.RESET}{DIM}({arg_preview}){C.RESET}")
spinner = Spinner(f"Running {name}", color=C.BLUE, interval=0.07)
spinner.start()
time.sleep(0.25)
result = run_tool(name, arguments, self.settings, self.transport)
spinner.stop()
if result.success:
ok(result.message[:300])
if result.files:
for f in result.files[:8]:
print(f" {DIM}{Box.BULLET}{C.RESET} {C.GREEN}{f}{C.RESET}")
if len(result.files) > 8:
print(f" {DIM}... and {len(result.files) - 8} more{C.RESET}")
else:
err(result.message[:300])
if result.error:
print(f" {C.RED}{result.error}{C.RESET}")
# Feed result back to the LLM.
feedback = (f"Tool '{name}' result: success={result.success}, "
f"message={result.message}")
if result.files:
feedback += f", files={result.files}"
if result.error:
feedback += f", error={result.error}"
self.messages.append({"role": "user", "content": feedback})
self._record_history("tool", feedback)
# If the tool created a project, remember it.
if name in ("generate_complete_game", "setup_unity_project") and result.success:
if result.files:
# Prefer the project root (a dir) over individual files.
for f in result.files:
if os.path.isdir(f):
self.current_project = f
break
else:
self.current_project = self.settings.output_dir
# -- helpers ------------------------------------------------------------
def _record_history(self, role: str, content: str) -> None:
self.history.append({
"role": role,
"content": content,
"ts": datetime.now().strftime("%H:%M:%S"),
})
# Persist a rolling log of the last 200 entries.
try:
path = os.path.join(os.getcwd(), self.HISTORY_FILE)
keep = self.history[-200:]
with open(path, "w", encoding="utf-8") as fh:
json.dump(keep, fh, indent=2, ensure_ascii=False)
except OSError:
pass
def _uptime(self) -> str:
secs = int(time.time() - self.start_time)
h, rem = divmod(secs, 3600)
m, s = divmod(rem, 60)
if h:
return f"{h}h {m}m {s}s"
if m:
return f"{m}m {s}s"
return f"{s}s"
# ===========================================================================
# MODULE-LEVEL HELPERS
# ===========================================================================
def _split_paragraphs(text: str) -> List[str]:
"""Split prose into paragraphs (double-newline separated)."""
parts = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()]
return parts or [text.strip()]
def _coerce(value: str) -> Any:
"""Best-effort coerce a string token to int/float/bool/JSON/str."""
if value.lower() in ("true", "false"):
return value.lower() == "true"
if value.lower() in ("null", "none"):
return None
try:
return int(value)
except ValueError:
pass
try:
return float(value)
except ValueError:
pass
if (value.startswith("{") and value.endswith("}")) or (value.startswith("[") and value.endswith("]")):
try:
return json.loads(value)
except json.JSONDecodeError:
pass
return value
_DEFAULT_SYSTEM_PROMPT = """\
You are Unity Agent, an expert Unity C# game developer assistant.
You help the user design, scaffold, and write code for Unity games.
Always:
- Be concise but complete.
- When a tool is needed, emit a ```tool_call``` block with JSON {"name": ..., "arguments": {...}}.
- After calling a tool, briefly summarize what happened.
- Prefer the existing tools over guessing file paths.
- When the task is fully done, include the literal text TASK COMPLETE.
"""
def interactive_repl() -> None:
"""Entry point — create a Settings + CLI and run the REPL."""
settings = Settings()
cli = UnityCLI(settings)
cli.run()
def main(argv: Optional[List[str]] = None) -> int:
"""Command-line entry point with a few non-interactive shortcuts."""
argv = list(sys.argv[1:] if argv is None else argv)
if not argv:
interactive_repl()
return 0
cmd = argv[0].lower()
settings = Settings()
cli = UnityCLI(settings)
if cmd in ("--help", "-h", "help"):
print_banner()
cli.cmd_help("")
return 0
if cmd in ("tools", "list-tools"):
cli.cmd_tools("")
return 0
if cmd in ("presets", "list-presets"):
cli.cmd_presets("")
return 0
if cmd == "generate":
preset = argv[1] if len(argv) > 1 else ""
cli.cmd_generate(preset)
return 0
# Default: treat the whole arg list as a one-shot prompt.
err(f"Unknown command: {cmd}")
info("Run with no arguments to start the interactive REPL.")
return 1
if __name__ == "__main__":
sys.exit(main())