homelab-brain/mac-clients/chatlog_mcp.py

#!/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)"
        )
    lines.append("")
    lines.append(
        "WICHTIG: Diese Liste enthaelt nur Metadaten (Titel + Anzahl). "
        "Um die NACHRICHTEN selbst zu sehen, MUSST du jetzt zwingend einen "
        "zweiten Tool-Call machen: chatlog_recent(chat='<titel-substring>') "
        "fuer die letzten Nachrichten ODER "
        "chatlog_since(chat='<titel-substring>', hours=24) fuer alles seit "
        "den letzten 24h. Antworte NICHT bevor du das gemacht hast."
    )
    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": (
            "Returns ONLY the metadata (title, id, message count) of every "
            "Telegram group being recorded. Does NOT contain any messages. "
            "Use this only when you don't know which group the user means. "
            "After calling this you MUST make a second call to chatlog_recent "
            "or chatlog_since to actually read the messages — the user "
            "wants the conversation content, not just chat titles."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {},
            "required": [],
        },
    },
    {
        "name": "chatlog_recent",
        "description": (
            "PREFERRED tool to actually READ the latest messages of a "
            "recorded Telegram group. Returns a chronological transcript "
            "with timestamps, senders and full message bodies. Use this "
            "for: 'Was wurde heute / zuletzt geschrieben?', 'Aktueller "
            "Stand im <Gruppe>-Chat?', 'Was hat <Person> gesagt?'. "
            "If you don't know the chat title, call chatlog_list_chats "
            "first and then ALWAYS follow up with this tool. "
            "If 'chat' is omitted, returns recent messages across all "
            "groups (works fine when only one group is recorded)."
        ),
        "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())