from __future__ import annotations import argparse import json import mimetypes import threading from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import parse_qs, urlparse from .adapters import adapter_for from .avatar import available_states, get_avatar from .config import find_agent, load_agents from .sessions import SessionStore, new_session_id, utc_now PROJECT_ROOT = Path(__file__).resolve().parent.parent STATIC_ROOT = PROJECT_ROOT / "static" DEFAULT_CONFIG = PROJECT_ROOT / "configs" / "agents.json" DEFAULT_SESSIONS = PROJECT_ROOT / "data" / "sessions" TTS_DIR = PROJECT_ROOT / "data" / "tts" TTS_DIR.mkdir(parents=True, exist_ok=True) # keyed by session_id: None=pending, dict=ready, str=error message _anim_store: dict[str, None | dict | str] = {} _anim_lock = threading.Lock() def _start_avatar_pipeline(session_id: str, text: str) -> None: with _anim_lock: _anim_store[session_id] = None # pending def _run() -> None: try: from .tts import synthesize from .avatar_lam import audio_to_blendshapes wav_path = TTS_DIR / f"{session_id}.wav" synthesize(text, wav_path) bs_data = audio_to_blendshapes(wav_path) with _anim_lock: _anim_store[session_id] = bs_data except Exception as exc: with _anim_lock: _anim_store[session_id] = str(exc) threading.Thread(target=_run, daemon=True).start() class AgentGuiServer(ThreadingHTTPServer): def __init__(self, server_address, RequestHandlerClass, config_path: Path, sessions_root: Path): super().__init__(server_address, RequestHandlerClass) self.config_path = config_path self.sessions = SessionStore(sessions_root) @property def agents(self): return load_agents(self.config_path) class Handler(BaseHTTPRequestHandler): server: AgentGuiServer def log_message(self, format: str, *args): print("[agent-gui] " + format % args) def do_GET(self): parsed = urlparse(self.path) if parsed.path == "/health": return self.json_response({"ok": True, "service": "agent-gui", "time": utc_now()}) if parsed.path == "/api/agents": return self.json_response({"agents": self.server.agents}) if parsed.path == "/api/avatar": state = parse_qs(parsed.query).get("state", ["idle"])[0] return self.json_response({"avatar": get_avatar(state), "states": available_states()}) if parsed.path == "/api/sessions": return self.json_response({"sessions": self.server.sessions.list_sessions()}) if parsed.path == "/api/history": session_id = parse_qs(parsed.query).get("session_id", [""])[0] return self.json_response({"events": self.server.sessions.read(session_id)}) if parsed.path == "/api/avatar/anim": session_id = parse_qs(parsed.query).get("session_id", [""])[0] with _anim_lock: val = _anim_store.get(session_id) if val is None: return self.json_response({"status": "pending"}) if isinstance(val, str): return self.json_response({"status": "error", "message": val}) return self.json_response({"status": "ready", "bsData": val}) if parsed.path == "/api/avatar/audio": session_id = parse_qs(parsed.query).get("session_id", [""])[0] wav = TTS_DIR / f"{session_id}.wav" if not wav.exists(): return self.json_response({"error": "audio not found"}, status=404) body = wav.read_bytes() self.send_response(200) self.send_header("Content-Type", "audio/wav") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) return return self.serve_static(parsed.path) def do_POST(self): if self.path == "/api/session": return self.json_response({"session_id": new_session_id()}) if self.path == "/api/chat": return self.handle_chat() return self.json_response({"error": "Endpoint inconnu"}, status=404) def handle_chat(self): try: payload = self.read_json() agent_id = str(payload.get("agent", "echo")) message = str(payload.get("message", "")).strip() session_id = str(payload.get("session_id") or new_session_id()) if not message: return self.json_response({"error": "Message vide"}, status=400) agent = find_agent(self.server.agents, agent_id) self.server.sessions.append(session_id, {"type": "user_message", "agent": agent_id, "content": message}) self.server.sessions.append(session_id, {"type": "status", "agent": agent_id, "content": "Agent en réflexion..."}) result = adapter_for(agent).run(message, agent, session_id) stored_events = [self.server.sessions.append(session_id, event) for event in result.events] assistant_text = next( (e["content"] for e in stored_events if e.get("type") == "assistant_message"), None, ) if assistant_text: _start_avatar_pipeline(session_id, assistant_text) return self.json_response( { "session_id": session_id, "agent": agent_id, "events": stored_events, "avatar": get_avatar(result.avatar_state), "avatar_anim": "pending" if assistant_text else None, } ) except KeyError as exc: return self.json_response({"error": str(exc), "avatar": get_avatar("error")}, status=404) except Exception as exc: # defensive boundary for local UI return self.json_response({"error": str(exc), "avatar": get_avatar("error")}, status=500) def serve_static(self, request_path: str): relative = "index.html" if request_path in ("/", "") else request_path.lstrip("/") path = (STATIC_ROOT / relative).resolve() if STATIC_ROOT.resolve() not in path.parents and path != STATIC_ROOT.resolve(): return self.json_response({"error": "Chemin interdit"}, status=403) if not path.exists() or not path.is_file(): return self.json_response({"error": "Fichier introuvable"}, status=404) content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream" body = path.read_bytes() self.send_response(200) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def read_json(self): length = int(self.headers.get("Content-Length", "0")) body = self.rfile.read(length).decode("utf-8") if length else "{}" return json.loads(body) def json_response(self, payload, status: int = 200): body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def main(argv: list[str] | None = None): parser = argparse.ArgumentParser(description="Agent GUI local") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8765) parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG) parser.add_argument("--sessions", type=Path, default=DEFAULT_SESSIONS) args = parser.parse_args(argv) server = AgentGuiServer((args.host, args.port), Handler, args.config, args.sessions) print(f"Agent GUI lancé: http://{args.host}:{args.port}") print(f"Config agents: {args.config}") print("Ctrl+C pour arrêter.") server.serve_forever() if __name__ == "__main__": main()