Merge pull request #82 from makespacemadrid/codex/update-telegram-bot-for-natural-language-support

Add optional Ocabra-compatible Telegram assistant, bounded executor, deduplication, and Telegram client improvements

Author: luisriverag

README.md

@@ -19,7 +19,7 @@ A Flask + SQLite application for managing budget proposals in a hackerspace.
 - **Proposals**: weighted vote thresholds, creator auto-vote, edit/delete by owner/admin, approval undo, purchase tracking.
 - **Polls**: 2..12 options, transparent results, close/reopen/delete, web/Telegram vote modes.
 - **Group purchases**: shared orders with individually priced options, proportional shipping/tax costs, per-member quantities, deadlines, payment tracking, fulfillment states, and Telegram lifecycle notifications.
-- **Telegram integration**: `/link`, `/vote`, `/pvote`, inline poll/proposal callbacks, webhook processing.
+- **Telegram integration**: `/link`, `/vote`, `/pvote`, inline callbacks, and an optional Ocabra-compatible natural-language assistant with MCP tools, DB-backed access, confirmed admin mutations, retry deduplication, and bounded background work.
 - **Budget lifecycle**: approval when threshold+budget are met, over-budget queue with auto-approval later.
 - **Timezone-aware UI**: all timestamps are rendered in configured timezone.
 
@@ -48,6 +48,13 @@ See [`docs/QUICKSTART.md`](docs/QUICKSTART.md) for Docker/local setup, bootstrap
 - Main docs index: [`docs/INDEX.md`](docs/INDEX.md)
 - Direct links: [`docs/QUICKSTART.md`](docs/QUICKSTART.md), [`docs/APIDOC.md`](docs/APIDOC.md), [`docs/SPEC.md`](docs/SPEC.md), [`docs/TESTING.md`](docs/TESTING.md)
 
