| """ |
| KanBot web view — live board over Postgres LISTEN/NOTIFY. |
| |
| Local-first, dependency-light: a single asyncio server that |
| - serves the static branded SPA (index.html) over HTTP, |
| - LISTENs on 'kb_events' and pushes every transition to connected browsers |
| over a raw WebSocket (RFC 6455, hand-rolled — no node, no npm, no build step), |
| - answers a tiny JSON snapshot endpoint so a fresh page loads the full board. |
| |
| Why not Next.js: this is a dashboard, not a marketing site. One file, runs on the |
| M4 with the same asyncpg the rest of the stack uses. Swap it for a Next app later |
| without touching the core — it's just another MCP-contract client. |
| |
| Run: python -m web.server then open http://localhost:8800 |
| """ |
| from __future__ import annotations |
|
|
| import asyncio |
| import base64 |
| import hashlib |
| import json |
| import os |
| import struct |
| from pathlib import Path |
|
|
| import asyncpg |
|
|
| HOST = os.environ.get("KANBOT_WEB_HOST", "127.0.0.1") |
| PORT = int(os.environ.get("KANBOT_WEB_PORT", "8800")) |
| DSN = os.environ.get("DATABASE_URL", "postgresql://kanbot:kanbot@localhost:5432/kanbot") |
| WS_MAGIC = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" |
| INDEX = (Path(__file__).parent / "index.html").read_text() |
|
|
| _clients: set[asyncio.StreamWriter] = set() |
|
|
|
|
| |
| |
| |
| async def board_snapshot(pool: asyncpg.Pool) -> dict: |
| async with pool.acquire() as c: |
| rows = await c.fetch( |
| 'SELECT id,title,"column",agent_role,assigned_worker,attempt,' |
| "requires_human,tags FROM cards ORDER BY created_at") |
| cols: dict[str, list] = {k: [] for k in |
| ("backlog", "ready", "in-flight", "review", "blocked", "done")} |
| for r in rows: |
| cols[r["column"]].append({ |
| "id": r["id"], "title": r["title"], "role": r["agent_role"], |
| "worker": r["assigned_worker"], "attempt": r["attempt"], |
| "requires_human": r["requires_human"], "tags": json.loads(r["tags"]), |
| }) |
| return cols |
|
|
|
|
| |
| |
| |
| def _ws_frame(text: str) -> bytes: |
| data = text.encode() |
| n = len(data) |
| header = bytearray([0x81]) |
| if n < 126: |
| header.append(n) |
| elif n < 65536: |
| header.append(126) |
| header += struct.pack(">H", n) |
| else: |
| header.append(127) |
| header += struct.pack(">Q", n) |
| return bytes(header) + data |
|
|
|
|
| async def _broadcast(text: str) -> None: |
| frame = _ws_frame(text) |
| dead = [] |
| for w in list(_clients): |
| try: |
| w.write(frame) |
| await w.drain() |
| except Exception: |
| dead.append(w) |
| for w in dead: |
| _clients.discard(w) |
|
|
|
|
| |
| |
| |
| async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter, |
| pool: asyncpg.Pool) -> None: |
| try: |
| req = await reader.readuntil(b"\r\n\r\n") |
| except (asyncio.IncompleteReadError, ConnectionResetError): |
| writer.close(); return |
| head = req.decode(errors="ignore") |
| line = head.split("\r\n", 1)[0] |
| try: |
| _, path, _ = line.split(" ", 2) |
| except ValueError: |
| writer.close(); return |
|
|
| |
| if "upgrade: websocket" in head.lower(): |
| key = "" |
| for h in head.split("\r\n"): |
| if h.lower().startswith("sec-websocket-key:"): |
| key = h.split(":", 1)[1].strip() |
| accept = base64.b64encode( |
| hashlib.sha1((key + WS_MAGIC).encode()).digest()).decode() |
| writer.write( |
| b"HTTP/1.1 101 Switching Protocols\r\n" |
| b"Upgrade: websocket\r\nConnection: Upgrade\r\n" |
| b"Sec-WebSocket-Accept: " + accept.encode() + b"\r\n\r\n") |
| await writer.drain() |
| _clients.add(writer) |
| |
| writer.write(_ws_frame(json.dumps( |
| {"type": "snapshot", "board": await board_snapshot(pool)}))) |
| await writer.drain() |
| try: |
| while not reader.at_eof(): |
| await asyncio.sleep(30) |
| finally: |
| _clients.discard(writer) |
| return |
|
|
| |
| if path == "/snapshot": |
| body = json.dumps(await board_snapshot(pool)).encode() |
| ctype = b"application/json" |
| else: |
| body = INDEX.encode() |
| ctype = b"text/html; charset=utf-8" |
| writer.write(b"HTTP/1.1 200 OK\r\nContent-Type: " + ctype + |
| b"\r\nContent-Length: " + str(len(body)).encode() + |
| b"\r\nConnection: close\r\n\r\n" + body) |
| await writer.drain() |
| writer.close() |
|
|
|
|
| async def main() -> None: |
| pool = await asyncpg.create_pool(DSN, min_size=1, max_size=4) |
| listen_conn = await asyncpg.connect(DSN) |
|
|
| async def on_notify(_c, _pid, _chan, payload): |
| |
| await _broadcast(json.dumps( |
| {"type": "event", "event": json.loads(payload), |
| "board": await board_snapshot(pool)})) |
|
|
| await listen_conn.add_listener("kb_events", on_notify) |
|
|
| server = await asyncio.start_server( |
| lambda r, w: handle(r, w, pool), HOST, PORT) |
| print(f"[web] KanBot board live at http://{HOST}:{PORT}") |
| async with server: |
| await server.serve_forever() |
|
|
|
|
| if __name__ == "__main__": |
| asyncio.run(main()) |
|
|
