#!/usr/bin/env python3 """ emo-agent — Local execution agent for Émo (HTTP long-polling). Hugo runs this on his PC. It long-polls the Émo backend for tool requests (exec_shell, read_file, write_file, list_dir), executes them locally, and posts results back. No WebSocket required (more resilient through proxies). USAGE: python3 emo-agent.py --token Env vars: EMO_AGENT_TOKEN EMO_BACKEND_URL (default: http://127.0.0.1:8010 or your cloud URL) DEPS: pip install httpx SECURITY: The agent runs ARBITRARY shell commands. Keep your token secret. Rotate it from the Émo UI if it leaks. The agent has the same privileges as the user that started it. """ import asyncio import argparse import json import os import platform import sys from pathlib import Path try: import httpx except ImportError: print("[emo-agent] Missing dependency. Run: pip install httpx", file=sys.stderr) sys.exit(1) DEFAULT_BACKEND = os.environ.get("EMO_BACKEND_URL", "http://127.0.0.1:8010") # ----------------------------- TOOLS ----------------------------- # async def tool_exec_shell(args: dict) -> dict: cmd = args.get("cmd") or "" cwd = args.get("cwd") or None timeout = int(args.get("timeout") or 60) if not cmd: return {"ok": False, "error": "cmd missing"} try: proc = await asyncio.create_subprocess_shell( cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd, ) try: stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) except asyncio.TimeoutError: proc.kill() return {"ok": False, "error": f"timeout after {timeout}s", "exit_code": -1} return { "ok": True, "exit_code": proc.returncode, "stdout": stdout.decode(errors="replace"), "stderr": stderr.decode(errors="replace"), } except Exception as e: return {"ok": False, "error": str(e)} async def tool_read_file(args: dict) -> dict: path = args.get("path") if not path: return {"ok": False, "error": "path missing"} try: p = Path(path).expanduser() if not p.exists(): return {"ok": False, "error": "file not found"} if p.stat().st_size > 5 * 1024 * 1024: return {"ok": False, "error": "file too large (>5MB) — use offset/limit or grep"} content = p.read_text(errors="replace") offset = int(args.get("offset") or 1) limit = int(args.get("limit") or 0) if offset > 1 or limit > 0: lines = content.splitlines(keepends=True) start = max(0, offset - 1) end = len(lines) if limit <= 0 else min(len(lines), start + limit) numbered = [] for i, ln in enumerate(lines[start:end], start=start + 1): numbered.append(f"{i:6d}|{ln.rstrip(chr(10))}") content = "\n".join(numbered) + ("\n" if numbered else "") return {"ok": True, "content": content, "path": str(p.resolve()), "offset": offset, "limit": limit} except Exception as e: err = str(e) if "aswMonFltProxy" in err or "Permission denied" in err: return {"ok": False, "error": "Avast bloque l acces au fichier. Ajoute une exception pour Emo Online dans Avast."} return {"ok": False, "error": err} async def tool_write_file(args: dict) -> dict: path = args.get("path") content = args.get("content") if not path or content is None: return {"ok": False, "error": "path and content required"} try: p = Path(path).expanduser() p.parent.mkdir(parents=True, exist_ok=True) p.write_text(content) return {"ok": True, "path": str(p.resolve()), "bytes": len(content.encode())} except Exception as e: err = str(e) if "aswMonFltProxy" in err or "Permission denied" in err: return {"ok": False, "error": "Avast bloque l acces au fichier. Ajoute une exception pour Emo Online dans Avast."} return {"ok": False, "error": err} async def tool_list_dir(args: dict) -> dict: path = args.get("path") or "." depth = int(args.get("depth") or 1) depth = min(max(depth, 1), 4) try: p = Path(path).expanduser() if not p.exists(): return {"ok": False, "error": "path not found"} if not p.is_dir(): return {"ok": False, "error": "not a directory"} root = p.resolve() if depth <= 1: files, dirs = [], [] for entry in sorted(p.iterdir()): if entry.name.startswith("."): continue (dirs if entry.is_dir() else files).append(entry.name) return {"ok": True, "path": str(root), "files": files, "dirs": dirs} entries = [] skip_dirs = {".git", "node_modules", "__pycache__", "vendor"} def walk(d: Path, lvl: int): if lvl > depth or len(entries) >= 500: return try: items = sorted(d.iterdir()) except OSError: return for entry in items: if entry.name.startswith("."): continue rel = entry.relative_to(root).as_posix() entries.append({"path": rel, "is_dir": entry.is_dir()}) if entry.is_dir() and lvl < depth and entry.name not in skip_dirs and len(entries) < 500: walk(entry, lvl + 1) walk(p, 1) return {"ok": True, "path": str(root), "entries": entries, "truncated": len(entries) >= 500} except Exception as e: return {"ok": False, "error": str(e)} def _is_binary_sample(data: bytes) -> bool: if b"\x00" in data: return True return any(b < 9 or (13 < b < 32) for b in data) async def tool_grep(args: dict) -> dict: pattern = args.get("pattern") or "" if not pattern: return {"ok": False, "error": "pattern missing"} root = Path(args.get("path") or ".").expanduser() glob_pat = args.get("glob") or "*" max_results = min(int(args.get("max_results") or 100), 200) ignore_case = bool(args.get("ignore_case")) needle = pattern.lower() if ignore_case else pattern skip_dirs = {".git", "node_modules", "__pycache__", "vendor"} matches = [] for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if not d.startswith(".") and d not in skip_dirs] for fname in filenames: if len(matches) >= max_results: break if glob_pat != "*" and not Path(fname).match(glob_pat): continue fp = Path(dirpath) / fname try: if fp.stat().st_size > 2 * 1024 * 1024: continue head = fp.read_bytes()[:512] if _is_binary_sample(head): continue with fp.open("r", errors="replace") as f: for i, line in enumerate(f, 1): hay = line.lower() if ignore_case else line if needle in hay: matches.append({"file": str(fp), "line": i, "text": line[:300]}) if len(matches) >= max_results: break except OSError: continue if len(matches) >= max_results: break return {"ok": True, "pattern": pattern, "matches": matches, "truncated": len(matches) >= max_results} async def tool_edit_file(args: dict) -> dict: path = args.get("path") old = args.get("old_string") new = args.get("new_string") if not path or old is None or new is None: return {"ok": False, "error": "path, old_string, new_string required"} try: p = Path(path).expanduser() content = p.read_text(errors="replace") count = content.count(old) if count == 0: return {"ok": False, "error": "old_string not found"} replace_all = bool(args.get("replace_all")) if not replace_all and count > 1: return {"ok": False, "error": f"old_string found {count} times — use replace_all or be more specific"} if replace_all: updated = content.replace(old, new) replaced = count else: updated = content.replace(old, new, 1) replaced = 1 p.write_text(updated) return {"ok": True, "path": str(p.resolve()), "replacements": replaced, "content": updated} except Exception as e: return {"ok": False, "error": str(e)} async def tool_delete_path(args: dict) -> dict: path = args.get("path") if not path: return {"ok": False, "error": "path missing"} try: p = Path(path).expanduser() if p.is_dir(): import shutil shutil.rmtree(p) else: p.unlink(missing_ok=False) return {"ok": True, "deleted": str(p)} except Exception as e: return {"ok": False, "error": str(e)} async def tool_move_path(args: dict) -> dict: src = args.get("from") dst = args.get("to") if not src or not dst: return {"ok": False, "error": "from and to required"} try: f = Path(src).expanduser() t = Path(dst).expanduser() t.parent.mkdir(parents=True, exist_ok=True) f.rename(t) return {"ok": True, "from": str(f), "to": str(t.resolve())} except Exception as e: return {"ok": False, "error": str(e)} async def tool_find_files(args: dict) -> dict: pattern = args.get("pattern") or "" if not pattern: return {"ok": False, "error": "pattern missing"} root = Path(args.get("path") or ".").expanduser() max_results = min(int(args.get("max_results") or 200), 500) skip_dirs = {".git", "node_modules"} found = [] has_glob = "*" in pattern for dirpath, dirnames, filenames in os.walk(root): dirnames[:] = [d for d in dirnames if not d.startswith(".") and d not in skip_dirs] for fname in filenames: if len(found) >= max_results: break fp = Path(dirpath) / fname if has_glob: if not Path(fname).match(pattern): continue elif pattern.lower() not in str(fp).lower(): continue found.append(str(fp)) if len(found) >= max_results: break return {"ok": True, "pattern": pattern, "files": found, "truncated": len(found) >= max_results} async def tool_print_file(args: dict) -> dict: path = args.get("path") or "" printer = (args.get("printer") or "").strip() copies = max(1, int(args.get("copies") or 1)) if not path: return {"ok": False, "error": "path required"} p = Path(path).expanduser() if not p.is_file(): return {"ok": False, "error": f"file not found: {p}"} abs_path = str(p.resolve()) if sys.platform == "win32": if not printer: proc = await asyncio.create_subprocess_exec( "powershell", "-NoProfile", "-Command", "(Get-CimInstance Win32_Printer | Where-Object {$_.Default -eq $true}).Name", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) out, _ = await proc.communicate() printer = out.decode(errors="replace").strip() if not printer: return {"ok": False, "error": "no default printer found"} ext = p.suffix.lower() for _ in range(copies): if ext == ".pdf": acrobat_candidates = [ Path(os.environ.get("ProgramFiles", "")) / "Adobe/Acrobat DC/Acrobat/Acrobat.exe", Path(os.environ.get("ProgramFiles(x86)", "")) / "Adobe/Acrobat Reader DC/Reader/AcroRd32.exe", Path(os.environ.get("ProgramFiles", "")) / "Adobe/Acrobat Reader DC/Reader/AcroRd32.exe", ] acrobat = next((str(c) for c in acrobat_candidates if c.is_file()), "") if not acrobat: return {"ok": False, "error": "Adobe Acrobat/Reader required to print PDF"} proc = await asyncio.create_subprocess_exec( acrobat, "/t", abs_path, printer, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL, ) await proc.wait() else: ps = f'Start-Process -FilePath "{abs_path}" -Verb Print' proc = await asyncio.create_subprocess_exec( "powershell", "-NoProfile", "-Command", ps, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) _, err = await proc.communicate() if proc.returncode != 0: return {"ok": False, "error": err.decode(errors="replace") or "print failed"} else: for _ in range(copies): cmd = ["lp"] if printer: cmd.extend(["-d", printer]) cmd.append(abs_path) proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) out, err = await proc.communicate() if proc.returncode != 0: return {"ok": False, "error": err.decode(errors="replace") or out.decode(errors="replace")} return {"ok": True, "path": abs_path, "printer": printer or "default", "copies": copies} async def tool_codebase_search(args: dict) -> dict: query = (args.get("query") or args.get("pattern") or "").strip() if not query: return {"ok": False, "error": "query missing"} path = args.get("path") or "." dirs = args.get("target_directories") if isinstance(dirs, list) and dirs: path = dirs[0] return await tool_grep({ "pattern": query, "path": path, "ignore_case": True, "max_results": int(args.get("max_results") or 80), }) TOOLS = { "exec_shell": tool_exec_shell, "read_file": tool_read_file, "write_file": tool_write_file, "list_dir": tool_list_dir, "grep": tool_grep, "edit_file": tool_edit_file, "delete_path": tool_delete_path, "move_path": tool_move_path, "find_files": tool_find_files, "print_file": tool_print_file, "print_document": tool_print_file, "print": tool_print_file, "codebase_search": tool_codebase_search, "run_terminal_cmd": tool_exec_shell, "run_terminal_command": tool_exec_shell, "bash": tool_exec_shell, "file_search": tool_find_files, "delete_file": tool_delete_path, "create_file": tool_write_file, "grep_search": tool_grep, } # ----------------------------- POLL LOOP ----------------------------- # def _shrink(args: dict) -> str: s = json.dumps(args, ensure_ascii=False) return s if len(s) < 120 else s[:117] + "..." def _summarize(result: dict) -> str: if not result.get("ok"): return f"ERROR: {result.get('error', '')[:120]}" if "exit_code" in result: return f"exit={result['exit_code']}, out={len(result.get('stdout', ''))}c" if "content" in result: return f"read {len(result['content'])} chars" if "matches" in result: return f"{len(result['matches'])} matches" if "entries" in result: return f"{len(result['entries'])} entries" if "replacements" in result: return f"{result['replacements']} edits" if "files" in result and isinstance(result["files"], list) and "dirs" not in result: return f"{len(result['files'])} files found" if "files" in result: return f"{len(result['files'])} files, {len(result.get('dirs', []))} dirs" if "deleted" in result: return "deleted" if "from" in result and "to" in result: return "moved" if "printer" in result: return f"printed → {result.get('printer')}" return "ok" async def execute_and_post(client: httpx.AsyncClient, backend: str, token: str, req: dict): rid = req.get("id") tool = req.get("tool") args = req.get("args") or {} handler = TOOLS.get(tool) if handler is None: result = {"ok": False, "error": f"unknown tool: {tool}"} else: try: print(f"[emo-agent] → {tool}({_shrink(args)})") result = await handler(args) print(f"[emo-agent] ← {_summarize(result)}") except Exception as e: result = {"ok": False, "error": f"handler crashed: {e}"} try: await client.post( f"{backend}/api/agent/result", params={"token": token}, json={"id": rid, "result": result}, timeout=15, ) except Exception as e: print(f"[emo-agent] ! failed to post result: {e}") async def heartbeat_loop(client: httpx.AsyncClient, backend: str, token: str): payload = agent_machine_context() while True: try: await client.post( f"{backend}/api/agent/heartbeat", params={"token": token}, json=payload, timeout=10, ) except Exception: pass await asyncio.sleep(5) def agent_machine_context() -> dict: home = Path.home() desktop = home / "Desktop" if not desktop.is_dir(): alt = home / "OneDrive" / "Desktop" if alt.is_dir(): desktop = alt return { "home": str(home), "desktop": str(desktop), "username": os.environ.get("USERNAME") or os.environ.get("USER") or "", "userprofile": os.environ.get("USERPROFILE") or str(home), "os": platform.system(), "hostname": platform.node(), } async def run(token: str, backend: str): backend = backend.rstrip("/") print(f"[emo-agent] backend = {backend}") print("[emo-agent] starting… Ctrl+C to quit.") async with httpx.AsyncClient() as client: # First heartbeat to validate token & go online try: r = await client.post( f"{backend}/api/agent/heartbeat", params={"token": token}, json=agent_machine_context(), timeout=10, ) if r.status_code != 200: print(f"[emo-agent] auth failed (HTTP {r.status_code}): {r.text}") return except Exception as e: print(f"[emo-agent] cannot reach backend: {e}") return print("[emo-agent] online. Émo can now pilot this machine.") asyncio.create_task(heartbeat_loop(client, backend, token)) while True: try: r = await client.get( f"{backend}/api/agent/poll", params={"token": token}, timeout=35, ) if r.status_code != 200: print(f"[emo-agent] poll error {r.status_code}; retrying in 3s") await asyncio.sleep(3) continue data = r.json() if data.get("empty"): continue # re-poll immediately req = data.get("request") or {} asyncio.create_task(execute_and_post(client, backend, token, req)) except httpx.ReadTimeout: continue except KeyboardInterrupt: print("\n[emo-agent] bye.") return except Exception as e: print(f"[emo-agent] poll exception: {e}; retrying in 3s") await asyncio.sleep(3) def main(): ap = argparse.ArgumentParser(description="Émo local agent") ap.add_argument("--token", default=os.environ.get("EMO_AGENT_TOKEN")) ap.add_argument("--backend", default=DEFAULT_BACKEND) args = ap.parse_args() if not args.token: print("[emo-agent] Token required. Use --token or EMO_AGENT_TOKEN env var.", file=sys.stderr) sys.exit(2) try: asyncio.run(run(args.token, args.backend)) except KeyboardInterrupt: print("\n[emo-agent] bye.") if __name__ == "__main__": main()