2fa011fa2a
- chatlogger/api.py — read-only HTTP API (stdlib http.server, bearer-token auth, bound to 100.123.47.7:8770) with /list_chats, /recent, /search, /since, /stats, /healthz - chatlogger/chatlogger-api.service — hardened systemd unit - chatlogger/api.env.example — env template (token + bind + db) - mac-clients/chatlog_mcp.py — stdio MCP wrapper for Hermes Agent on the Mac (4 tools: chatlog_list_chats / recent / search / since), uses only Python stdlib, talks to the REST API via Tailscale. Hermes registers as 4th MCP server (chatlog) — verified live with 76 tool(s) from 4 server(s) in gateway.log.
263 lines
10 KiB
Python
263 lines
10 KiB
Python
"""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()
|