diff --git a/backend/requirements.txt b/backend/requirements.txt index 98eb2d2..4a39930 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -1,6 +1,7 @@ Quart==0.20.0 quart-auth quart-cors +httpx sqlalchemy nest_asyncio replicate diff --git a/backend/src/app.py b/backend/src/app.py index 998eaff..872102f 100644 --- a/backend/src/app.py +++ b/backend/src/app.py @@ -22,6 +22,7 @@ from drives import init_drives from remote import RemoteConnection, RemoteToolSpec, parse_function_specs from app_mixins import ChatMixin, AgentMixin +from channels import Channels class App(ChatMixin, AgentMixin): def __init__(self, data_directory, replicate_token=None, openrouter_token=None): @@ -34,6 +35,9 @@ class App(ChatMixin, AgentMixin): # active websocket tasks, for manual cancelling upon server shutdown self.ws_tasks = [] + # Channel bindings, in-memory transcripts, pollers and HTTP. + self.channels = Channels(self) + self.tmp = tmp.State() # Clients @@ -236,13 +240,16 @@ async def main(): openrouter_token=os.getenv("OPENROUTER_API_KEY") ) - # 3. Start Server + # 3. Set up channel listeners (webhook reconciliation or long polling) + await horde_app.channels.setup() + + # 4. Start Server # manually trigger server shutdown, avoids hypercorn installing signal handlers shutdown_event = asyncio.Event() # TODO: get port from env or don't pass it (actually server should handle none) server_task = asyncio.create_task(start_server(horde_app, port, shutdown_event)) - # 4. Graceful Termination + # 5. Graceful Termination loop = asyncio.get_running_loop() def interrupt(): print("Interrupt Signal") @@ -250,6 +257,8 @@ async def main(): # Unblock websocket handlers parked in websocket.receive() for task in list(horde_app.ws_tasks): task.cancel() + # Stop channel pollers and any in-flight channel runs + horde_app.channels.shutdown() for sig in (signal.SIGINT, signal.SIGTERM): loop.add_signal_handler(sig, interrupt) diff --git a/backend/src/app_mixins/__init__.py b/backend/src/app_mixins/__init__.py index 1cd9ca1..bcd5212 100644 --- a/backend/src/app_mixins/__init__.py +++ b/backend/src/app_mixins/__init__.py @@ -1,2 +1,2 @@ from app_mixins.chat import ChatMixin -from app_mixins.agent import AgentMixin \ No newline at end of file +from app_mixins.agent import AgentMixin diff --git a/backend/src/channels.py b/backend/src/channels.py new file mode 100644 index 0000000..8300e54 --- /dev/null +++ b/backend/src/channels.py @@ -0,0 +1,397 @@ +"""Channel integration, currently Telegram only. + +A *channel* is an inbound/outbound pipe between an external messaging service +and an agent. The binding is stored as an ordinary ``AgentResource`` on the +responding agent:: + + kind = "channel" + key = "telegram" + value = {"token": "", "secret": "", "name": "..."} + +Nothing else is persisted: a Telegram conversation never creates ``Chat`` / +``Thread`` / ``Message`` rows. The rolling transcript lives in memory here +(``Channels``) and only for the lifetime of the process. + +Inbound delivery has two mutually exclusive modes, chosen at startup: + +* **Webhook** when ``HUB_HOST`` is set: Telegram POSTs updates to + ``/channel/telegram/`` (served by ``server/telegram.py``). +* **Long polling** otherwise: one background task per binding calls + ``getUpdates``. + +Both modes converge on :meth:`Channels.handle_update`. + +Outbound is a single ``sendMessage`` call, exposed to the model as a +``ToolSpec`` carrying a ``func`` so the agent's run loop can execute it. +""" + +import asyncio +import os +import time +from collections import deque +from datetime import datetime +from functools import partial + +import httpx + +import database as db +from tool import ToolSpec, ToolParameter + +TELEGRAM_API = "https://api.telegram.org" + +# Telegram long-poll timeout (server holds the request open this long). +POLL_TIMEOUT = 50 +# Rolling transcript cap per conversation, in bytes (oldest trimmed first). +TRANSCRIPT_BYTE_CAP = 8 * 1024 +# Telegram hard limit for a single text message. +MAX_MESSAGE_LENGTH = 4096 + + +class Channels: + """Owns channel bindings, in-memory transcripts, pollers and HTTP.""" + + def __init__(self, app): + self._app = app + self._client = None + self._tasks = [] + self._transcripts = {} # (agent_id, chat_id) -> {transcript, lock} + self._last_update = {} # (agent_id, kind) -> update_id + + # --- Startup / lifecycle --- + + async def setup(self): + """Query channel resources and register listeners where necessary. + + Called once at startup, after ``db.init_database`` and App + construction. Webhook mode reconciles the webhook per binding; when + ``HUB_HOST`` is absent we warn and start one long-polling task per + binding instead. + """ + resources = await db.AgentResource.filter(kind="channel", key="telegram").all() + if not resources: + print("[channel] no telegram channel bindings found") + return + + hub_host = os.getenv("HUB_HOST") + if hub_host: + for resource in resources: + try: + await self._reconcile_webhook(resource, hub_host) + except Exception as e: + print(f"[channel] webhook setup failed for agent {resource.agent_id}: {e}") + else: + print("WARNING: HUB_HOST is not set; telegram channels fall back to long polling") + for resource in resources: + self._track(asyncio.create_task(self._poll(resource))) + + def shutdown(self): + """Cancel every tracked poller / in-flight run.""" + for task in list(self._tasks): + task.cancel() + + # --- Telegram HTTP --- + + def _get_client(self): + if self._client is None: + self._client = httpx.AsyncClient() + return self._client + + async def _api(self, token, method, payload=None, timeout=30): + """Call a Telegram Bot API method and return its ``result``. + + Raises on transport errors and on ``ok: false`` responses. + """ + url = f"{TELEGRAM_API}/bot{token}/{method}" + response = await self._get_client().post(url, json=payload or {}, timeout=timeout) + data = response.json() + if not data.get("ok"): + raise Exception(f"telegram {method} failed: {data.get('description')}") + return data.get("result") + + async def get_binding(self, agent_id): + return await db.AgentResource.get_or_none( + agent_id=agent_id, kind="channel", key="telegram") + + # --- Webhook mode --- + + async def _reconcile_webhook(self, resource, hub_host): + value = resource.value or {} + token = value.get("token") + if not token: + print(f"[channel] telegram binding of agent {resource.agent_id} has no token, skipping") + return + + url = f"https://{hub_host}/channel/telegram/{resource.agent_id}" + info = await self._api(token, "getWebhookInfo") + if info.get("url") == url: + print(f"[channel] webhook already registered for agent {resource.agent_id}") + return + + payload = {"url": url, "allowed_updates": ["message"]} + secret = value.get("secret") + if secret: + payload["secret_token"] = secret + await self._api(token, "setWebhook", payload) + print(f"[channel] registered webhook {url} for agent {resource.agent_id}") + + def accept_update(self, agent_id, kind, update_id): + """Drop duplicate webhook updates by remembering the last update id.""" + if update_id is None: + return True + key = (agent_id, kind) + last = self._last_update.get(key) + if last is not None and update_id <= last: + return False + self._last_update[key] = update_id + return True + + def dispatch(self, agent_id, kind, payload): + """Accept an inbound update and run it in the background. + + Does the duplicate check, spawns the tracked run and returns + immediately so the caller (the webhook route) holds no channel + bookkeeping. + """ + if not self.accept_update(agent_id, kind, payload.get("update_id")): + return + self._track(asyncio.create_task(self.handle_update(agent_id, kind, payload))) + + # --- Long-polling mode --- + + async def _poll(self, resource): + """Long-poll ``getUpdates`` for one telegram binding until cancelled. + + ``offset`` is recomputed from the last seen ``update_id`` so each + update is acknowledged and delivered exactly once. Transport errors + back off instead of busy-looping. + """ + value = resource.value or {} + token = value.get("token") + agent_id = resource.agent_id + if not token: + print(f"[channel] telegram binding of agent {agent_id} has no token, not polling") + return + + # Webhooks and getUpdates are mutually exclusive: clear any webhook + # once before the first poll. + try: + await self._api(token, "deleteWebhook", {"drop_pending_updates": False}) + except Exception as e: + print(f"[channel] deleteWebhook failed for agent {agent_id}: {e}") + + offset = None + backoff = 1 + print(f"[channel] polling telegram updates for agent {agent_id}") + while True: + try: + payload = { + "timeout": POLL_TIMEOUT, + "limit": 100, + "allowed_updates": ["message"], + } + if offset is not None: + payload["offset"] = offset + updates = await self._api( + token, "getUpdates", payload, timeout=POLL_TIMEOUT + 15) + backoff = 1 + for update in updates: + offset = update["update_id"] + 1 + await self.handle_update(agent_id, "telegram", update) + except asyncio.CancelledError: + print(f"[channel] polling stopped for agent {agent_id}") + raise + except Exception as e: + print(f"[channel] telegram polling error for agent {agent_id}: {e}") + await asyncio.sleep(backoff) + backoff = min(backoff * 2, 60) + + # --- Inbound --- + + async def handle_update(self, agent_id, kind, payload): + """Handle one inbound update from a channel. + + Minimal path: only plain text messages from non-bot senders are + consumed. Everything else (edited messages, media, callbacks) is + ignored. + """ + if kind != "telegram": + print(f"[channel] unsupported channel kind '{kind}'") + return + + message = payload.get("message") + if not message: + return + text = message.get("text") + if not text: + return + + sender = message.get("from") or {} + if sender.get("is_bot"): + return + + chat = message.get("chat") or {} + chat_id = chat.get("id") + if chat_id is None: + return + + binding = await self.get_binding(agent_id) + if binding is None: + print(f"[channel] no telegram binding for agent {agent_id}, dropping update") + return + + sender_name = self._sender_name(sender, chat) + state = self._state(agent_id, chat_id) + async with state["lock"]: + state["transcript"].append({ + "sender": sender_name, + "text": text, + "time": time.time(), + }) + self._trim(state["transcript"]) + transcript = self._render(state["transcript"]) + await self._run_agent(agent_id, kind, chat_id, chat, sender_name, transcript) + + async def _run_agent(self, agent_id, kind, chat_id, chat, sender_name, transcript): + """Fire one prompt run for a conversation event and drain it. + + The run input is flattened to an instruction plus exactly two user + messages: a status/info message and the rendered conversation. + """ + instruction = ( + "You are responding to a Telegram conversation. The messages below " + "are the recent conversation; the first user message carries channel " + "context. Reply to the latest message by calling the `send_message` " + "tool with the text to send. Do not add commentary outside the tool call." + ) + + title = chat.get("title") or sender_name + status = "\n".join([ + "Channel: telegram", + f"Chat id: {chat_id}", + f"Chat title: {title}", + f"Chat type: {chat.get('type')}", + f"Current time: {datetime.now().isoformat(timespec='seconds')}", + f"Latest message from: {sender_name}", + ]) + + tool = ToolSpec( + name="send_message", + description="Send a text message back to this Telegram conversation.", + parameters=[ToolParameter( + name="text", + type="string", + description="The message text to send to the conversation.", + required=True, + )], + func=partial(self.send_message, agent_id, kind, chat_id), + ) + + messages = [ + {"role": "user", "content": status}, + {"role": "user", "content": transcript}, + ] + + async def drain(): + async for msg in self._app.prompt_agent( + agent_id, + instruction, + input=None, + options={"tools": [tool], "messages": messages}, + ): + self._log_message(kind, chat_id, msg) + + task = self._track(asyncio.create_task(drain())) + try: + await task + except asyncio.CancelledError: + raise + except Exception as e: + print(f"[channel] run failed for agent {agent_id}: {e}") + + def _log_message(self, kind, chat_id, msg): + content = msg.content + if not isinstance(content, str): + content = repr(content) + print(f"[channel:{kind}:{chat_id}] {msg.role}: {content}") + + # --- Outbound --- + + async def send_message(self, agent_id, kind, chat_id, text): + """Send a text message to a channel conversation. + + Returns a short result string so the model sees the tool outcome. + """ + if kind != "telegram": + return f"error: unsupported channel '{kind}'" + + binding = await self.get_binding(agent_id) + if binding is None: + return "error: no telegram channel binding for this agent" + + token = (binding.value or {}).get("token") + if not token: + return "error: telegram binding has no token" + + text = (text or "").strip() + if not text: + return "error: message text is empty" + if len(text) > MAX_MESSAGE_LENGTH: + text = text[:MAX_MESSAGE_LENGTH] + + try: + await self._api(token, "sendMessage", {"chat_id": chat_id, "text": text}) + return "ok" + except Exception as e: + print(f"[channel] sendMessage failed for agent {agent_id}: {e}") + return f"error: {e}" + + # --- In-memory transcript --- + + def _state(self, agent_id, chat_id): + key = (agent_id, chat_id) + state = self._transcripts.get(key) + if state is None: + state = {"transcript": deque(), "lock": asyncio.Lock()} + self._transcripts[key] = state + return state + + def _trim(self, transcript): + """Drop oldest messages until the deque fits the byte cap.""" + while transcript and self._size(transcript) > TRANSCRIPT_BYTE_CAP: + transcript.popleft() + + def _size(self, transcript): + return sum( + len(m["sender"].encode("utf-8")) + len(m["text"].encode("utf-8")) + 2 + for m in transcript + ) + + def _render(self, transcript): + if not transcript: + return "(no messages)" + return "\n".join(f"{m['sender']}: {m['text']}" for m in transcript) + + # --- Parse helpers --- + + def _sender_name(self, user, chat): + name = " ".join(filter(None, [user.get("first_name"), user.get("last_name")])) + username = user.get("username") + if name and username: + return f"{name} (@{username})" + if name: + return name + if username: + return f"@{username}" + return chat.get("title") or "unknown" + + # --- Tracking --- + + def _track(self, task): + self._tasks.append(task) + + def discard(t): + if t in self._tasks: + self._tasks.remove(t) + + task.add_done_callback(discard) + return task diff --git a/backend/src/server/telegram.py b/backend/src/server/telegram.py new file mode 100644 index 0000000..6ade1f6 --- /dev/null +++ b/backend/src/server/telegram.py @@ -0,0 +1,36 @@ +"""Telegram webhook endpoint. + +Telegram delivers updates to ``POST /channel/telegram/`` once a +webhook is registered for that agent's binding (see ``channels.py``). The +route is always served; it is only *useful* when ``HUB_HOST`` is set and +the webhook has been reconciled, but that is a startup concern, not a +routing one. + +Mirrors ``server/openai.py``'s ``register(server, app)`` pattern. +""" + +from quart import request + + +def register(server, app): + @server.post('/channel/telegram/') + async def telegram_webhook(agent_id): + try: + agent_id = int(agent_id) + except (TypeError, ValueError): + return "invalid agent id", 404 + + binding = await app.channels.get_binding(agent_id) + if binding is None: + return "no telegram channel for this agent", 404 + + secret = request.headers.get('X-Telegram-Bot-Api-Secret-Token') + expected = (binding.value or {}).get('secret') + if expected and secret != expected: + return "forbidden", 403 + + payload = await request.get_json(silent=True) or {} + + # Never block the response on the run: Telegram retries on timeouts. + app.channels.dispatch(agent_id, 'telegram', payload) + return {"ok": True} diff --git a/backend/src/server/web.py b/backend/src/server/web.py index 96a7177..1af8e61 100644 --- a/backend/src/server/web.py +++ b/backend/src/server/web.py @@ -2,6 +2,7 @@ from quart import Quart, request, send_from_directory, send_file, current_app from quart_auth import QuartAuth, AuthUser, login_required, login_user, current_user from server.websocket import register as register_websocket from server.openai import register as register_openai +from server.telegram import register as register_telegram import database as db import drives @@ -175,5 +176,6 @@ async def start_server(app, port, shutdown_event): register_websocket(server, app, auth_manager) register_openai(server, app) + register_telegram(server, app) - await server.run_task(debug=True, shutdown_trigger=shutdown_event.wait, port=port) \ No newline at end of file + await server.run_task(debug=True, shutdown_trigger=shutdown_event.wait, host="0.0.0.0", port=port) \ No newline at end of file diff --git a/docs/telegram.md b/docs/telegram.md new file mode 100644 index 0000000..be3de62 --- /dev/null +++ b/docs/telegram.md @@ -0,0 +1,91 @@ +# Telegram Channel + +Connect a Telegram bot to an agent so the agent can read incoming messages and +reply. Telegram is the only channel implemented today; the machinery lives in +`backend/src/app_mixins/channel.py` (logic) and `backend/src/server/telegram.py` +(webhook route). + +An agent that owns a `channel`/`telegram` resource *is* the agent that answers. +You can bind the same bot token to several agents, and one agent can be bound to +several bots. + +## 1. Create a bot + +Talk to [@BotFather](https://t.me/BotFather), create a bot, and copy its token +(looks like `123456:ABC-DEF...`). Treat the token as a secret. + +## 2. Choose a delivery mode + +Telegram can push updates to a webhook, or the hub can pull them with long +polling. The choice is driven entirely by the `HUB_HOST` environment variable: + +- **Webhook** — set `HUB_HOST` to a publicly reachable host. Telegram requires + HTTPS on ports 443, 80, 88 or 8443, so the hub must be reachable there. +- **Long polling** — leave `HUB_HOST` unset. The hub prints a warning and polls + `getUpdates` instead. No public endpoint or TLS certificate is needed. + +The `POST /channel/telegram/` route is always served regardless of +mode; in polling mode it simply never receives traffic. + +Webhooks and polling are mutually exclusive. A webhook is never registered when +`HUB_HOST` is unset, and the poller clears any existing webhook (`deleteWebhook`) +before it starts. + +## 3. Bind the bot to an agent + +Add a resource to the responding agent (via the resources UI or the +`save-agent-resource` action): + +``` +kind = "channel" +key = "telegram" +value = {"token": "", "secret": "", "name": ""} +``` + +- `token` — the BotFather token. Required. +- `secret` — a random string of `A-Z a-z 0-9 _ -`, 1–256 characters. Optional, + and only meaningful in webhook mode. When present, Telegram sends it back on + every request in the `X-Telegram-Bot-Api-Secret-Token` header and the route + rejects mismatches with `403`. +- `name` — a display label. Optional. + +This is the only state the channel persists. Conversations are **not** stored — +no `Chat`, `Thread` or `Message` rows are created. The rolling transcript is +held in memory on the hub and discarded when the process exits. + +## 4. Restart the hub + +On startup the hub queries all `channel`/`telegram` resources and reconciles the +provider: + +- **Webhook mode**: it calls `getWebhookInfo` for each binding and, when the + registered URL does not match `https:///channel/telegram/`, + calls `setWebhook` with that URL, the `secret_token`, and + `allowed_updates=["message"]`. Verify with `getWebhookInfo`. +- **Polling mode**: it starts one background task per binding. Confirm the bot is + reachable with `getMe`, then send a test message. + +## 5. Talk to the bot + +Message the bot. On each text message the hub: + +1. Appends the message to the conversation's in-memory transcript. +2. Runs the agent with a flattened input — an instruction, a status message + (channel, chat id/title/type, current time, sender), and the rendered + conversation. +3. The agent replies by calling the `send_message` tool, which issues a + `sendMessage` call to Telegram. + +Non-text updates (edited messages, media, callbacks) are ignored. Messages +authored by other bots are ignored. + +## 6. Troubleshooting + +| Symptom | Cause | +|---|---| +| Webhook not set | `HUB_HOST` unset, or the host is not reachable over HTTPS. Check the startup log. | +| Polling active when you expected a webhook | `HUB_HOST` is unset. | +| `403` from the webhook route | `X-Telegram-Bot-Api-Secret-Token` does not match the resource's `secret`. | +| `404` from the webhook route | No `channel`/`telegram` resource on that agent. | +| No reply | The agent has no model/tool configuration, or the run errored — check the hub log for `[channel]` lines. | +| `Conflict: can't use getUpdates method while webhook is active` | A stale webhook from a previous deployment. Remove it (`deleteWebhook`) or set `HUB_HOST` so webhook mode takes over. |