fnt/gistfile1.txt

gistfile1.txt · 892 B · Text Raw

1	Шаги
2	1. Взять mcp-token.py. Нужен только python3, внешних зависимостей нет.
3	2. Один раз запустить python3 mcp-token.py в терминале. Откроется браузер, логин обычный, с 2FA. На macOS страница откроется сама, на Linux и Windows URL и код печатаются в терминал.
4	3. Прописать в Claude Code (~/.claude.json на user scope или .mcp.json проекта):
5	{"mcpServers":{"superset":{"type":"http","url":"https://superset.swiftdrive-tech.ru/mcp",
6	"headersHelper":"python3 /путь/к/mcp-token.py --header"}}}
7	Дальше токен обновляется сам. Когда закончится сессия Keycloak, Claude Code покажет ошибку подключения. Тогда снова выполнить шаг 2.

gistfile2.txt · 4.0 KiB · Text Raw

#!/usr/bin/env python3
"""Bearer token for the Superset MCP server (https://superset.swiftdrive-tech.ru/mcp).

Keycloak realm `infra`, public client `superset-mcp`, Device Authorization Grant.
First run opens a browser login (realm 2FA applies); after that the cached
refresh token renews silently until the Keycloak session ends.

mcp-token.py            print the access token
  mcp-token.py --header   print {"Authorization": "Bearer ..."} (Claude Code headersHelper)
  mcp-token.py --logout   drop the cached token
"""
import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

ISSUER = os.environ.get("SUPERSET_MCP_ISSUER", "https://auth.swiftdrive-tech.ru/realms/infra")
CLIENT_ID = os.environ.get("SUPERSET_MCP_CLIENT_ID", "superset-mcp")
CACHE = os.path.expanduser("~/.cache/superset-mcp/token.json")
MIN_TTL = 300  # renew when fewer seconds than this remain

def post(path, **form):
    body = urllib.parse.urlencode({"client_id": CLIENT_ID, **form}).encode()
    req = urllib.request.Request(f"{ISSUER}/protocol/openid-connect/{path}", body)
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.load(resp)
    except urllib.error.HTTPError as e:
        try:
            return json.load(e)
        except ValueError:
            return {"error": f"HTTP {e.code}"}
    except urllib.error.URLError as e:
        sys.exit(f"superset-mcp: {ISSUER} unreachable: {e.reason}")

def save(tok):
    now = time.time()
    tok["expires_at"] = now + tok["expires_in"]
    tok["refresh_expires_at"] = now + tok.get("refresh_expires_in", 0)
    cache_dir = os.path.dirname(CACHE)
    os.makedirs(cache_dir, mode=0o700, exist_ok=True)
    os.chmod(cache_dir, 0o700)
    tmp = f"{CACHE}.{os.getpid()}"
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(tok, f)
    os.replace(tmp, CACHE)  # concurrent headersHelper runs never see a torn file
    return tok

def load():
    try:
        with open(CACHE) as f:
            return json.load(f)
    except (OSError, ValueError):
        return None

def device_login():
    dev = post("auth/device")
    if "device_code" not in dev:
        sys.exit(f"device authorization failed: {dev}")
    url = dev.get("verification_uri_complete") or dev["verification_uri"]
    print(f"Superset MCP login: open {url} (code {dev['user_code']})", file=sys.stderr)
    if sys.platform == "darwin":
        subprocess.run(["open", url], check=False)
    interval = dev.get("interval", 5)
    deadline = time.time() + dev.get("expires_in", 600)
    while time.time() < deadline:
        time.sleep(interval)
        tok = post("token", grant_type="urn:ietf:params:oauth:grant-type:device_code",
                   device_code=dev["device_code"])
        if "access_token" in tok:
            return save(tok)
        err = tok.get("error")
        if err == "slow_down":
            interval += 5
        elif err != "authorization_pending":
            sys.exit(f"login failed: {tok}")
    sys.exit("login timed out")

def token():
    tok = load()
    now = time.time()
    if tok and tok["expires_at"] - now > MIN_TTL:
        return tok
    if tok and tok.get("refresh_token") and tok["refresh_expires_at"] > now:
        new = post("token", grant_type="refresh_token", refresh_token=tok["refresh_token"])
        if "access_token" in new:
            return save(new)
    if not sys.stdin.isatty() and not sys.stderr.isatty():
        # headersHelper runs non-interactively: don't hang on a login nobody sees
        sys.exit("superset-mcp: not logged in, run mcp-token.py in a terminal")
    return device_login()

def main():
    arg = sys.argv[1] if len(sys.argv) > 1 else ""
    if arg == "--logout":
        if os.path.exists(CACHE):
            os.remove(CACHE)
        return
    access = token()["access_token"]
    print(json.dumps({"Authorization": f"Bearer {access}"}) if arg == "--header" else access)

