1#!/usr/bin/env python3
2import argparse
3import base64
4import fcntl
5import hashlib
6import http.cookiejar
7import json
8import os
9from pathlib import Path
10import secrets
11import subprocess
12import sys
13import tempfile
14import time
15import urllib.error
16import urllib.parse
17import urllib.request
18
19ORIGIN = "https://snowglobe.paperclover.net"
20RESOURCE = ORIGIN + "/mcp/shale"
21CALLBACK = "http://127.0.0.1:8766/callback"
22STATE = Path.home() / ".cache/shale-ai/mcp.json"
23
24
25class Redirect(urllib.request.HTTPRedirectHandler):
26 def redirect_request(self, req, fp, code, msg, headers, newurl):
27 url = urllib.parse.urlsplit(newurl)
28 if url.scheme != "https" or url.netloc not in {"snowglobe.paperclover.net", "shale.paperclover.net"}:
29 raise RuntimeError("Sign-in redirected outside Snowglobe and Shale.")
30 return super().redirect_request(req, fp, code, msg, headers, newurl)
31
32
33def request(opener, path, data=None, *, form=False, headers=None):
34 url = path if path.startswith("https://") else ORIGIN + path
35 parsed = urllib.parse.urlsplit(url)
36 if parsed.scheme != "https" or parsed.netloc not in {"snowglobe.paperclover.net", "shale.paperclover.net"}:
37 raise RuntimeError("Only this Snowglobe and Shale instance are allowed.")
38 encoded = None
39 configured = dict(headers or {})
40 if data is not None:
41 encoded = urllib.parse.urlencode(data).encode() if form else json.dumps(data).encode()
42 configured.update({"Origin": ORIGIN, "Content-Type": "application/x-www-form-urlencoded" if form else "application/json"})
43 with opener.open(urllib.request.Request(url, encoded, configured), timeout=30) as response:
44 body = response.read().decode()
45 if not body:
46 value = None
47 elif response.headers.get_content_type() == "text/event-stream":
48 values = [json.loads(line[6:]) for line in body.splitlines() if line.startswith("data: ")]
49 value = next((v for v in values if "id" in v), None)
50 elif response.headers.get_content_type() == "application/json":
51 value = json.loads(body)
52 else:
53 value = None
54 return value, response.url, response.headers
55
56
57def save(state):
58 with tempfile.NamedTemporaryFile(mode="w", dir=STATE.parent, delete=False) as file:
59 json.dump(state, file)
60 os.replace(file.name, STATE)
61
62
63def login(opener, state):
64 password = subprocess.run(
65 ["security", "find-generic-password", "-a", "ai", "-s", "net.paperclover.shale.ai", "-w"],
66 text=True, capture_output=True, check=True,
67 ).stdout.rstrip("\n")
68 auth, _, _ = request(opener, "/auth/status")
69 request(opener, "/auth/password", {"username": "ai", "password": password, "csrf": auth["csrf"], "next": "/mcp"})
70 account, _, _ = request(opener, "/api/account")
71 if account["username"] != "ai" or account["requiredActions"] or any(g["name"] == "infra-admin" for g in account["groups"]):
72 raise RuntimeError("Expected the standard ai account with completed password setup.")
73 linked, _, _ = request(opener, "/api/mcp/shale")
74 if not linked["linked"]:
75 link, _, _ = request(opener, "/api/mcp/shale", {})
76 request(opener, link["redirect"])
77 linked, _, _ = request(opener, "/api/mcp/shale")
78 if not linked["linked"]:
79 raise RuntimeError("The ai Shale account could not be linked.")
80 if "client_id" not in state:
81 client, _, _ = request(opener, "/oauth/register", {
82 "client_name": "Codex Shale issues", "redirect_uris": [CALLBACK],
83 "token_endpoint_auth_method": "none", "grant_types": ["authorization_code", "refresh_token"],
84 "response_types": ["code"],
85 })
86 state["client_id"] = client["client_id"]
87 save(state)
88 verifier = secrets.token_urlsafe(32)
89 challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=")
90 nonce = secrets.token_urlsafe(24)
91 _, destination, _ = request(opener, "/oauth/authorize?" + urllib.parse.urlencode({
92 "client_id": state["client_id"], "redirect_uri": CALLBACK, "response_type": "code",
93 "resource": RESOURCE, "scope": "shale:read shale:write offline_access",
94 "code_challenge": challenge, "code_challenge_method": "S256", "state": nonce,
95 }))
96 path = urllib.parse.urlsplit(destination).path
97 if not path.startswith("/connect/") or not path.removeprefix("/connect/"):
98 raise RuntimeError("The Shale consent destination changed.")
99 consent = "/api/mcp/consent/" + path.removeprefix("/connect/")
100 request(opener, consent)
101 result, _, _ = request(opener, consent, {"resources": "all"})
102 returned = urllib.parse.urlsplit(result["redirect"])
103 query = urllib.parse.parse_qs(returned.query)
104 if urllib.parse.urlunsplit(returned._replace(query="")) != CALLBACK or query.get("state") != [nonce]:
105 raise RuntimeError("The consent response did not match this sign-in.")
106 tokens, _, _ = request(opener, "/oauth/token", {
107 "grant_type": "authorization_code", "client_id": state["client_id"], "code": query["code"][0],
108 "redirect_uri": CALLBACK, "code_verifier": verifier, "resource": RESOURCE,
109 }, form=True)
110 request(opener, "/auth/sign-out", {})
111 return tokens
112
113
114def token(force=False):
115 STATE.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
116 with open(STATE.with_suffix(".lock"), "a") as lock:
117 os.chmod(lock.name, 0o600)
118 fcntl.flock(lock, fcntl.LOCK_EX)
119 state = {}
120 if STATE.exists():
121 stat = STATE.stat()
122 if stat.st_uid != os.getuid() or stat.st_mode & 0o077:
123 raise RuntimeError("The token cache must belong to this user and have mode 600.")
124 state = json.loads(STATE.read_text())
125 if not force and state.get("expires", 0) > time.time() + 60:
126 return state["access_token"]
127 opener = urllib.request.build_opener(Redirect(), urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
128 if not force and state.get("refresh_token"):
129 tokens, _, _ = request(opener, "/oauth/token", {
130 "grant_type": "refresh_token", "client_id": state["client_id"],
131 "refresh_token": state["refresh_token"], "resource": RESOURCE,
132 }, form=True)
133 else:
134 tokens = login(opener, state)
135 state.update(tokens)
136 state["expires"] = time.time() + tokens["expires_in"]
137 save(state)
138 return state["access_token"]
139
140
141def mcp(access, method, params=None):
142 opener = urllib.request.build_opener(Redirect())
143 headers = {"Authorization": "Bearer " + access, "Accept": "application/json, text/event-stream"}
144 initialized, _, response = request(opener, RESOURCE, {
145 "jsonrpc": "2.0", "id": 1, "method": "initialize",
146 "params": {"protocolVersion": "2025-03-26", "capabilities": {}, "clientInfo": {"name": "shale-issues", "version": "1"}},
147 }, headers=headers)
148 if not initialized or "error" in initialized:
149 raise RuntimeError("Shale MCP initialization failed.")
150 headers["MCP-Protocol-Version"] = initialized["result"]["protocolVersion"]
151 if response.get("Mcp-Session-Id"):
152 headers["Mcp-Session-Id"] = response["Mcp-Session-Id"]
153 request(opener, RESOURCE, {"jsonrpc": "2.0", "method": "notifications/initialized"}, headers=headers)
154 result, _, _ = request(opener, RESOURCE, {"jsonrpc": "2.0", "id": 2, "method": method, "params": params or {}}, headers=headers)
155 if not result or "error" in result:
156 raise RuntimeError("Shale MCP rejected the request.")
157 return result["result"]
158
159
160def main():
161 parser = argparse.ArgumentParser()
162 commands = parser.add_subparsers(dest="command", required=True)
163 commands.add_parser("login").add_argument("--reauthorize", action="store_true")
164 commands.add_parser("tools")
165 call = commands.add_parser("call")
166 call.add_argument("tool", choices=["list_repositories", "list_issues", "get_issue", "create_issue", "comment_issue", "set_issue_status", "set_issue_title"])
167 call.add_argument("--args-file", type=Path)
168 args = parser.parse_args()
169 access = token(force=args.command == "login" and args.reauthorize)
170 if args.command == "call":
171 values = json.loads(args.args_file.read_text() if args.args_file else sys.stdin.read())
172 result = mcp(access, "tools/call", {"name": args.tool, "arguments": values})
173 else:
174 result = mcp(access, "tools/list")
175 if args.command == "login":
176 result = {"account": "ai", "catalog": "shale", "tools": [t["name"] for t in result["tools"]]}
177 print(json.dumps(result, indent=2))
178 if result.get("isError"):
179 raise SystemExit(1)
180
181
182if __name__ == "__main__":
183 try:
184 main()
185 except urllib.error.HTTPError as error:
186 path = urllib.parse.urlsplit(error.url).path
187 print(f"HTTP {error.code} at {path}; no write was automatically retried.", file=sys.stderr)
188 raise SystemExit(1)
189 except (OSError, ValueError, KeyError, RuntimeError, subprocess.CalledProcessError) as error:
190 print(f"Shale issues: {type(error).__name__}: {error if not isinstance(error, subprocess.CalledProcessError) else 'Keychain credential unavailable'}", file=sys.stderr)
191 raise SystemExit(1)