#!/usr/bin/env python3 """AISquare Explainability — learnings MCP server (initial release). Gives a coding agent (Claude Code, Cursor, anything MCP-capable) direct access to the outcomes-and-learnings surface of the AISquare Explainability platform, so it can read what its agents did, why runs failed rules, what the platform has learned across runs, and feed dispositions back — the iterative-improvement loop, as tools. Setup (three env vars, then register the server over stdio): export EXPLAINABILITY_GATEWAY_URL="https://explainability-api.aisquare.studio" export EXPLAINABILITY_API_KEY="" # workspace keys 403 on reads export AISQUARE_STUDIO_ID="" # e.g. 702 # Claude Code: claude mcp add aisquare-learnings -- python3 aisquare_learnings_mcp.py Dependencies: ``pip install mcp``. HTTP is stdlib urllib — nothing else. Every tool is a read, except two deliberately safe writes: ``run_checkup`` (deterministic, idempotent, stores nothing) and ``send_feedback`` (the disposition write path — an agent that dispositions findings trains the thing that grades it). Nothing here can delete or alter a run. The praxis-backed tools (learnings_context, list_lessons, get_trends) answer 503 "praxis not configured" on deployments without the Praxis learning service; that is a deployment fact, not an error in this server. """ from __future__ import annotations import json import os import time import urllib.error import urllib.parse import urllib.request from typing import Any, Dict, Optional from mcp.server.fastmcp import FastMCP GATEWAY = os.environ.get("EXPLAINABILITY_GATEWAY_URL", "").rstrip("/") API_KEY = os.environ.get("EXPLAINABILITY_API_KEY", "") STUDIO = os.environ.get("AISQUARE_STUDIO_ID", "") mcp = FastMCP("aisquare-learnings") def _call(method: str, path: str, params: Optional[Dict[str, Any]] = None, body: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """One HTTP call with the retry the platform documents: 429 AND 500 are retryable (some throttles surface as 500 with Retry-After stripped).""" if not (GATEWAY and API_KEY and STUDIO): return {"error": "Set EXPLAINABILITY_GATEWAY_URL, EXPLAINABILITY_API_KEY " "and AISQUARE_STUDIO_ID in the environment."} qs = "?" + urllib.parse.urlencode({k: v for k, v in (params or {}).items() if v is not None}) if params else "" url = f"{GATEWAY}{path}{qs}" data = json.dumps(body).encode() if body is not None else None last: Dict[str, Any] = {} for attempt in range(4): req = urllib.request.Request(url, data=data, method=method, headers={ "X-API-KEY": API_KEY, "Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=60) as resp: return json.loads(resp.read().decode() or "{}") except urllib.error.HTTPError as e: detail = e.read().decode(errors="replace")[:500] last = {"error": f"HTTP {e.code}", "detail": detail, "url": url} if e.code in (429, 500): time.sleep(min(2 ** attempt, 8)); continue if e.code == 403 and "Studio ID mismatch" in detail: last["hint"] = ("Your key is a WORKSPACE key; this surface needs a " "STUDIO key. Ask your AISquare contact.") if e.code == 404: if "/praxis/" in path or path.endswith("/checkup"): last["hint"] = ("Not deployed on this gateway version yet — " "the Praxis learning rollout has not reached " "this deployment. Nothing to fix on your side; " "this tool starts answering when it lands.") else: last["hint"] = ("For analyses (rml, policies) a 404 usually " "means 'not computed yet' — retry in 30-60s, " "or read rml/v3 header.status_reasons to " "distinguish pending from failed.") return last except urllib.error.URLError as e: last = {"error": f"connection failed: {e.reason}", "url": url} time.sleep(min(2 ** attempt, 8)) return last def _s(path: str) -> str: return f"/v1/studios/{STUDIO}{path}" # ── what happened ──────────────────────────────────────────────────────────── @mcp.tool() def list_runs(limit: int = 20, has_errors: Optional[bool] = None, has_policies: Optional[bool] = None, agent: Optional[str] = None, date_from: Optional[str] = None) -> dict: """List recent runs, newest first. has_policies=True -> runs that broke a rule; has_errors=True -> runs that blew up. date_from must be timezone-aware ISO-8601 (e.g. 2026-08-01T00:00:00Z); other forms are unreliable and epoch seconds are silently ignored by the API.""" return _call("GET", _s("/ui/runs"), { "limit": limit, "sort": "recent", "has_errors": has_errors, "has_policies": has_policies, "agent": agent, "date_from": date_from}) @mcp.tool() def get_run_graph(run_id: str) -> dict: """The run as nodes+edges (LLM calls, tools, retrievals) with timings, tokens and cost per node — the structural 'what happened, in what order'.""" return _call("GET", _s(f"/runs/{run_id}/graph")) # ── why, and against which rules ───────────────────────────────────────────── @mcp.tool() def get_reasoning(run_id: str) -> dict: """The RML v3 reasoning document: claims, inference chain, assumptions (where the run states something it did NOT verify — the usual source of wrong answers). header.status_reasons says whether AI extraction is pending / failed / complete.""" return _call("GET", _s(f"/runs/{run_id}/rml/v3")) @mcp.tool() def get_verdicts(run_id: str) -> dict: """Per-clause rule-book verdicts under aisquare_rule_book_audit. Read each gate as: passed=false -> FAILED; passed=true,triggered=true -> passed; triggered=false -> not applicable. The gate's `detail` string is the judge's own reasoning — the most useful field for fixing the agent.""" return _call("GET", _s(f"/runs/{run_id}/policies")) @mcp.tool() def get_findings(run_id: str) -> dict: """RML v3 findings with fingerprints (needed for send_feedback) and scores — the machine-readable gaps for this run.""" return _call("GET", _s(f"/runs/{run_id}/rml/v3/findings")) # ── improvement surfaces ───────────────────────────────────────────────────── @mcp.tool() def run_checkup(run_id: str) -> dict: """The pipeline doctor: on-demand, fully deterministic examination of one run — failures, performance vs this agent's own history, known failure patterns, governance decisions. No LLM, nothing stored, idempotent.""" return _call("POST", _s(f"/runs/{run_id}/checkup")) @mcp.tool() def get_optimizations(run_id: str) -> dict: """Heuristic optimization suggestions computed for one run (cost, latency, structure). Empty list = nothing flagged.""" return _call("GET", _s(f"/runs/{run_id}/optimizations")) @mcp.tool() def studio_insights(window: int = 20) -> dict: """Cross-run synthesis over the last `window` runs: which optimization categories keep firing, which policy gates keep failing, cost by model, fleet spend — the 'common learnings' a single-run view can't show.""" return _call("GET", _s("/insights"), {"window": window}) # ── the learning loop (Praxis service) ─────────────────────────────────────── @mcp.tool() def learnings_context(agent_uid: Optional[str] = None, run_id: Optional[str] = None) -> dict: """The Praxis context packet: distilled lessons ready to inject into an agent's prompt/context. Pass agent_uid to scope to one agent; pass run_id so the injection is logged against that run.""" return _call("GET", _s("/praxis/context"), {"agent_uid": agent_uid, "run_id": run_id}) @mcp.tool() def list_lessons(status: Optional[str] = None, agent_uid: Optional[str] = None, limit: int = 50) -> dict: """Lessons the platform has learned from run outcomes. status filters: candidate / active / promoted. Promoted lessons are also armed as observe-first runtime rules.""" return _call("GET", _s("/praxis/insights"), {"status": status, "agent_uid": agent_uid, "limit": limit}) @mcp.tool() def get_trends(status: Optional[str] = None, limit: int = 50) -> dict: """Mechanically computed cost / tokens / latency / reliability trend findings, each carrying the signal and run ids behind every number.""" return _call("GET", _s("/praxis/trends"), {"status": status, "limit": limit}) # ── the write path: close the loop ─────────────────────────────────────────── @mcp.tool() def send_feedback(run_id: str, fingerprint: str, disposition: str, note: Optional[str] = None) -> dict: """Disposition one finding (fingerprint from get_findings). disposition: accepted | dismissed | corrected | suppressed | unsuppressed. Accepted and dismissed accumulate into per-rule precision — this trains the grader.""" body: Dict[str, Any] = {"fingerprint": fingerprint, "disposition": disposition} if note: body["note"] = note return _call("POST", _s(f"/runs/{run_id}/rml/v3/feedback"), body=body) if __name__ == "__main__": mcp.run()