if __name__ == "__main__":
    main()

1	#!/usr/bin/env python3
2	"""Bearer token for the Superset MCP server (https://superset.swiftdrive-tech.ru/mcp).
3
4	Keycloak realm `infra`, public client `superset-mcp`, Device Authorization Grant.
5	First run opens a browser login (realm 2FA applies); after that the cached
6	refresh token renews silently until the Keycloak session ends.
7
8	mcp-token.py print the access token
9	mcp-token.py --header print {"Authorization": "Bearer ..."} (Claude Code headersHelper)
10	mcp-token.py --logout drop the cached token
11	"""
12	import json
13	import os
14	import subprocess
15	import sys
16	import time
17	import urllib.error
18	import urllib.parse
19	import urllib.request
20
21	ISSUER = os.environ.get("SUPERSET_MCP_ISSUER", "https://auth.swiftdrive-tech.ru/realms/infra")
22	CLIENT_ID = os.environ.get("SUPERSET_MCP_CLIENT_ID", "superset-mcp")
23	CACHE = os.path.expanduser("~/.cache/superset-mcp/token.json")
24	MIN_TTL = 300 # renew when fewer seconds than this remain
25
26
27	def post(path, **form):
28	body = urllib.parse.urlencode({"client_id": CLIENT_ID, **form}).encode()
29	req = urllib.request.Request(f"{ISSUER}/protocol/openid-connect/{path}", body)
30	try:
31	with urllib.request.urlopen(req, timeout=30) as resp:
32	return json.load(resp)
33	except urllib.error.HTTPError as e:
34	try:
35	return json.load(e)
36	except ValueError:
37	return {"error": f"HTTP {e.code}"}
38	except urllib.error.URLError as e:
39	sys.exit(f"superset-mcp: {ISSUER} unreachable: {e.reason}")
40
41
42	def save(tok):
43	now = time.time()
44	tok["expires_at"] = now + tok["expires_in"]
45	tok["refresh_expires_at"] = now + tok.get("refresh_expires_in", 0)
46	cache_dir = os.path.dirname(CACHE)
47	os.makedirs(cache_dir, mode=0o700, exist_ok=True)
48	os.chmod(cache_dir, 0o700)
49	tmp = f"{CACHE}.{os.getpid()}"
50	fd = os.open(tmp, os.O_WRONLY \| os.O_CREAT \| os.O_TRUNC, 0o600)
51	with os.fdopen(fd, "w") as f:
52	json.dump(tok, f)
53	os.replace(tmp, CACHE) # concurrent headersHelper runs never see a torn file
54	return tok
55
56
57	def load():
58	try:
59	with open(CACHE) as f:
60	return json.load(f)
61	except (OSError, ValueError):
62	return None
63
64
65	def device_login():
66	dev = post("auth/device")
67	if "device_code" not in dev:
68	sys.exit(f"device authorization failed: {dev}")
69	url = dev.get("verification_uri_complete") or dev["verification_uri"]
70	print(f"Superset MCP login: open {url} (code {dev['user_code']})", file=sys.stderr)
71	if sys.platform == "darwin":
72	subprocess.run(["open", url], check=False)
73	interval = dev.get("interval", 5)
74	deadline = time.time() + dev.get("expires_in", 600)
75	while time.time() < deadline:
76	time.sleep(interval)
77	tok = post("token", grant_type="urn:ietf:params:oauth:grant-type:device_code",
78	device_code=dev["device_code"])
79	if "access_token" in tok:
80	return save(tok)
81	err = tok.get("error")
82	if err == "slow_down":
83	interval += 5
84	elif err != "authorization_pending":
85	sys.exit(f"login failed: {tok}")
86	sys.exit("login timed out")
87
88
89	def token():
90	tok = load()
91	now = time.time()
92	if tok and tok["expires_at"] - now > MIN_TTL:
93	return tok
94	if tok and tok.get("refresh_token") and tok["refresh_expires_at"] > now:
95	new = post("token", grant_type="refresh_token", refresh_token=tok["refresh_token"])
96	if "access_token" in new:
97	return save(new)
98	if not sys.stdin.isatty() and not sys.stderr.isatty():
99	# headersHelper runs non-interactively: don't hang on a login nobody sees
100	sys.exit("superset-mcp: not logged in, run mcp-token.py in a terminal")
101	return device_login()
102
103
104	def main():
105	arg = sys.argv[1] if len(sys.argv) > 1 else ""
106	if arg == "--logout":
107	if os.path.exists(CACHE):
108	os.remove(CACHE)
109	return
110	access = token()["access_token"]
111	print(json.dumps({"Authorization": f"Bearer {access}"}) if arg == "--header" else access)
112
113
114	if __name__ == "__main__":
115	main()
116