diff --git a/chatlogger/api.env.example b/chatlogger/api.env.example new file mode 100644 index 000000000..9950a9bcd --- /dev/null +++ b/chatlogger/api.env.example @@ -0,0 +1,5 @@ +CHATLOGGER_API_BIND=100.123.47.7 +CHATLOGGER_API_PORT=8770 +CHATLOGGER_API_DB=/opt/chatlogger/chats.db +CHATLOGGER_API_TZ_OFFSET_HOURS=7 +CHATLOGGER_API_TOKEN=replace-me-with-a-long-random-secret diff --git a/chatlogger/api.py b/chatlogger/api.py new file mode 100644 index 000000000..c5ee313dd --- /dev/null +++ b/chatlogger/api.py @@ -0,0 +1,263 @@ +"""Read-only HTTP API for the Arakawa Concierge chat logger. + +Exposes recent / since / search / list_chats over Tailscale so that the +Hermes Agent on the Mac (and any other Tailscale-connected client) can +query the SQLite chat log via a small stdio MCP wrapper. + +- Bound to a Tailscale-only address (no public exposure). +- Bearer-token auth (CHATLOGGER_API_TOKEN). +- Read-only: never writes to the database. +- Implementation uses only the Python standard library (no FastAPI/Flask) + to avoid pulling new deps into the bot virtualenv. +""" +from __future__ import annotations + +import json +import logging +import os +import sys +import time +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any +from urllib.parse import parse_qs, urlparse + +# Reuse Store / formatting from the bot package next to this file. +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from store import Store, StoredMessage, format_messages # noqa: E402 + +LOG = logging.getLogger("chatlogger.api") + +CONFIG = { + "bind_host": os.environ.get("CHATLOGGER_API_BIND", "100.123.47.7"), + "bind_port": int(os.environ.get("CHATLOGGER_API_PORT", "8770")), + "token": os.environ.get("CHATLOGGER_API_TOKEN", ""), + "db_path": os.environ.get("CHATLOGGER_API_DB", "/opt/chatlogger/chats.db"), + "tz_offset_hours": int(os.environ.get("CHATLOGGER_API_TZ_OFFSET_HOURS", "7")), + "default_recent_limit": 30, + "max_recent_limit": 200, + "default_search_limit": 30, + "max_search_limit": 100, + "max_since_hours": 24 * 30, +} + + +def _msg_to_dict(m: StoredMessage) -> dict[str, Any]: + return { + "id": m.id, + "chat_id": m.chat_id, + "chat_title": m.chat_title, + "message_id": m.message_id, + "user_id": m.user_id, + "user_display": m.user_display, + "text": m.text, + "media_type": m.media_type, + "media_caption": m.media_caption, + "reply_to_message_id": m.reply_to_message_id, + "timestamp_utc": m.timestamp_utc, + "edited": bool(m.edit_timestamp_utc), + } + + +def _resolve_chat_id(store: Store, chat_arg: str | None) -> int | None: + """Accept either a numeric chat_id or a (case-insensitive) substring of the title.""" + if not chat_arg: + return None + chat_arg = chat_arg.strip() + try: + return int(chat_arg) + except ValueError: + pass + needle = chat_arg.lower() + for c in store.chats(): + title = (c.get("title") or "").lower() + if needle and needle in title: + return int(c["chat_id"]) + return None + + +class Handler(BaseHTTPRequestHandler): + server_version = "ChatloggerAPI/1.0" + + # ---- helpers --------------------------------------------------------- + def log_message(self, fmt, *args): + LOG.info("%s - %s", self.address_string(), fmt % args) + + def _send_json(self, status: int, payload: Any) -> None: + body = json.dumps(payload, ensure_ascii=False).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.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def _error(self, status: int, message: str) -> None: + self._send_json(status, {"error": message, "status": status}) + + def _check_auth(self) -> bool: + if not CONFIG["token"]: + self._error(HTTPStatus.SERVICE_UNAVAILABLE, + "server has no token configured") + return False + provided = self.headers.get("Authorization", "") + if not provided.startswith("Bearer "): + self._error(HTTPStatus.UNAUTHORIZED, "missing bearer token") + return False + if provided[len("Bearer "):].strip() != CONFIG["token"]: + self._error(HTTPStatus.UNAUTHORIZED, "invalid token") + return False + return True + + def _qs(self) -> dict[str, list[str]]: + return parse_qs(urlparse(self.path).query, keep_blank_values=False) + + @staticmethod + def _first(qs: dict[str, list[str]], key: str, default: str | None = None) -> str | None: + v = qs.get(key) + return v[0] if v else default + + @staticmethod + def _int(qs: dict[str, list[str]], key: str, default: int, lo: int, hi: int) -> int: + try: + v = int(Handler._first(qs, key, str(default)) or default) + except (TypeError, ValueError): + v = default + return max(lo, min(hi, v)) + + # ---- routing --------------------------------------------------------- + def do_GET(self): # noqa: N802 (BaseHTTPRequestHandler API) + path = urlparse(self.path).path.rstrip("/") or "/" + + if path == "/healthz": + self._send_json(HTTPStatus.OK, {"status": "ok", "service": "chatlogger-api"}) + return + + if not self._check_auth(): + return + + try: + store: Store = self.server.store # type: ignore[attr-defined] + qs = self._qs() + tz = CONFIG["tz_offset_hours"] + + if path == "/list_chats": + chats = store.chats() + return self._send_json(HTTPStatus.OK, { + "tz_offset_hours": tz, + "chats": [ + { + "chat_id": c["chat_id"], + "title": c.get("title"), + "type": c.get("type"), + "message_count": c["n"], + "last_message_utc": c["last_ts"], + } + for c in chats + ], + }) + + if path == "/recent": + chat_arg = self._first(qs, "chat") + chat_id = _resolve_chat_id(store, chat_arg) + if chat_arg and chat_id is None: + return self._error(HTTPStatus.NOT_FOUND, + f"no chat matches '{chat_arg}'") + limit = self._int(qs, "limit", + CONFIG["default_recent_limit"], 1, + CONFIG["max_recent_limit"]) + msgs = store.recent(chat_id=chat_id, limit=limit) + return self._send_json(HTTPStatus.OK, { + "tz_offset_hours": tz, + "chat_id": chat_id, + "count": len(msgs), + "messages": [_msg_to_dict(m) for m in msgs], + "transcript": format_messages(msgs, tz_offset_hours=tz), + }) + + if path == "/search": + query = self._first(qs, "query") or self._first(qs, "q") + if not query or not query.strip(): + return self._error(HTTPStatus.BAD_REQUEST, + "missing 'query' parameter") + chat_arg = self._first(qs, "chat") + chat_id = _resolve_chat_id(store, chat_arg) + if chat_arg and chat_id is None: + return self._error(HTTPStatus.NOT_FOUND, + f"no chat matches '{chat_arg}'") + limit = self._int(qs, "limit", + CONFIG["default_search_limit"], 1, + CONFIG["max_search_limit"]) + try: + msgs = store.search(query.strip(), chat_id=chat_id, limit=limit) + except Exception as exc: # FTS5 syntax errors etc. + return self._error(HTTPStatus.BAD_REQUEST, + f"search failed: {exc}") + # search returns newest-first; flip to chronological for transcript + msgs_chrono = list(reversed(msgs)) + return self._send_json(HTTPStatus.OK, { + "tz_offset_hours": tz, + "chat_id": chat_id, + "query": query.strip(), + "count": len(msgs), + "messages": [_msg_to_dict(m) for m in msgs], + "transcript": format_messages(msgs_chrono, tz_offset_hours=tz), + }) + + if path == "/since": + chat_arg = self._first(qs, "chat") + chat_id = _resolve_chat_id(store, chat_arg) + if chat_arg and chat_id is None: + return self._error(HTTPStatus.NOT_FOUND, + f"no chat matches '{chat_arg}'") + hours = self._int(qs, "hours", 12, 1, CONFIG["max_since_hours"]) + limit = self._int(qs, "limit", 500, 1, 5000) + since_ts = int(time.time()) - hours * 3600 + msgs = store.since(chat_id=chat_id, since_ts=since_ts, limit=limit) + return self._send_json(HTTPStatus.OK, { + "tz_offset_hours": tz, + "chat_id": chat_id, + "since_utc": since_ts, + "hours": hours, + "count": len(msgs), + "messages": [_msg_to_dict(m) for m in msgs], + "transcript": format_messages(msgs, tz_offset_hours=tz), + }) + + if path == "/stats": + return self._send_json(HTTPStatus.OK, store.stats()) + + self._error(HTTPStatus.NOT_FOUND, f"unknown route: {path}") + + except Exception as exc: # last-resort catch + LOG.exception("unhandled error on %s", path) + self._error(HTTPStatus.INTERNAL_SERVER_ERROR, f"internal error: {exc}") + + +def main() -> None: + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", + ) + + if not CONFIG["token"]: + LOG.error("CHATLOGGER_API_TOKEN is not set; refusing to start") + sys.exit(2) + if not os.path.exists(CONFIG["db_path"]): + LOG.error("DB file not found: %s", CONFIG["db_path"]) + sys.exit(2) + + store = Store(CONFIG["db_path"]) + server = ThreadingHTTPServer((CONFIG["bind_host"], CONFIG["bind_port"]), Handler) + server.store = store # type: ignore[attr-defined] + LOG.info("chatlogger-api listening on %s:%s (db=%s)", + CONFIG["bind_host"], CONFIG["bind_port"], CONFIG["db_path"]) + try: + server.serve_forever() + except KeyboardInterrupt: + LOG.info("shutting down") + server.server_close() + + +if __name__ == "__main__": + main() diff --git a/chatlogger/chatlogger-api.service b/chatlogger/chatlogger-api.service new file mode 100644 index 000000000..a88448c00 --- /dev/null +++ b/chatlogger/chatlogger-api.service @@ -0,0 +1,25 @@ +[Unit] +Description=Chatlogger read-only HTTP API (Tailscale-only) +After=network-online.target chatlogger-bot@arakawa.service +Wants=network-online.target + +[Service] +Type=simple +WorkingDirectory=/opt/homelab-brain/chatlogger +EnvironmentFile=/opt/chatlogger/api.env +Environment=PYTHONUNBUFFERED=1 +ExecStart=/opt/bot-venv/bin/python /opt/homelab-brain/chatlogger/api.py +Restart=always +RestartSec=5 +KillMode=control-group +TimeoutStopSec=10 + +ProtectSystem=strict +ReadOnlyPaths=/opt/homelab-brain +ReadWritePaths=/opt/chatlogger +ProtectHome=true +PrivateTmp=true +NoNewPrivileges=true + +[Install] +WantedBy=multi-user.target diff --git a/mac-clients/chatlog_mcp.py b/mac-clients/chatlog_mcp.py new file mode 100644 index 000000000..5959c30b4 --- /dev/null +++ b/mac-clients/chatlog_mcp.py @@ -0,0 +1,320 @@ +#!/usr/bin/env python3 +"""Chatlog MCP stdio wrapper for the Hermes Agent. + +Talks to the read-only HTTP API exposed by the Arakawa Concierge chat +logger on CT 116 over Tailscale. Provides four tools so Hermes can +answer questions about live Telegram group conversations: + + chatlog_list_chats — overview of every group being recorded + chatlog_recent — last N messages of a group (or globally) + chatlog_search — FTS5 keyword search over all logged messages + chatlog_since — every message of the last N hours (default 12) + +Configuration via environment variables (set in the Hermes config or +launchd plist): + + CHATLOG_API_BASE default http://100.123.47.7:8770 + CHATLOG_API_TOKEN required, bearer token for the API + +This wrapper has no third-party dependencies; only the Python standard +library is used so it can run with /usr/bin/python3 on macOS. +""" +from __future__ import annotations + +import asyncio +import json +import os +import sys +import urllib.error +import urllib.parse +import urllib.request + +API_BASE = os.environ.get("CHATLOG_API_BASE", "http://100.123.47.7:8770").rstrip("/") +API_TOKEN = os.environ.get("CHATLOG_API_TOKEN", "") +HTTP_TIMEOUT = float(os.environ.get("CHATLOG_API_TIMEOUT", "10")) + + +def _http_get(path: str, params: dict | None = None) -> dict: + qs = "" + if params: + clean = {k: v for k, v in params.items() if v is not None and v != ""} + if clean: + qs = "?" + urllib.parse.urlencode(clean) + url = f"{API_BASE}{path}{qs}" + req = urllib.request.Request( + url, + headers={ + "Accept": "application/json", + "Authorization": f"Bearer {API_TOKEN}", + }, + ) + try: + with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as r: + return json.loads(r.read().decode("utf-8")) + except urllib.error.HTTPError as e: + body = e.read().decode("utf-8", errors="replace") + raise RuntimeError(f"HTTP {e.code} on {path}: {body}") from None + except urllib.error.URLError as e: + raise RuntimeError(f"connection failed: {e.reason}") from None + + +def _format_chats(payload: dict) -> str: + chats = payload.get("chats", []) + if not chats: + return "Keine aufgezeichneten Chats." + lines = [f"{len(chats)} aufgezeichnete Chats:"] + for c in chats: + lines.append( + f"- {c.get('title') or '(ohne Titel)'} " + f"(id={c['chat_id']}, {c['message_count']} Nachrichten)" + ) + return "\n".join(lines) + + +def _format_messages_payload(payload: dict, header: str) -> str: + transcript = payload.get("transcript", "").strip() + count = payload.get("count", 0) + if count == 0 or not transcript: + return f"{header} — keine Treffer." + return f"{header} ({count} Treffer):\n\n{transcript}" + + +# --------------------- tool handlers ---------------------------------- + +def tool_list_chats(_args: dict) -> str: + return _format_chats(_http_get("/list_chats")) + + +def tool_recent(args: dict) -> str: + chat = args.get("chat") + limit = int(args.get("limit", 30)) + payload = _http_get("/recent", {"chat": chat, "limit": limit}) + chat_label = f"chat='{chat}'" if chat else "alle Chats" + return _format_messages_payload( + payload, f"Letzte {limit} Nachrichten ({chat_label})" + ) + + +def tool_search(args: dict) -> str: + query = (args.get("query") or "").strip() + if not query: + return "Fehler: 'query' darf nicht leer sein." + chat = args.get("chat") + limit = int(args.get("limit", 30)) + payload = _http_get( + "/search", {"query": query, "chat": chat, "limit": limit} + ) + chat_label = f", chat='{chat}'" if chat else "" + return _format_messages_payload( + payload, f"Suche '{query}'{chat_label}" + ) + + +def tool_since(args: dict) -> str: + hours = int(args.get("hours", 12)) + chat = args.get("chat") + limit = int(args.get("limit", 500)) + payload = _http_get( + "/since", {"hours": hours, "chat": chat, "limit": limit} + ) + chat_label = f", chat='{chat}'" if chat else "" + return _format_messages_payload( + payload, f"Nachrichten der letzten {hours}h{chat_label}" + ) + + +TOOLS = [ + { + "name": "chatlog_list_chats", + "description": ( + "List every Telegram group that the Arakawa Concierge logger " + "is currently recording. Returns the chat title, internal id " + "and message count. Always call this first when you don't yet " + "know which group the user is asking about." + ), + "inputSchema": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "name": "chatlog_recent", + "description": ( + "Return the most recent messages of a recorded Telegram group " + "as a chronological transcript. Use this for questions like " + "'Was wurde zuletzt geschrieben?' or 'Aktueller Stand?'. " + "If 'chat' is omitted, returns recent messages across all groups." + ), + "inputSchema": { + "type": "object", + "properties": { + "chat": { + "type": "string", + "description": ( + "Numeric chat_id, or a case-insensitive substring " + "of the chat title (e.g. 'Airbnb', 'G 210', 'Concierge')." + ), + }, + "limit": { + "type": "integer", + "description": "Maximum messages to return (1-200, default 30).", + }, + }, + "required": [], + }, + }, + { + "name": "chatlog_search", + "description": ( + "Full-text search over every logged Telegram message (FTS5). " + "Use this for questions like 'wer hat X gesagt?', " + "'was wurde zu Y vereinbart?', 'gab es eine Zusage zu Z?'. " + "Supports SQLite FTS5 query syntax (single words, phrases in " + "double-quotes, AND/OR, prefix with *)." + ), + "inputSchema": { + "type": "object", + "properties": { + "query": { + "type": "string", + "description": "FTS5 search expression.", + }, + "chat": { + "type": "string", + "description": "Optional chat filter (numeric id or title substring).", + }, + "limit": { + "type": "integer", + "description": "Maximum hits to return (1-100, default 30).", + }, + }, + "required": ["query"], + }, + }, + { + "name": "chatlog_since", + "description": ( + "Return every recorded message of the last N hours, oldest " + "first. Use this for 'was war heute?' / 'was wurde seit " + "gestern besprochen?' / 'gibt es offene Fragen seit X Stunden?'." + ), + "inputSchema": { + "type": "object", + "properties": { + "hours": { + "type": "integer", + "description": "Look-back window in hours (1-720, default 12).", + }, + "chat": { + "type": "string", + "description": "Optional chat filter (numeric id or title substring).", + }, + "limit": { + "type": "integer", + "description": "Maximum messages to return (1-5000, default 500).", + }, + }, + "required": [], + }, + }, +] + +HANDLERS = { + "chatlog_list_chats": tool_list_chats, + "chatlog_recent": tool_recent, + "chatlog_search": tool_search, + "chatlog_since": tool_since, +} + + +# --------------------- MCP plumbing ----------------------------------- + +async def handle_request(request: dict) -> dict | None: + method = request.get("method") + req_id = request.get("id") + + if method == "initialize": + return { + "jsonrpc": "2.0", + "id": req_id, + "result": { + "protocolVersion": "2024-11-05", + "capabilities": {"tools": {}}, + "serverInfo": {"name": "chatlog", "version": "1.0.0"}, + }, + } + if method in ("notifications/initialized", "initialized"): + return None + if method == "tools/list": + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"tools": TOOLS}, + } + if method == "tools/call": + params = request.get("params", {}) + name = params.get("name") + args = params.get("arguments", {}) or {} + handler = HANDLERS.get(name) + if handler is None: + text = f"Unbekanntes Tool: {name}" + else: + try: + text = handler(args) + except Exception as exc: # noqa: BLE001 + text = f"Fehler bei {name}: {exc}" + return { + "jsonrpc": "2.0", + "id": req_id, + "result": {"content": [{"type": "text", "text": text}]}, + } + if method == "ping": + return {"jsonrpc": "2.0", "id": req_id, "result": {}} + + return { + "jsonrpc": "2.0", + "id": req_id, + "error": {"code": -32601, "message": f"Method not found: {method}"}, + } + + +async def main() -> None: + if not API_TOKEN: + sys.stderr.write( + "chatlog_mcp: CHATLOG_API_TOKEN env var is empty — refusing to start.\n" + ) + sys.exit(2) + + loop = asyncio.get_event_loop() + reader = asyncio.StreamReader() + proto = asyncio.StreamReaderProtocol(reader) + await loop.connect_read_pipe(lambda: proto, sys.stdin) + + out = sys.stdout.buffer + + while True: + try: + line = await reader.readline() + if not line: + break + line = line.strip() + if not line: + continue + try: + request = json.loads(line) + except json.JSONDecodeError: + continue + response = await handle_request(request) + if response is not None: + out.write((json.dumps(response, ensure_ascii=False) + "\n").encode("utf-8")) + out.flush() + except (EOFError, BrokenPipeError): + break + except Exception as exc: # noqa: BLE001 + sys.stderr.write(f"chatlog_mcp error: {exc}\n") + sys.stderr.flush() + + +if __name__ == "__main__": + asyncio.run(main())