"""Connect a Hermes Agent to a DonutChat bot. Listens to the bot's event stream, sends each message to Hermes, and streams the answer into the thread under that message. Requires Python 3.9 or later and: pip install "websockets>=13" requests """ import base64 import json import os import random import threading import time from collections import defaultdict, deque from concurrent.futures import ThreadPoolExecutor from urllib.parse import quote import requests from websockets.sync.client import connect BOT_TOKEN = os.environ["DONUT_BOT_TOKEN"] HERMES_KEY = os.environ["HERMES_API_KEY"] HERMES_URL = os.environ.get("HERMES_URL", "http://127.0.0.1:8642") HERMES_MODEL = os.environ.get("HERMES_MODEL", "hermes-agent") STREAM = os.environ.get("DONUT_STREAM_URL", "wss://api.donutchat.com/bots/v1/stream") API = os.environ.get("DONUT_API_URL", "https://api.donutchat.com/bots/v1") ANSWER_TIMEOUT = 600 # seconds before an answer is finished as failed MAX_BYTES = 79_000 # a reply holds at most 80,000 UTF-8 bytes AUTH = {"Authorization": f"Bearer {BOT_TOKEN}"} workers = ThreadPoolExecutor(max_workers=4) # answers written at the same time thread_locks = defaultdict(threading.Lock) # one answer at a time per thread seen = deque(maxlen=1000) # event IDs already handled started = deque(maxlen=1000) # threads answered since the script started; Hermes has their first message last_event_id = None def log(*parts): print(time.strftime("%H:%M:%S"), *parts, flush=True) def call(path, body): data = requests.post(f"{API}/{path}", json=body, headers=AUTH, timeout=15).json() if not data.get("ok"): raise RuntimeError(f"{path} failed: {data.get('error')}") return data def ask_hermes(session_id, prompt): """Yield the answer text from Hermes as it is written.""" with requests.post( f"{HERMES_URL}/v1/chat/completions", headers={"Authorization": f"Bearer {HERMES_KEY}", "X-Hermes-Session-Id": session_id}, json={"model": HERMES_MODEL, "stream": True, "messages": [{"role": "user", "content": prompt}]}, stream=True, timeout=(10, 120), ) as res: res.raise_for_status() res.encoding = "utf-8" for line in res.iter_lines(decode_unicode=True): if not line.startswith("data:"): continue data = line[5:].strip() if data == "[DONE]": return try: text = json.loads(data)["choices"][0]["delta"].get("content") except (ValueError, LookupError, TypeError, AttributeError): continue # tool progress and other events without answer text if text: yield text MAX_IMAGE_BYTES = 10 * 1024 * 1024 # aggregate per turn; also the API's per-file limit MAX_IMAGES = 4 def image_mime(data): """Recognize formats accepted by the agent without trusting event metadata.""" if data.startswith(b"\x89PNG\r\n\x1a\n"): return "image/png" if data.startswith(b"\xff\xd8\xff"): return "image/jpeg" if data.startswith((b"GIF87a", b"GIF89a")): return "image/gif" if data.startswith(b"RIFF") and data[8:12] == b"WEBP": return "image/webp" raise ValueError("unsupported image format") def add_attachments(text, attachments, source, parts, left): """Add a message's images to parts. Return its text with a note for each file not sent.""" if len(attachments) > 64: text += "\n[Additional attachments not sent: attachment limit exceeded.]" for attachment in attachments[:64]: note = "[Attachment not sent: unsupported file type.]" if attachment.get("file_type") == "image" or attachment.get("mime_type", "").startswith("image/"): note = "[Image not sent: download unavailable, unsupported format, or image limit exceeded.]" try: budget = left["bytes"] if left["images"] <= 0 or budget <= 0 or not 0 <= attachment.get("file_size", 0) <= budget: raise ValueError("image limit exceeded") left["images"] -= 1 attachment_id = int(attachment["id"]) if attachment_id <= 0: raise ValueError("invalid attachment ID") # No redirects: never forward the bearer token to a storage URL. deadline = time.monotonic() + 30 with requests.get(f"{API}/{source}/attachments/{attachment_id}/content", headers=AUTH, stream=True, timeout=(10, 30), allow_redirects=False) as res: if res.status_code != 200: raise ValueError("attachment unavailable") if int(res.headers.get("Content-Length", 0)) > budget: raise ValueError("image limit exceeded") image = bytearray() for chunk in res.iter_content(chunk_size=64 * 1024): image.extend(chunk) if len(image) > budget or time.monotonic() > deadline: raise ValueError("image limit exceeded") mime = image_mime(image) encoded = base64.b64encode(image).decode("ascii") parts.append({"type": "image_url", "image_url": {"url": f"data:{mime};base64,{encoded}"}}) left["bytes"] -= len(image) continue except (requests.RequestException, ValueError, KeyError, TypeError): pass # report only a generic note; no private URLs or tokens in the prompt text += "\n" + note return text def context_message(label, ref, chat_id, parts, left): """Describe a message the event carries as context: the thread's first message or a linked one.""" source = f"chats/{chat_id}/messages/{int(ref['message_id'])}" if ref.get("reply_id"): source += f"/replies/{int(ref['reply_id'])}" text = f"[{label} {ref['sender']['name']}: {ref.get('text', '')}]" return add_attachments(text, ref.get("attachments", []), source, parts, left) def user_content(data, thread_id, with_parent=True): chat_id = int(data["chat_id"]) parts, left = [], {"bytes": MAX_IMAGE_BYTES, "images": MAX_IMAGES} source = f"chats/{chat_id}/messages/{int(thread_id)}" if "reply_id" in data: source += f"/replies/{int(data['reply_id'])}" # The sender's own images come first, so context can't use up the image limits. text = add_attachments(f"{data['sender']['name']}: {data.get('text', '')}", data.get("attachments", []), source, parts, left) refs = [("Linked message from", ref) for ref in data.get("linked_messages", [])] if with_parent and data.get("parent"): refs.insert(0, ("Thread started by", data["parent"])) context = [context_message(label, ref, chat_id, parts, left) for label, ref in refs] text = "\n".join([*context, text]) if not data.get("attachments") and not any(ref.get("attachments") for _, ref in refs): return text # plain-text turns keep the plain format return [{"type": "text", "text": text}, *parts] def fit(text): if len(text.encode()) <= MAX_BYTES: return text return text.encode()[:MAX_BYTES].decode(errors="ignore") + "\n\n*The answer was cut off here.*" def answer(chat_id, thread_id, chunks): """Stream text into one thread reply, then finish it as OK or FL.""" created = call("thread-replies", {"chat_id": chat_id, "parent_message_id": thread_id, "content": "", "content_type": "MD", "status": "GN"}) reply = {"chat_id": chat_id, "parent_message_id": thread_id, "reply_id": created["reply_id"]} text, sent, deadline = "", 0.0, time.monotonic() + ANSWER_TIMEOUT try: for chunk in chunks: text += chunk if time.monotonic() > deadline: raise TimeoutError("the agent took too long") if time.monotonic() - sent >= 0.4: # about 2-3 updates per second call("thread-replies/stream", {**reply, "content": fit(text), "status": "GN"}) sent = time.monotonic() if not text.strip(): raise RuntimeError("the agent returned no text") call("thread-replies/stream", {**reply, "content": fit(text), "status": "OK"}) except Exception: call("thread-replies/stream", {**reply, "status": "FL", "content": fit(text) + "\n\n*The agent couldn't finish this answer.*"}) raise def handle(data, thread_id): chat_id, sender = data["chat_id"], data["sender"] with thread_locks[(chat_id, thread_id)]: try: # One Hermes session per thread: Hermes keeps that thread's earlier turns. # A reply in a thread this script hasn't answered in also sends the thread's # first message: someone else may have written it, or it never reached the script. with_parent = (chat_id, thread_id) not in started if with_parent: started.append((chat_id, thread_id)) chunks = ask_hermes(f"donutchat-{chat_id}-{thread_id}", user_content(data, thread_id, with_parent)) answer(chat_id, thread_id, chunks) log("answered", sender["name"], "in chat", chat_id) except Exception as error: log("failed in chat", chat_id, "-", error) def on_event(event): kind, data = event.get("type"), event.get("data", {}) if kind == "message.new": thread_id = data["message_id"] elif kind == "thread.reply.new": thread_id = data["parent_message_id"] else: return sender = data["sender"] if not data.get("text", "").strip() and not data.get("attachments"): return log("request from", sender["name"], "(user ID", str(sender["id"]) + ")") workers.submit(handle, data, thread_id) def listen(): """Read events until the connection closes. Returns True if it was live.""" global last_event_id url = STREAM + (f"?last_event_id={quote(last_event_id)}" if last_event_id else "") live = False with connect(url, additional_headers=AUTH, max_size=2**21) as ws: closed = threading.Event() def keep_alive(): # the server closes a connection that sends nothing for 5 minutes while not closed.wait(60): try: ws.send("ping") except Exception: return threading.Thread(target=keep_alive, daemon=True).start() try: for frame in ws: event = json.loads(frame) if event.get("type") == "connected": live = True log("connected as bot", event["data"]["bot_id"]) event_id = event.get("event_id") if not event_id or event_id in seen: continue seen.append(event_id) last_event_id = event_id on_event(event) finally: closed.set() return live def main(): delay = 1 while True: try: if listen(): delay = 1 except Exception as error: log("disconnected -", error) # Wait a slightly different time than every other bot, so a server # restart doesn't bring them all back in the same instant. time.sleep(delay * random.uniform(0.5, 1.5)) delay = min(delay * 2, 60) if __name__ == "__main__": main()