+## Acknowledgements
+
+The natural-language Telegram assistant was inspired by Luis Rivera's
+[`ocabra_telegram`](https://github.com/luisriverag/ocabra_telegram) project. ManaVote
+adapts its OpenAI-compatible Telegram conversation approach to the existing webhook,
+database-backed member access controls, and MCP tools in this application.
+
 ## Frontend development
 
 The shared application shell is implemented in React and built with Vite, while Flask

app/integrations/bounded_executor.py

@@ -0,0 +1,35 @@
+"""Small bounded wrapper around :mod:`concurrent.futures` executors."""
+
+from __future__ import annotations
+
+import threading
+from concurrent.futures import Future, ThreadPoolExecutor
+from typing import Callable
+
+
+class BoundedExecutor:
+    """Reject jobs instead of allowing an unbounded in-memory work queue."""
+
+    def __init__(self, *, max_workers: int, max_pending: int, thread_name_prefix: str):
+        if max_workers < 1 or max_pending < 0:
+            raise ValueError("max_workers must be positive and max_pending cannot be negative")
+        self._executor = ThreadPoolExecutor(
+            max_workers=max_workers,
+            thread_name_prefix=thread_name_prefix,
+        )
+        self._capacity = threading.BoundedSemaphore(max_workers + max_pending)
+
+    def submit(self, function: Callable, *args, **kwargs) -> Future | None:
+        """Submit immediately, or return ``None`` when all slots are occupied."""
+        if not self._capacity.acquire(blocking=False):
+            return None
+        try:
+            future = self._executor.submit(function, *args, **kwargs)
+        except RuntimeError:
+            self._capacity.release()
+            raise
+        future.add_done_callback(lambda _future: self._capacity.release())
+        return future
+
+    def shutdown(self, *, wait: bool = True) -> None:
+        self._executor.shutdown(wait=wait)

app/integrations/telegram_agent.py

@@ -0,0 +1,239 @@
+"""Natural-language Telegram agent backed by an OpenAI-compatible model and MCP.
+
+The Telegram conversation pattern is inspired by Luis Rivera's ocabra_telegram:
+https://github.com/luisriverag/ocabra_telegram
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import threading
+import time
+from collections import defaultdict, deque
+from dataclasses import dataclass, field
+from typing import Any
+
+import requests
+
+from app import mcp_server
+
+
+READ_ONLY_TOOLS = {
+    "list_proposals",
+    "current_budget",
+    "get_voting_settings",
+}
+MUTATING_TOOLS = {"create_member", "create_proposal", "create_poll", "update_voting_settings"}
+MAX_TOOL_ROUNDS = 4
+ConversationKey = tuple[int, int]
+_history: dict[ConversationKey, deque[dict[str, Any]]] = defaultdict(lambda: deque(maxlen=12))
+_pending_actions: dict[ConversationKey, "PendingAction"] = {}
+_state_lock = threading.RLock()
+
+
+@dataclass(frozen=True)
+class PendingAction:
+    tool_name: str
+    arguments: dict[str, Any]
+    created_at: float = field(default_factory=time.monotonic)
+
+
+def is_configured() -> bool:
+    return bool(os.getenv("OCABRA_CHAT_URL", "").strip() and mcp_server.MCP_API_KEY)
+
+
+def _headers() -> dict[str, str]:
+    headers = {"Content-Type": "application/json"}
+    api_key = os.getenv("OCABRA_API_KEY", "").strip()
+    if api_key:
+        headers["Authorization"] = f"Bearer {api_key}"
+    return headers
+
+
+def _openai_tools(*, is_admin: bool) -> list[dict[str, Any]]:
+    definitions = mcp_server._tool_definitions()
+    if not is_admin:
+        definitions = [item for item in definitions if item["name"] in READ_ONLY_TOOLS]
+    return [
+        {
+            "type": "function",
+            "function": {
+                "name": item["name"],
+                "description": item["description"],
+                "parameters": item["inputSchema"],
+            },
+        }
+        for item in definitions
+    ]
+
+
+def _call_mcp(tool_name: str, arguments: dict[str, Any], *, is_admin: bool) -> str:
+    allowed = {item["function"]["name"] for item in _openai_tools(is_admin=is_admin)}
+    if tool_name not in allowed:
+        return json.dumps({"error": "This Telegram account may not use that tool."})
+    response = mcp_server.handle_request(
+        {
+            "jsonrpc": "2.0",
+            "id": "telegram-agent",
+            "method": "tools/call",
+            "params": {
+                "name": tool_name,
+                "arguments": arguments,
+                "api_key": mcp_server.MCP_API_KEY,
+            },
+        }
+    )
+    if not response:
+        return json.dumps({"error": "MCP returned no response."})
+    if "error" in response:
+        return json.dumps({"error": response["error"]["message"]})
+    content = response.get("result", {}).get("content", [])
+    return content[0].get("text", "{}") if content else "{}"
+
+
+def _conversation_key(chat_id: int, telegram_user_id: int | None) -> ConversationKey:
+    # Private chats normally use the same value for both IDs. Including the user
+    # ID prevents one member in a group chat from seeing another member's context.
+    return int(chat_id), int(telegram_user_id if telegram_user_id is not None else chat_id)
+
+
+def _confirmation_text(action: PendingAction) -> str:
+    return (
+        f"⚠️ Please confirm MCP action {action.tool_name} with /confirm, "
+        "or discard it with /cancel.\n"
+        f"Arguments: {json.dumps(action.arguments, ensure_ascii=False, sort_keys=True)}"
+    )
+
+
+def _action_result_text(action: PendingAction, result: str) -> str:
+    """Turn MCP JSON into a compact, Telegram-friendly completion message."""
+    try:
+        payload = json.loads(result)
+    except (TypeError, json.JSONDecodeError):
+        payload = result
+    if isinstance(payload, dict) and payload.get("error"):
+        return f"❌ {action.tool_name} failed: {payload['error']}"
+    if isinstance(payload, dict):
+        details = "\n".join(
+            f"{str(key).replace('_', ' ').capitalize()}: "
+            f"{json.dumps(value, ensure_ascii=False) if isinstance(value, (dict, list)) else value}"
+            for key, value in payload.items()
+        )
+    else:
+        details = str(payload)
+    suffix = f"\n{details}" if details and details != "{}" else ""
+    return f"✅ {action.tool_name} completed.{suffix}"
+
+
+def answer(
+    chat_id: int,
+    text: str,
+    *,
+    telegram_user_id: int | None = None,
+    is_admin: bool = False,
+) -> str:
+    """Answer natural language and let the model operate ManaVote through MCP tools."""
+    url = os.getenv("OCABRA_CHAT_URL", "").strip()
+    if not url:
+        raise RuntimeError("Natural-language Telegram support is not configured.")
+
+    key = _conversation_key(chat_id, telegram_user_id)
+    normalized_text = text.strip().lower()
+    with _state_lock:
+        pending = _pending_actions.get(key)
+    if normalized_text in {"/cancel", "cancel"}:
+        with _state_lock:
+            removed = _pending_actions.pop(key, None)
+        return "✅ Pending action cancelled." if removed else "There is no pending action to cancel."
+    if normalized_text in {"/confirm", "confirm"}:
+        if not pending:
+            return "There is no pending action to confirm."
+        confirmation_ttl = float(os.getenv("TELEGRAM_CONFIRM_TTL_SECONDS", "300"))
+        if time.monotonic() - pending.created_at > confirmation_ttl:
+            with _state_lock:
+                _pending_actions.pop(key, None)
+            return "⌛ That pending action expired. Please request it again."
+        if not is_admin:
+            with _state_lock:
+                _pending_actions.pop(key, None)
+            return "❌ Only a linked administrator can confirm that action."
+        result = _call_mcp(pending.tool_name, pending.arguments, is_admin=True)
+        with _state_lock:
+            _pending_actions.pop(key, None)
+        return _action_result_text(pending, result)
+
+    with _state_lock:
+        history_snapshot = list(_history[key])
+    messages: list[dict[str, Any]] = [
+        {
+            "role": "system",
+            "content": os.getenv(
+                "TELEGRAM_AGENT_SYSTEM_PROMPT",
+                "You are the ManaVote Telegram assistant. Use the supplied tools for ManaVote facts and actions. "
+                "Never invent tool results. If a tool says confirmation_required, repeat its confirmation message "
+                "and do not claim the action succeeded. Reply concisely in the user's language using plain text "
+                "that reads well in Telegram.",
+            ),
+        },
+        *history_snapshot,
+        {"role": "user", "content": text},
+    ]
+    tools = _openai_tools(is_admin=is_admin)
+
+    for _ in range(MAX_TOOL_ROUNDS):
+        response = requests.post(
+            url,
+            headers=_headers(),
+            json={
+                "model": os.getenv("OCABRA_MODEL", "ocabra"),
+                "messages": messages,
+                "tools": tools,
+                "tool_choice": "auto",
+                "temperature": 0.2,
+                "stream": False,
+            },
+            timeout=float(os.getenv("OCABRA_TIMEOUT_SECONDS", "60")),
+        )
+        response.raise_for_status()
+        assistant = response.json()["choices"][0]["message"]
+        tool_calls = assistant.get("tool_calls") or []
+        if not tool_calls:
+            reply = (assistant.get("content") or "I couldn't produce a response.").strip()
+            with _state_lock:
+                _history[key].extend(
+                    ({"role": "user", "content": text}, {"role": "assistant", "content": reply})
+                )
+            return reply
+
+        messages.append(assistant)
+        for tool_call in tool_calls:
+            function = tool_call.get("function") or {}
+            try:
+                arguments = json.loads(function.get("arguments") or "{}")
+            except (TypeError, json.JSONDecodeError):
+                arguments = {}
+            if not isinstance(arguments, dict):
+                result = json.dumps({"error": "Tool arguments must be a JSON object."})
+            elif function.get("name") in MUTATING_TOOLS:
+                action = PendingAction(function["name"], arguments)
+                with _state_lock:
+                    _pending_actions[key] = action
+                # Do not rely on the model to reproduce a safety prompt. Return
+                # the exact confirmation instructions immediately and avoid an
+                # unnecessary second model round.
+                return _confirmation_text(action)
+            else:
+                result = _call_mcp(function.get("name", ""), arguments, is_admin=is_admin)
+            messages.append(
+                {"role": "tool", "tool_call_id": tool_call.get("id", ""), "content": result}
+            )
+
+    raise RuntimeError("The assistant exceeded the MCP tool-call limit.")
+
+
+def reset(chat_id: int, telegram_user_id: int | None = None) -> None:
+    key = _conversation_key(chat_id, telegram_user_id)
+    with _state_lock:
+        _history.pop(key, None)
+        _pending_actions.pop(key, None)

app/integrations/telegram_client.py

@@ -3,6 +3,8 @@
 
 
 class TelegramClient:
+    MESSAGE_CHUNK_SIZE = 3900
+
     def __init__(self, bot_token: str, chat_id: str, thread_id: str = ""):
         self.bot_token = bot_token
         self.chat_id = chat_id
@@ -29,6 +31,66 @@ def send_message(self, message: str) -> bool:
         except RequestException:
             return False
 
+    def send_message_with_id(self, message: str) -> int | None:
+        """Send a message and return its Telegram message ID when available."""
+        if not self.bot_token or not self.chat_id:
+            return None
+        url = f"https://api.telegram.org/bot{self.bot_token}/sendMessage"
+        payload = {"chat_id": self.chat_id, "text": message}
+        thread_id = self._thread_id()
+        if thread_id is not None:
+            payload["message_thread_id"] = thread_id
+        try:
+            response = requests.post(url, json=payload, timeout=10)
+            if response.status_code != 200:
+                return None
+            body = response.json()
+            if not body.get("ok"):
+                return None
+            message_id = (body.get("result") or {}).get("message_id")
+            return int(message_id) if message_id is not None else None
+        except (RequestException, TypeError, ValueError):
+            return None
+
+    def delete_message(self, message_id: int | None) -> bool:
+        """Delete a previously sent message, if Telegram returned an ID for it."""
+        if not self.bot_token or not self.chat_id or message_id is None:
+            return False
+        url = f"https://api.telegram.org/bot{self.bot_token}/deleteMessage"
+        try:
+            return self._telegram_ok(
+                url,
+                {"chat_id": self.chat_id, "message_id": int(message_id)},
+            )
+        except (RequestException, TypeError, ValueError):
+            return False
+
+    @classmethod
+    def _message_chunks(cls, message: str) -> list[str]:
+        """Split text below Telegram's limit, preferring readable boundaries."""
+        remaining = str(message or "")
+        if not remaining:
+            return []
+        chunks = []
+        while len(remaining) > cls.MESSAGE_CHUNK_SIZE:
+            boundary = remaining.rfind("\n", 0, cls.MESSAGE_CHUNK_SIZE + 1)
+            if boundary < cls.MESSAGE_CHUNK_SIZE // 2:
+                boundary = remaining.rfind(" ", 0, cls.MESSAGE_CHUNK_SIZE + 1)
+            if boundary < cls.MESSAGE_CHUNK_SIZE // 2:
+                boundary = cls.MESSAGE_CHUNK_SIZE
+            chunks.append(remaining[:boundary].rstrip())
+            remaining = remaining[boundary:].lstrip()
+        if remaining:
+            chunks.append(remaining)
+        return chunks
+
+    def send_long_message(self, message: str) -> bool:
+        """Send every chunk of a potentially long assistant response."""
+        chunks = self._message_chunks(message)
+        if not chunks:
+            return False
+        return all(self.send_message(chunk) for chunk in chunks)
+
     def send_poll_message(self, message: str, poll_id: int, options: list[str]) -> bool:
         if not self.bot_token or not self.chat_id:
             return False