#!/usr/bin/env python3 """Titan Agent — a small local agent that can actually do things on this computer. It talks to the Titan AI OpenAI-compatible API with your agent key (pa-...) and runs the tools it asks for on this machine, asking you first every time. Quick start: export TITAN_API_KEY=pa-your-key python3 titan-agent.py # local server (127.0.0.1:8787) python3 titan-agent.py --base-url https://YOUR-CLASS-SITE/v1 python3 titan-agent.py --task "open Calculator" python3 titan-agent.py --workdir ~/demo --yes Tools: open_app, open_url, run_shell, read_file, write_file, list_dir. Nothing runs without your approval unless you pass --yes. Safety model: - Every action is printed and needs a y/always before it runs (the main control). - read_file / write_file / list_dir stay inside the working directory (--allow-outside lifts that). - run_shell sets the working directory but can otherwise do anything the shell can, so read the command before approving it, and only use --yes with requests you trust. """ import argparse import base64 import ctypes import ctypes.util import hmac import json import os import platform import secrets import shutil import subprocess import sys import tempfile import time import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path SCREENSHOT_WIDTH = 1024 KEY_CODES = { "return": 36, "enter": 76, "tab": 48, "space": 49, "escape": 53, "esc": 53, "delete": 51, "backspace": 51, "arrowleft": 123, "arrowdown": 125, "arrowright": 124, "arrowup": 126, "home": 115, "end": 119, "pageup": 116, "pagedown": 121, } MODIFIER_FLAGS = { "cmd": 1 << 20, "command": 1 << 20, "shift": 1 << 17, "alt": 1 << 19, "option": 1 << 19, "ctrl": 1 << 18, "control": 1 << 18, } TIMEOUT_SECONDS = 180 MAX_STEPS = 8 OUTPUT_CHARS = 4000 CONFIG_PATH = Path.home() / ".titan-agent.json" DEFAULT_BASE_URL = "http://127.0.0.1:8787/v1" TOOLS = [ { "type": "function", "function": { "name": "open_app", "description": "Open a desktop application on this computer by its name (for example Calculator, Notes, Safari, Preview).", "parameters": { "type": "object", "properties": {"app": {"type": "string", "description": "Application name"}}, "required": ["app"], }, }, }, { "type": "function", "function": { "name": "open_url", "description": "Open a web address in the default browser on this computer.", "parameters": { "type": "object", "properties": {"url": {"type": "string", "description": "Full http(s) URL"}}, "required": ["url"], }, }, }, { "type": "function", "function": { "name": "run_shell", "description": "Run a shell command on this computer inside the working directory and return its output.", "parameters": { "type": "object", "properties": {"command": {"type": "string", "description": "Shell command to run"}}, "required": ["command"], }, }, }, { "type": "function", "function": { "name": "read_file", "description": "Read a text file on this computer (first part only) and return its contents.", "parameters": { "type": "object", "properties": {"path": {"type": "string", "description": "File path"}}, "required": ["path"], }, }, }, { "type": "function", "function": { "name": "write_file", "description": "Create or overwrite a text file on this computer with the given contents.", "parameters": { "type": "object", "properties": { "path": {"type": "string", "description": "File path"}, "content": {"type": "string", "description": "Full text to write"}, }, "required": ["path", "content"], }, }, }, { "type": "function", "function": { "name": "list_dir", "description": "List the files and folders in a directory on this computer.", "parameters": { "type": "object", "properties": {"path": {"type": "string", "description": "Directory path, default ."}}, }, }, }, ] def load_config(): try: document = json.loads(CONFIG_PATH.read_text(encoding="utf-8")) return document if isinstance(document, dict) else {} except (OSError, json.JSONDecodeError): return {} def system_prompt(workdir): return ( "You are Titan Agent, running on the user's own %s computer. You can really act on this " "machine through your tools: open apps, open URLs, run shell commands, read and write files " "inside %s. To make an app do something, prefer command-line automation over guessing at its " "window: osascript/AppleScript (macOS), the app's own CLI, or URL schemes; use non-interactive " "flags (-y/--yes) and never start editors or pagers (no TTY). Use tools to do the job instead " "of only explaining; run one small step at a time, check the result, then continue. The user " "approves or rejects every action, so keep each action small and explain briefly what you are " "doing. If a step fails, read the error and try another way. Finish with a short summary." % (platform.system(), workdir) ) def request_chat(base_url, api_key, model, messages): payload = { "model": model, "messages": messages, "tools": TOOLS, "tool_choice": "auto", "stream": False, } request = urllib.request.Request( base_url.rstrip("/") + "/chat/completions", data=json.dumps(payload).encode("utf-8"), headers={ "Authorization": "Bearer " + api_key, "Content-Type": "application/json", "User-Agent": "TitanAgent/1.0", }, method="POST", ) try: with urllib.request.urlopen(request, timeout=TIMEOUT_SECONDS) as response: return json.loads(response.read().decode("utf-8", "replace")) except urllib.error.HTTPError as error: body = error.read().decode("utf-8", "replace") try: message = json.loads(body).get("error", {}).get("message", body) except ValueError: message = body raise SystemExit("Titan AI error %s: %s" % (error.code, message[:300])) except urllib.error.URLError as error: raise SystemExit("Cannot reach %s (%s). Is the server running?" % (base_url, error.reason)) def inside_workspace(target, workdir, allow_outside): if allow_outside: return True target = Path(target) return target == workdir or workdir in target.parents def resolve_path(raw, workdir, allow_outside): path = Path(str(raw or ".")).expanduser() target = (workdir / path).resolve() if not path.is_absolute() else path.resolve() if not inside_workspace(target, workdir, allow_outside): raise ValueError( "refused: %s is outside the working directory (use --allow-outside to permit)" % target ) return target def truncate(text): text = text or "" if len(text) <= OUTPUT_CHARS: return text return text[:OUTPUT_CHARS] + "\n… (%d more characters)" % (len(text) - OUTPUT_CHARS) def tool_open_app(app): name = str(app or "").strip() if not name: return "error: no application name given" system = platform.system() if system == "Darwin": subprocess.Popen(["open", "-a", name]) return "opened %s with `open -a`" % name if system == "Windows": subprocess.Popen(["cmd", "/c", "start", "", name], shell=False) return "opened %s with `start`" % name binary = shutil.which(name) or shutil.which(name.lower()) if binary: subprocess.Popen([binary], start_new_session=True) return "started %s" % binary subprocess.Popen(["xdg-open", name], start_new_session=True) return "asked xdg-open for %s" % name def tool_open_url(url): value = str(url or "").strip() if not value.startswith(("http://", "https://")): return "error: only http(s) URLs are allowed" system = platform.system() if system == "Darwin": subprocess.Popen(["open", value]) elif system == "Windows": subprocess.Popen(["cmd", "/c", "start", "", value], shell=False) else: subprocess.Popen(["xdg-open", value], start_new_session=True) return "opened %s in the browser" % value def tool_run_shell(command, workdir, timeout): completed = subprocess.run( str(command), shell=True, cwd=str(workdir), capture_output=True, text=True, timeout=timeout ) parts = ["exit code: %d" % completed.returncode] if completed.stdout.strip(): parts.append("stdout:\n" + completed.stdout) if completed.stderr.strip(): parts.append("stderr:\n" + completed.stderr) return truncate("\n".join(parts)) def tool_read_file(path, workdir, allow_outside): target = resolve_path(path, workdir, allow_outside) if not target.is_file(): return "error: %s is not a file" % target return truncate(target.read_text(encoding="utf-8", errors="replace")) def tool_write_file(path, content, workdir, allow_outside): target = resolve_path(path, workdir, allow_outside) target.parent.mkdir(parents=True, exist_ok=True) target.write_text(str(content), encoding="utf-8") return "wrote %d characters to %s" % (len(str(content)), target) def tool_list_dir(path, workdir, allow_outside): target = resolve_path(path, workdir, allow_outside) if not target.is_dir(): return "error: %s is not a directory" % target lines = [] for entry in sorted(target.iterdir())[:200]: kind = "dir " if entry.is_dir() else "file" size = "" if entry.is_dir() else " %d bytes" % entry.stat().st_size lines.append("%s %s%s" % (kind, entry.name, size)) return truncate("\n".join(lines) or "(empty)") def describe(name, arguments): if name == "run_shell": return arguments.get("command", "") if name == "open_app": return arguments.get("app", "") if name == "open_url": return arguments.get("url", "") if name in ("read_file", "list_dir"): return arguments.get("path", ".") if name == "write_file": content = str(arguments.get("content", "")) preview = content[:120].replace("\n", "\\n") return "%s (%d characters) %s" % (arguments.get("path", ""), len(content), preview) return json.dumps(arguments)[:200] _core_graphics = None def core_graphics(): global _core_graphics if _core_graphics is None: if platform.system() != "Darwin": raise RuntimeError("GUI tools only work on macOS for now") path = ctypes.util.find_library("ApplicationServices") or ( "/System/Library/Frameworks/ApplicationServices.framework/ApplicationServices" ) _core_graphics = ctypes.cdll.LoadLibrary(path) return _core_graphics class CGPoint(ctypes.Structure): _fields_ = [("x", ctypes.c_double), ("y", ctypes.c_double)] class CGSize(ctypes.Structure): _fields_ = [("width", ctypes.c_double), ("height", ctypes.c_double)] class CGRect(ctypes.Structure): _fields_ = [("origin", CGPoint), ("size", CGSize)] def main_display_point_size(): try: library = core_graphics() library.CGMainDisplayID.restype = ctypes.c_uint32 library.CGDisplayBounds.restype = CGRect library.CGDisplayBounds.argtypes = [ctypes.c_uint32] bounds = library.CGDisplayBounds(library.CGMainDisplayID()) return float(bounds.size.width), float(bounds.size.height) except Exception: return 0.0, 0.0 def gui_trusted(): try: library = core_graphics() return bool(library.AXIsProcessTrusted()) except Exception: return False def screen_recording_ok(): try: library = core_graphics() if not hasattr(library, "CGPreflightScreenCaptureAccess"): return None library.CGPreflightScreenCaptureAccess.restype = ctypes.c_bool return bool(library.CGPreflightScreenCaptureAccess()) except Exception: return None def responsible_app_name(): bundle_id = os.environ.get("__CFBundleIdentifier", "").strip() if not bundle_id: return "" for folder in ("/Applications", str(Path.home() / "Applications"), "/System/Applications"): base = Path(folder) if not base.is_dir(): continue for bundle in base.glob("*.app"): try: info = (bundle / "Contents" / "Info.plist").read_bytes() except OSError: continue if bundle_id.encode() in info: return bundle.stem return bundle_id def screen_recording_note(): app = responsible_app_name() or "the app running the bridge" return ( "Screen Recording permission is missing for %s, so every screenshot comes back as the wallpaper " "with no windows and GUI mode is blind. Ask the student to open System Settings → Privacy & " "Security → Screen Recording, switch on %s, then quit and reopen it and restart the bridge." % (app, app) ) def gui_permission_note(): return ( "macOS blocked the action. Open System Settings → Privacy & Security → Accessibility " "and enable the terminal app you started the bridge from, then try again." ) def tool_gui_check(): library = core_graphics() library.CGEventCreate.restype = ctypes.c_void_p library.CGEventCreate.argtypes = [ctypes.c_void_p] library.CGEventGetLocation.restype = CGPoint library.CGEventGetLocation.argtypes = [ctypes.c_void_p] library.CGEventCreateMouseEvent.restype = ctypes.c_void_p library.CGEventCreateMouseEvent.argtypes = [ ctypes.c_void_p, ctypes.c_uint32, CGPoint, ctypes.c_uint32, ] library.CGEventPost.argtypes = [ctypes.c_uint32, ctypes.c_void_p] library.CFRelease.argtypes = [ctypes.c_void_p] event = library.CGEventCreate(None) origin = library.CGEventGetLocation(event) library.CFRelease(event) moved = library.CGEventCreateMouseEvent(None, 5, CGPoint(origin.x + 4, origin.y + 4), 0) library.CGEventPost(0, moved) library.CFRelease(moved) time.sleep(0.2) event = library.CGEventCreate(None) after = library.CGEventGetLocation(event) library.CFRelease(event) back = library.CGEventCreateMouseEvent(None, 5, origin, 0) library.CGEventPost(0, back) library.CFRelease(back) works = abs(after.x - origin.x) > 1 or abs(after.y - origin.y) > 1 _output, script_error = run_applescript( 'tell application "System Events" to return name of first application process whose frontmost is true' ) apple_ok = script_error == "" problems = [] if not apple_ok: problems.append( "macOS is blocking automation: grant Accessibility AND Input Monitoring permission to the " "app running the bridge, then restart the bridge" ) if screen_recording_ok() is False: problems.append(screen_recording_note()) verdict = "input ready" if not problems else " | ".join(problems) return ( "trusted=%s screenRecording=%s quartzCursorMoved=%s appleScript=%s | %s" % (gui_trusted(), screen_recording_ok(), works, apple_ok, verdict) ) def tool_screenshot(server): if screen_recording_ok() is False: return "error: %s" % screen_recording_note() with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "screen.jpg" completed = subprocess.run( ["screencapture", "-x", "-t", "jpg", str(path)], capture_output=True, text=True ) if completed.returncode != 0 or not path.exists(): return "error: screencapture failed: %s" % (completed.stderr.strip() or completed.returncode) native = subprocess.run( ["sips", "-g", "pixelWidth", "-g", "pixelHeight", str(path)], capture_output=True, text=True ).stdout width = height = 0 for line in native.splitlines(): if "pixelWidth" in line: width = int(line.split(":")[-1]) if "pixelHeight" in line: height = int(line.split(":")[-1]) subprocess.run( ["sips", "-Z", str(SCREENSHOT_WIDTH), "-s", "formatOptions", "65", str(path)], capture_output=True, text=True, ) scaled = subprocess.run( ["sips", "-g", "pixelWidth", "-g", "pixelHeight", str(path)], capture_output=True, text=True ).stdout image_width = image_height = 0 for line in scaled.splitlines(): if "pixelWidth" in line: image_width = int(line.split(":")[-1]) if "pixelHeight" in line: image_height = int(line.split(":")[-1]) payload = base64.b64encode(path.read_bytes()).decode("ascii") point_width, _point_height = main_display_point_size() reference_width = point_width if point_width > 0 else float(width) server.last_scale = (reference_width / image_width) if image_width else 1.0 server.last_image_size = (image_width, image_height) return { "output": "screenshot %dx%d (image space); screen %dx%d points; coordinates you send are in image space" % (image_width, image_height, int(reference_width), int(point_width and _point_height or height)), "image": "data:image/jpeg;base64," + payload, } def escape_applescript_text(text): return str(text).replace("\\", "\\\\").replace('"', '\\"') def apple_key(keys): combo = [part.strip().lower() for part in str(keys).split("+") if part.strip()] modifiers = [part for part in combo if part in MODIFIER_FLAGS] others = [part for part in combo if part not in MODIFIER_FLAGS] using = "" if modifiers: names = {"cmd": "command down", "command": "command down", "shift": "shift down", "alt": "option down", "option": "option down", "ctrl": "control down", "control": "control down"} using = " using {%s}" % ", ".join(names.get(name, name) for name in modifiers) if not others: return "error: no key given (try: return, tab, escape, cmd+space)" key = others[0] if key in KEY_CODES: command = "key code %d%s" % (KEY_CODES[key], using) elif len(key) == 1: command = 'keystroke "%s"%s' % (escape_applescript_text(key), using) else: return "error: unknown key %r" % key output, error = run_applescript( 'tell application "System Events" to %s' % command ) if output is None: return error return "pressed %s" % keys def tool_click(server, x, y, button="left"): if not gui_trusted(): return gui_permission_note() scale = getattr(server, "last_scale", 1.0) or 1.0 point = CGPoint(float(x) * scale, float(y) * scale) library = core_graphics() library.CGEventCreateMouseEvent.restype = ctypes.c_void_p library.CGEventCreateMouseEvent.argtypes = [ ctypes.c_void_p, ctypes.c_uint32, CGPoint, ctypes.c_uint32, ] library.CGEventPost.argtypes = [ctypes.c_uint32, ctypes.c_void_p] library.CFRelease.argtypes = [ctypes.c_void_p] which = 0 if str(button).lower() != "right" else 1 library.CGEventSetIntegerValueField.argtypes = [ ctypes.c_void_p, ctypes.c_uint32, ctypes.c_int64, ] for event_type in (1 if which == 0 else 3, 2 if which == 0 else 4): event = library.CGEventCreateMouseEvent(None, event_type, point, which) library.CGEventSetIntegerValueField(event, 1, 1) library.CGEventPost(0, event) library.CFRelease(event) return "clicked %s at image (%s, %s)" % (button, x, y) def tool_type_text(text): if not gui_trusted(): return gui_permission_note() value = str(text) if not value: return "error: no text given" saved = subprocess.run(["pbpaste"], capture_output=True).stdout subprocess.run(["pbcopy"], input=value.encode("utf-8")) try: _output, error = run_applescript( 'tell application "System Events" to keystroke "v" using command down' ) time.sleep(0.4) finally: subprocess.run(["pbcopy"], input=saved) if error: return error return "typed %d characters" % len(value) def tool_key(keys): if not gui_trusted(): return gui_permission_note() combo = [part.strip().lower() for part in str(keys).split("+") if part.strip()] flags = 0 key_code = None for part in combo: if part in MODIFIER_FLAGS: flags |= MODIFIER_FLAGS[part] elif part in KEY_CODES: key_code = KEY_CODES[part] elif len(part) == 1: key_code = ord(part.upper()) if key_code is None: return "error: unknown key combination %r (try: return, tab, escape, cmd+space, cmd+c)" % keys library = core_graphics() library.CGEventCreateKeyboardEvent.restype = ctypes.c_void_p library.CGEventCreateKeyboardEvent.argtypes = [ctypes.c_void_p, ctypes.c_uint16, ctypes.c_bool] library.CGEventSetFlags.argtypes = [ctypes.c_void_p, ctypes.c_uint64] library.CGEventPost.argtypes = [ctypes.c_uint32, ctypes.c_void_p] library.CFRelease.argtypes = [ctypes.c_void_p] for pressed in (True, False): event = library.CGEventCreateKeyboardEvent(None, key_code, pressed) library.CGEventSetFlags(event, flags) library.CGEventPost(0, event) library.CFRelease(event) return "pressed %s" % keys def tool_scroll(amount): if not gui_trusted(): return gui_permission_note() library = core_graphics() library.CGEventCreateScrollWheelEvent.restype = ctypes.c_void_p library.CGEventCreateScrollWheelEvent.argtypes = [ ctypes.c_void_p, ctypes.c_uint32, ctypes.c_uint32, ctypes.c_int32, ] library.CGEventPost.argtypes = [ctypes.c_uint32, ctypes.c_void_p] library.CFRelease.argtypes = [ctypes.c_void_p] event = library.CGEventCreateScrollWheelEvent(None, 1, 1, int(amount)) library.CGEventPost(0, event) library.CFRelease(event) return "scrolled by %s" % amount AX_WALKER = """ script ElementWalker property mode : "list" property wanted : 0 property newText : "" property limit : 60 property maxDepth : 5 property n : 0 property output : {} property resultText : "" on walk(el, depth) if depth > maxDepth then return tell application "System Events" repeat with child in (UI elements of el) if mode is "list" and n is greater than or equal to limit then return if resultText is not "" then return try set r to role of child if n is wanted then if mode is "press" then click child set resultText to "pressed " & r return else if mode is "set_text" then try set value of child to newText on error set focused of child to true keystroke newText end try set resultText to "typed into " & r return end if end if set t to "" try set t to title of child end try if t is "" then try set t to (value of child) as text end try end if if t is "" then try set t to description of child end try end if set p to position of child set s to size of child set lineText to (n as text) & " | " & r & " | " & t & " | " & ((item 1 of p) as text) & "," & ((item 2 of p) as text) & " | " & ((item 1 of s) as text) & "x" & ((item 2 of s) as text) set end of output to lineText set n to n + 1 my walk(child, depth + 1) end try end repeat end tell end walk end script """ def ax_script(mode, index=0, text="", limit=60): return ( AX_WALKER + '\nset ElementWalker\'s mode to "%s"\n' % mode + "set ElementWalker's wanted to %d\n" % int(index) + 'set ElementWalker\'s newText to "%s"\n' % str(text).replace('"', '\\"') + "set ElementWalker's limit to %d\n" % int(limit) + """ tell application "System Events" tell (first application process whose frontmost is true) set appName to name try ElementWalker's walk(window 1, 0) on error errText return "error: " & errText end try end tell end tell if ElementWalker's resultText is not "" then return ElementWalker's resultText set AppleScript's text item delimiters to linefeed return appName & linefeed & (ElementWalker's output as text) """ ) def run_applescript(source): completed = subprocess.run( ["osascript", "-e", source], capture_output=True, text=True, timeout=60 ) if completed.returncode != 0: message = (completed.stderr or completed.stdout).strip() if "assistive access" in message or "not allowed" in message: return None, gui_permission_note() return None, "error: %s" % message[:300] return completed.stdout.strip(), "" def tool_ui_tree(limit=60): output, error = run_applescript(ax_script("list", limit=limit)) if output is None: return error return truncate(output) def tool_ui_action(index, action, text=""): action = str(action or "press").lower() if action not in ("press", "set_text"): return "error: action must be press or set_text" output, error = run_applescript(ax_script(action, index=index, text=text)) if output is None: return error return truncate(output) def execute(call, workdir, allow_outside, timeout, server=None): name = call.get("function", {}).get("name", "") try: arguments = json.loads(call.get("function", {}).get("arguments") or "{}") except ValueError: return "error: could not read the tool arguments" if name == "gui_check": if server is None: return "error: GUI tools need the bridge" try: return tool_gui_check() except Exception as error: return "error: %s" % error if name in ("ui_tree", "ui_action"): if server is None: return "error: these tools need the bridge" try: if name == "ui_tree": return tool_ui_tree(arguments.get("limit", 60)) return tool_ui_action( arguments.get("index", 0), arguments.get("action", "press"), arguments.get("text", ""), ) except Exception as error: return "error: %s" % error if name in ("screenshot", "click", "type_text", "key_press", "scroll"): if server is None: return "error: GUI tools need the bridge" try: if name == "screenshot": result = tool_screenshot(server) if isinstance(result, dict): return result return result if name == "click": return tool_click( server, arguments.get("x", 0), arguments.get("y", 0), arguments.get("button", "left"), ) if name == "type_text": return tool_type_text(arguments.get("text", "")) if name == "key_press": return apple_key(arguments.get("keys", "")) if name == "scroll": return tool_scroll(arguments.get("amount", 0)) except Exception as error: return "error: %s" % error return execute_tool(call, workdir, allow_outside, timeout) def execute_tool(call, workdir, allow_outside, timeout): name = call.get("function", {}).get("name", "") try: arguments = json.loads(call.get("function", {}).get("arguments") or "{}") except ValueError: return "error: could not read the tool arguments" try: if name == "open_app": return tool_open_app(arguments.get("app")) if name == "open_url": return tool_open_url(arguments.get("url")) if name == "run_shell": return tool_run_shell(arguments.get("command"), workdir, timeout) if name == "read_file": return tool_read_file(arguments.get("path"), workdir, allow_outside) if name == "write_file": return tool_write_file(arguments.get("content"), arguments.get("path"), workdir, allow_outside) if name == "list_dir": return tool_list_dir(arguments.get("path", "."), workdir, allow_outside) return "error: unknown tool %s" % name except subprocess.TimeoutExpired: return "error: the command took longer than %d seconds and was stopped" % timeout except (OSError, ValueError) as error: return "error: %s" % error def approve(description, state): if state["yes"]: print(" approved automatically (--yes)") return True while True: answer = input(" run? [y/N/a=always] ").strip().lower() if answer in ("y", "yes"): return True if answer in ("a", "always"): state["yes"] = True return True if answer in ("", "n", "no"): return False def run_task(task, state): workdir = state["workdir"] messages = [ {"role": "system", "content": system_prompt(workdir)}, {"role": "user", "content": task}, ] for _ in range(MAX_STEPS): data = request_chat(state["base_url"], state["api_key"], state["model"], messages) choices = data.get("choices") or [] message = choices[0].get("message", {}) if choices else {} calls = message.get("tool_calls") or [] if not calls: text = (message.get("content") or "").strip() provider = data.get("provider") or "" model = data.get("model") or "" print() print(text or "(the model returned no text)") print("\n[%s — %s]" % (provider, model)) return messages.append( {"role": "assistant", "content": message.get("content") or "", "tool_calls": calls} ) for call in calls: name = call.get("function", {}).get("name", "?") try: arguments = json.loads(call.get("function", {}).get("arguments") or "{}") except ValueError: arguments = {} print("\n→ %s: %s" % (name, describe(name, arguments))) if approve(describe(name, arguments), state): result = execute(call, workdir, state["allow_outside"], state["timeout"]) else: result = "the user denied this action; do not retry it without asking" print(" %s" % truncate(result).splitlines()[0][:160]) messages.append({"role": "tool", "tool_call_id": call.get("id"), "content": truncate(result)}) print("\n(stopped after %d steps — ask again to continue)" % MAX_STEPS) class BridgeHandler(BaseHTTPRequestHandler): server_version = "TitanAgent/1.0" def log_message(self, format, *args): return def origin_allowed(self): origin = self.headers.get("Origin", "") if not origin: return False if origin in self.server.allowed_origins: return True return origin.startswith("http://127.0.0.1:") or origin.startswith("http://localhost:") def send_cors(self): origin = self.headers.get("Origin", "") if origin and self.origin_allowed(): self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Access-Control-Allow-Headers", "Content-Type, X-Titan-Pair") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Max-Age", "600") self.send_header("Vary", "Origin") def send_json(self, document, status=200): payload = json.dumps(document).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Cache-Control", "no-store") self.send_cors() self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def paired(self): code = self.headers.get("X-Titan-Pair", "") return bool(code) and hmac.compare_digest(code, self.server.pair_code) def do_OPTIONS(self): self.send_response(204) self.send_cors() self.send_header("Content-Length", "0") self.end_headers() def do_GET(self): path = self.path.split("?", 1)[0] if path == "/ping": self.send_json( { "ok": True, "name": "Titan Agent bridge", "paired": self.paired(), "workdir": str(self.server.workdir), "tools": [ "open_app", "open_url", "run_shell", "read_file", "write_file", "list_dir", "ui_tree", "ui_action", "gui_check", "screenshot", "click", "type_text", "key_press", "scroll", ], "guiReady": gui_trusted(), } ) return self.send_json({"ok": False, "error": "not found"}, status=404) def do_POST(self): path = self.path.split("?", 1)[0] length = int(self.headers.get("Content-Length") or 0) try: document = json.loads(self.rfile.read(length) or b"{}") except ValueError: self.send_json({"ok": False, "error": "body must be JSON"}, status=400) return if path == "/pair": code = str(document.get("code") or "").strip().upper() if hmac.compare_digest(code, self.server.pair_code): self.send_json({"ok": True}) else: self.send_json({"ok": False, "error": "wrong pairing code"}, status=401) return if not self.paired(): self.send_json({"ok": False, "error": "not paired"}, status=401) return if path == "/exec": tool = str(document.get("tool") or "") arguments = document.get("arguments") if isinstance(document.get("arguments"), dict) else {} call = {"function": {"name": tool, "arguments": json.dumps(arguments)}} output = execute( call, self.server.workdir, self.server.allow_outside, self.server.timeout, self.server, ) if isinstance(output, dict): self.send_json( { "ok": True, "output": truncate(output.get("output", "")), "image": output.get("image", ""), } ) return self.send_json({"ok": True, "output": truncate(output)}) return self.send_json({"ok": False, "error": "not found"}, status=404) def serve_bridge(state, port, extra_origins, fixed_code=""): code = fixed_code.strip().upper() or "-".join( ["".join(secrets.choice("ABCDEFGHJKMNPQRSTUVWXYZ23456789") for _ in range(4)) for _ in range(2)] ) site = os.environ.get("TITAN_SITE", "https://titan-ai.me").strip().rstrip("/") host = site.split("://", 1)[-1] allowed = {site, "https://www." + host, "http://" + host, "http://127.0.0.1:8787", "http://localhost:8787"} for origin in extra_origins: if origin.strip(): allowed.add(origin.strip()) server = ThreadingHTTPServer(("127.0.0.1", port), BridgeHandler) server.pair_code = code server.allowed_origins = allowed server.workdir = state["workdir"] server.allow_outside = state["allow_outside"] server.timeout = state["timeout"] server.last_scale = 1.0 server.last_image_size = (0, 0) print("Titan Agent bridge listening on http://127.0.0.1:%d" % port) print("Working directory: %s" % state["workdir"]) print() print("Pairing code: %s" % code) print("Open the site, switch to Code, click 'Connect local agent' and paste this code.") print("Press Ctrl+C to stop.") try: server.serve_forever() except KeyboardInterrupt: print() finally: server.server_close() return 0 def main(): config = load_config() parser = argparse.ArgumentParser(description="Titan Agent — a local agent on your own computer") parser.add_argument("--base-url", default=os.environ.get("TITAN_BASE_URL") or config.get("baseUrl") or DEFAULT_BASE_URL) parser.add_argument("--key", default=os.environ.get("TITAN_API_KEY") or config.get("apiKey") or "") parser.add_argument("--model", default=os.environ.get("TITAN_MODEL") or config.get("model") or "auto") parser.add_argument("--workdir", default=os.environ.get("TITAN_WORKDIR") or os.getcwd()) parser.add_argument("--task", default="", help="run one task and exit") parser.add_argument("--yes", action="store_true", help="skip the approval prompt (careful)") parser.add_argument("--allow-outside", action="store_true", help="allow file tools outside the workdir") parser.add_argument("--timeout", type=int, default=60, help="seconds per shell command") parser.add_argument("--serve", action="store_true", help="run the local bridge for the website") parser.add_argument("--port", type=int, default=8790, help="bridge port (default 8790)") parser.add_argument("--allow-origin", action="append", default=[], help="extra website origin for the bridge") parser.add_argument("--pair-code", default=os.environ.get("TITAN_PAIR_CODE") or "", help="fixed pairing code") args = parser.parse_args() if args.serve: workdir = Path(args.workdir).expanduser().resolve() if not workdir.is_dir(): print("Working directory does not exist: %s" % workdir) return 2 return serve_bridge( { "workdir": workdir, "allow_outside": args.allow_outside, "timeout": args.timeout, }, args.port, args.allow_origin, args.pair_code, ) if not args.key: print("No agent key. Create one in the owner dashboard (Classmate API keys → Agent), then:") print(" export TITAN_API_KEY=pa-... (or pass --key pa-...)") return 2 state = { "base_url": args.base_url, "api_key": args.key, "model": args.model, "workdir": Path(args.workdir).expanduser().resolve(), "yes": args.yes, "allow_outside": args.allow_outside, "timeout": args.timeout, } if not state["workdir"].is_dir(): print("Working directory does not exist: %s" % state["workdir"]) return 2 print("Titan Agent — %s" % state["base_url"]) print("Working directory: %s" % state["workdir"]) print("Every action is shown before it runs%s." % (" (auto-approved: --yes)" if args.yes else "")) print("File tools stay in the working directory; run_shell can run any command you approve.") print("Type a request, or 'exit' to quit.\n") if args.task: run_task(args.task, state) return 0 while True: try: task = input("you> ").strip() except (EOFError, KeyboardInterrupt): print() return 0 if not task: continue if task.lower() in ("exit", "quit", ":q"): return 0 try: run_task(task, state) except SystemExit as error: print(error) print() if __name__ == "__main__": raise SystemExit(main())