causeries.py

#!/usr/bin/env python3

import json, os, sys
from dotenv import load_dotenv

load_dotenv()
from http.server import HTTPServer, BaseHTTPRequestHandler
from string import Template
import requests

PORT  = int(os.getenv("PORT", "8080"))
TOKEN = (os.getenv("KAGI_API_KEY") or os.getenv("KAGI_TOKEN") or "").strip()
if not TOKEN:
    sys.exit("KAGI_API_KEY not found. Copy .env.example to .env and add your API key:\n  cp .env.example .env && nano .env")
# Remove "Bot " prefix if user accidentally included it
if TOKEN.startswith("Bot "):
    TOKEN = TOKEN[4:]

HTML_PAGE = Template(
    open(os.path.join(os.path.dirname(__file__), "index.html")).read()
).safe_substitute(__TOKEN_PROVIDED__=1).encode()

KAGI_API_URL = "https://kagi.com/api/v0/fastgpt"
KAGI_HEADERS = {"Authorization": f"Bot {TOKEN}"}


def fastgpt(query: str) -> dict:
    """Call Kagi FastGPT API and return the response data."""
    response = requests.post(KAGI_API_URL, headers=KAGI_HEADERS, json={"query": query})
    response.raise_for_status()
    data = response.json()["data"]
    return {"output": data["output"], "references": data.get("references", [])}


def format_conversation(history: list) -> str:
    """Format conversation history into a single query for FastGPT."""
    parts = []
    for msg in history:
        role = "User" if msg["role"] == "user" else "Assistant"
        parts.append(f"{role}: {msg['content']}")
    return "\n\n".join(parts)


# Validate API key at startup
try:
    fastgpt("test")["output"]
    print("API key validated successfully")
except requests.HTTPError as e:
    if e.response.status_code == 401:
        sys.exit("Invalid API key - got 401 Unauthorized from Kagi API. Please check your KAGI_API_KEY or KAGI_TOKEN.")
    else:
        sys.exit(f"API error: {e}")
except Exception as e:
    print(f"Warning: API validation failed: {e}")


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path != "/":
            return self.send_error(404)
        self.send_response(200)
        self.send_header("Content-Type", "text/html; charset=utf-8")
        self.send_header("Content-Length", str(len(HTML_PAGE)))
        self.end_headers()
        self.wfile.write(HTML_PAGE)

    def do_POST(self):
        if self.path != "/ask":
            return self.send_error(404)
        data = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        history = data.get("history", [])
        query = format_conversation(history)
        try:
            result = fastgpt(query)
            reply = result["output"]
            references = result["references"]
        except requests.HTTPError as e:
            if e.response.status_code == 401:
                reply = "Error: Invalid API key. Please check your KAGI_API_KEY configuration."
            else:
                reply = f"Error: {e}"
            references = []
        except Exception as e:
            reply = f"Error: {e}"
            references = []
        body = json.dumps({"reply": reply, "references": references}).encode()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)

    def log_message(self, *a, **k):
        pass  # quiet


httpd = HTTPServer(("0.0.0.0", PORT), Handler)
print(f"Serving on http://localhost:{PORT}")
try:
    httpd.serve_forever()
except KeyboardInterrupt:
    print("\nShutting down…")