minis-analyse-skill/skills/swads-minis-launch-watch/scripts/swads_mcp.py

"""Minimal Streamable-HTTP client for the swads MCP gateway, so one watch round is one script run.

Credentials are never stored here. Resolution order:
  1. SWADS_MCP_URL + SWADS_MCP_TOKEN environment variables;
  2. the `swads` entry under `mcpServers` in ~/.claude.json (project-scoped first, then global),
     i.e. the same server the operator's Claude Code session already uses;
  3. the `swads` entry under `mcp_servers` in ~/.codex/config.toml (Codex users).
"""
import json
import os
import time
import urllib.error
import urllib.request


def _load_config():
    url, token = os.environ.get("SWADS_MCP_URL"), os.environ.get("SWADS_MCP_TOKEN")
    if url and token:
        return url, {"Authorization": f"Bearer {token}"}
    path = os.path.expanduser("~/.claude.json")
    try:
        cfg = json.load(open(path))
    except (OSError, ValueError):
        cfg = {}
    candidates = []
    cwd = os.environ.get("SWADS_PROJECT_DIR") or os.getcwd()
    projects = cfg.get("projects") or {}
    for proj in [cwd] + [p for p in projects if p != cwd]:
        server = ((projects.get(proj) or {}).get("mcpServers") or {}).get("swads")
        if server:
            candidates.append(server)
    if (cfg.get("mcpServers") or {}).get("swads"):
        candidates.append(cfg["mcpServers"]["swads"])
    for server in candidates:
        if server.get("url", "").startswith("http") and "127.0.0.1" not in server["url"]:
            return server["url"], dict(server.get("headers") or {})
    for server in candidates:
        if server.get("url"):
            return server["url"], dict(server.get("headers") or {})
    try:
        import tomllib
        server = (tomllib.load(open(os.path.expanduser("~/.codex/config.toml"), "rb"))
                  .get("mcp_servers") or {}).get("swads") or {}
    except (OSError, ValueError, ImportError):
        server = {}
    if server.get("url"):
        return server["url"], dict(server.get("http_headers") or {})
    raise SystemExit("找不到 swads MCP 配置:设置 SWADS_MCP_URL / SWADS_MCP_TOKEN,或先在 Claude Code / Codex 里配置 swads MCP 服务器")


URL, HEADERS = _load_config()
_sid = None
_n = 0


def _post_once(body):
    global _sid
    headers = {"Content-Type": "application/json", "Accept": "application/json, text/event-stream", **HEADERS}
    if _sid:
        headers["Mcp-Session-Id"] = _sid
    req = urllib.request.Request(URL, data=json.dumps(body).encode(), headers=headers, method="POST")
    with urllib.request.urlopen(req, timeout=300) as resp:
        if resp.headers.get("Mcp-Session-Id"):
            _sid = resp.headers["Mcp-Session-Id"]
        raw = resp.read().decode()
        ctype = resp.headers.get("Content-Type", "")
    if not raw.strip():
        return None
    if "event-stream" in ctype:
        msgs = [json.loads(line[5:]) for line in raw.splitlines() if line.startswith("data:")]
        return msgs[-1] if msgs else None
    return json.loads(raw)


def _post(body, tries=4):
    """Retry transport failures (SSL EOF, 5xx, dropped connections); never retry a 4xx."""
    for i in range(tries):
        try:
            return _post_once(body)
        except urllib.error.HTTPError as e:
            if e.code < 500 or i == tries - 1:
                raise
        except Exception:
            if i == tries - 1:
                raise
        time.sleep(5 * (i + 1))


def init():
    global _sid
    _sid = None
    _post({"jsonrpc": "2.0", "id": 0, "method": "initialize",
           "params": {"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "minis-launch-watch", "version": "1"}}})
    _post({"jsonrpc": "2.0", "method": "notifications/initialized"})


def call(name, args):
    """Call one gateway tool and return its structured result; raise on a tool error."""
    global _n
    _n += 1
    r = _post({"jsonrpc": "2.0", "id": _n, "method": "tools/call", "params": {"name": name, "arguments": args}})
    if "error" in r:
        raise RuntimeError(json.dumps(r["error"], ensure_ascii=False))
    res = r["result"]
    sc = res.get("structuredContent")
    if sc is not None:
        out = sc.get("result", sc) if isinstance(sc, dict) and list(sc) == ["result"] else sc
    else:
        text = "".join(c.get("text", "") for c in res.get("content", []))
        try:
            out = json.loads(text)
        except ValueError:
            out = text
    if res.get("isError"):
        raise RuntimeError(str(out)[:2000])
    if isinstance(out, dict) and out.get("error_code") and out.get("upstream_status"):
        raise RuntimeError(json.dumps(out, ensure_ascii=False)[:2000])
    return out


init()