"""Weekly digest (roadmap #13): the week in Kanbante, as a note. Composes a markdown digest from the event journal (last 7 days) and the current focus snapshot, then — ONLY after a human approves on /digest (standing rule: cross-app writes need approval) — delivers it to ObsiGate through its MCP server (Kanbante as MCP *client*: initialize → tools/call) or to a generic JSON webhook. The ObsiGate tool contract is configurable (URL, tool name, JSON argument template with {title}/{content} placeholders) because the exact tool names of the target server are an operator decision, not something to hardcode blindly. """ import json import logging import httpx logger = logging.getLogger("kanbante.digest") MCP_PROTOCOL_VERSION = "2024-11-05" def format_digest_markdown( week_label: str, closed_items: list[dict], created_count: int, moved_count: int, focus_groups: list[dict], ) -> str: """Render the weekly digest as markdown (Obsidian-friendly).""" lines = [f"# 🧠 Kanbante — semaine du {week_label}", ""] lines.append(f"## ✅ Livré ({len(closed_items)})") if closed_items: for item in closed_items: title = item.get("title") or "" lines.append( f"- [{item['repo']}#{item['number']}]({item.get('html_url', '')})" f" {title}".rstrip() ) else: lines.append("- Rien n'a été fermé cette semaine.") lines.append("") lines.append("## 🚧 En cours") total_focus = sum(len(g["cards"]) for g in focus_groups) if total_focus: for group in focus_groups: lines.append(f"### {group['full_name']}") for card in group["cards"]: issue = card["issue"] lines.append( f"- #{issue['number']} {issue.get('title', '')}" f" — *{card['column']}*" ) else: lines.append("- Aucune carte en cours, en QA ou en révision.") lines.append("") lines.append("## 📈 Activité de la semaine") lines.append(f"- Issues créées : {created_count}") lines.append(f"- Déplacements de cartes : {moved_count}") lines.append(f"- Issues fermées : {len(closed_items)}") lines.append("") lines.append("*Digest composé par Kanbante à partir de son journal d'événements.*") return "\n".join(lines) def render_mcp_args(template: str, title: str, content: str) -> dict: """Substitute {title}/{content} in a JSON template, safely.""" def esc(value: str) -> str: return json.dumps(value, ensure_ascii=False)[1:-1] rendered = template.replace("{title}", esc(title)).replace( "{content}", esc(content) ) try: args = json.loads(rendered) except ValueError as exc: raise ValueError(f"KANBANTE_OBSIGATE_MCP_ARGS is not valid JSON: {exc}") if not isinstance(args, dict): raise ValueError("KANBANTE_OBSIGATE_MCP_ARGS must be a JSON object") return args async def send_webhook(url: str, title: str, markdown: str, week: str) -> tuple[bool, str]: """POST the digest as JSON to a generic webhook.""" try: async with httpx.AsyncClient(timeout=20) as client: response = await client.post( url, json={"title": title, "markdown": markdown, "week": week} ) if response.status_code < 300: return True, f"webhook HTTP {response.status_code}" return False, f"webhook HTTP {response.status_code}: {response.text[:200]}" except Exception as exc: return False, f"webhook error: {exc}" def _mcp_headers(token: str) -> dict: headers = { "Accept": "application/json, text/event-stream", "Content-Type": "application/json", } if token: headers["Authorization"] = f"Bearer {token}" return headers def _parse_mcp_body(response: httpx.Response) -> dict: """MCP streamable HTTP answers JSON or an SSE frame — accept both.""" text = response.text.strip() if not text: return {} if text.startswith("{"): return json.loads(text) payload: dict = {} for line in text.splitlines(): if line.startswith("data:"): chunk = line[5:].strip() if chunk.startswith("{"): payload = json.loads(chunk) return payload async def _mcp_rpc(client: httpx.AsyncClient, url: str, token: str, method: str, params: dict, req_id: int) -> dict: response = await client.post( url, json={"jsonrpc": "2.0", "id": req_id, "method": method, "params": params}, headers=_mcp_headers(token), ) response.raise_for_status() payload = _parse_mcp_body(response) if "error" in payload: err = payload["error"] or {} raise RuntimeError(f"MCP error {err.get('code')}: {err.get('message')}") return payload.get("result") or {} async def _mcp_session(client: httpx.AsyncClient, url: str, token: str) -> None: await _mcp_rpc( client, url, token, "initialize", { "protocolVersion": MCP_PROTOCOL_VERSION, "capabilities": {}, "clientInfo": {"name": "kanbante", "version": "0.24.0"}, }, req_id=1, ) await client.post( url, json={"jsonrpc": "2.0", "method": "notifications/initialized"}, headers=_mcp_headers(token), ) async def mcp_list_tools(url: str, token: str) -> tuple[bool, list[str] | str]: """(True, [tool names]) or (False, error message).""" try: async with httpx.AsyncClient(timeout=20) as client: await _mcp_session(client, url, token) result = await _mcp_rpc(client, url, token, "tools/list", {}, req_id=2) return True, [t.get("name", "?") for t in result.get("tools", [])] except Exception as exc: return False, f"MCP tools/list failed: {exc}" async def mcp_call_tool( url: str, token: str, tool: str, arguments: dict ) -> tuple[bool, str]: """Call one tool; (True, short result) or (False, error message).""" try: async with httpx.AsyncClient(timeout=30) as client: await _mcp_session(client, url, token) result = await _mcp_rpc( client, url, token, "tools/call", {"name": tool, "arguments": arguments}, req_id=3, ) if result.get("isError"): return False, f"tool {tool} reported an error: {str(result)[:300]}" texts = [ block.get("text", "") for block in result.get("content", []) if isinstance(block, dict) and block.get("type") == "text" ] return True, (" ".join(texts)[:300] or f"tool {tool} OK") except Exception as exc: return False, f"MCP tools/call failed: {exc}"