完整示例

curl -O https://tokentosea.com/app/examples/tokentosea-connect.mjs
node server.mjs

import http from 'node:http';
import { randomBytes, randomUUID } from 'node:crypto';
import { beginLogin, finishLogin, refreshTokens, callTool, services } from './tokentosea-connect.mjs';

const config = {
  clientId: process.env.TTS_CLIENT_ID,
  clientSecret: process.env.TTS_CLIENT_SECRET || undefined, // 公共客户端不设置
  redirectUri: 'http://localhost:3000/callback',
};
const SCOPES = ['openid', 'profile', 'offline_access', 'tools:read', 'tools:invoke', 'wallet:read'];
const sessions = new Map(); // 演示用内存会话；生产环境请用加密的服务端会话存储

function sessionOf(req, res) {
  const sid = /(?:^|;\s*)sid=([\w-]{43})(?:;|$)/.exec(req.headers.cookie ?? '')?.[1];
  if (sid && sessions.has(sid)) return sessions.get(sid);
  const id = randomBytes(32).toString('base64url');
  const session = { id };
  sessions.set(id, session);
  res.setHeader('Set-Cookie', `sid=${id}; HttpOnly; SameSite=Lax; Path=/`);
  return session;
}

async function accessToken(session) {
  if (Date.now() < session.expiresAt - 60_000) return session.tokens.access_token;
  if (!session.tokens.refresh_token) throw new Error('未授予 offline_access，请重新登录');
  // 刷新令牌只能使用一次：同一会话的刷新必须串行，并原子替换
  const tokens = await refreshTokens(session.tokens.refresh_token, config);
  session.tokens = tokens;
  session.expiresAt = Date.now() + tokens.expires_in * 1000;
  return tokens.access_token;
}

async function connect(session, path) {
  const response = await fetch(`${services}${path}`, { headers: { Authorization: `Bearer ${await accessToken(session)}` } });
  if (!response.ok) throw new Error(`${path} → ${response.status} ${await response.text()}`);
  return response.json();
}

const send = (res, status, body) => res.writeHead(status, { 'Content-Type': 'application/json; charset=utf-8' }).end(JSON.stringify(body, null, 2));

http.createServer(async (req, res) => {
  const url = new URL(req.url, 'http://localhost:3000');
  const session = sessionOf(req, res);
  try {
    if (url.pathname === '/login') {
      const { url: authorizeUrl, transaction } = beginLogin(config.clientId, config.redirectUri, SCOPES);
      session.transaction = transaction;
      res.writeHead(302, { Location: authorizeUrl }).end();
      return;
    }
    if (url.pathname === '/callback') {
      const transaction = session.transaction;
      delete session.transaction; // 每次登录的随机值只用一次
      if (!transaction) throw new Error('没有进行中的登录');
      const { tokens, identity } = await finishLogin(url.href, transaction, config); // 校验 state、iss、ID Token
      Object.assign(session, { tokens, identity, expiresAt: Date.now() + tokens.expires_in * 1000 });
      res.writeHead(302, { Location: '/' }).end();
      return;
    }
    if (!session.tokens) {
      res.writeHead(200, { 'Content-Type': 'text/html; charset=utf-8' }).end('<a href="/login">用 TokenToSea 登录</a>');
      return;
    }
    if (url.pathname === '/tool') {
      const actions = await connect(session, '/tools/public_hn/actions');
      const action = actions.find((a) => a.id === 'hn.top');
      const run = await callTool(await accessToken(session), action, { limit: 3 }, randomUUID());
      send(res, 200, { status: run.status, chargedUsd: run.chargedUsd, output: run.output, errorCode: run.errorCode });
      return;
    }
    const [wallet, grant] = await Promise.all([connect(session, '/wallet'), connect(session, '/grant')]);
    send(res, 200, { sub: session.identity?.sub, name: session.identity?.name, wallet, remainingUsd: grant.remainingUsd });
  } catch (error) {
    send(res, 500, { error: error.message });
  }
}).listen(3000, () => console.log('打开 http://localhost:3000'));

pip install flask requests "PyJWT[crypto]"
export FLASK_SECRET_KEY="$(python3 -c 'import secrets; print(secrets.token_urlsafe(32))')"
flask --app app run --port 5000

import base64
import hashlib
import hmac
import os
import secrets
import time
import uuid
from urllib.parse import urlencode

import jwt
import requests
from flask import Flask, abort, redirect, request, session

ISSUER = "https://tokentosea.com/api/oauth"
SERVICES = "https://tokentosea.com/api/v1/connect"
CLIENT_ID = os.environ["TTS_CLIENT_ID"]
CLIENT_SECRET = os.environ.get("TTS_CLIENT_SECRET")  # 公共客户端不设置
REDIRECT_URI = "http://localhost:5000/callback"
SCOPES = "openid profile offline_access tools:read tools:invoke wallet:read"

app = Flask(__name__)
app.secret_key = os.environ["FLASK_SECRET_KEY"]
store = {}  # 演示用服务端内存存储；生产环境请用数据库并加密保存令牌
jwks = jwt.PyJWKClient(ISSUER + "/jwks")


def user_state() -> dict:
    sid = session.setdefault("sid", secrets.token_urlsafe(32))
    return store.setdefault(sid, {})


def with_client(data: dict) -> dict:
    data["client_id"] = CLIENT_ID
    if CLIENT_SECRET:
        data["client_secret"] = CLIENT_SECRET
    return data


@app.get("/login")
def login():
    state = user_state()
    verifier = secrets.token_urlsafe(48)
    state["login"] = {"state": secrets.token_urlsafe(32), "nonce": secrets.token_urlsafe(32), "verifier": verifier, "at": time.time()}
    challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode()
    query = {
        "response_type": "code",
        "client_id": CLIENT_ID,
        "redirect_uri": REDIRECT_URI,
        "scope": SCOPES,
        "state": state["login"]["state"],
        "nonce": state["login"]["nonce"],
        "code_challenge": challenge,
        "code_challenge_method": "S256",
    }
    return redirect(ISSUER + "/authorize?" + urlencode(query))


@app.get("/callback")
def callback():
    state = user_state()
    pending = state.pop("login", None)  # 每次登录的随机值只用一次
    if not pending or time.time() - pending["at"] > 600:
        abort(400, "登录已过期")
    if not hmac.compare_digest(request.args.get("state", ""), pending["state"]) or request.args.get("iss") != ISSUER:
        abort(400, "state 或 iss 不匹配")
    if request.args.get("error"):
        return "用户拒绝了授权", 403

    response = requests.post(ISSUER + "/token", data=with_client({
        "grant_type": "authorization_code",
        "code": request.args["code"],
        "redirect_uri": REDIRECT_URI,
        "code_verifier": pending["verifier"],
    }), timeout=30)
    response.raise_for_status()
    tokens = response.json()

    signing_key = jwks.get_signing_key_from_jwt(tokens["id_token"])
    claims = jwt.decode(tokens["id_token"], signing_key.key, algorithms=["RS256"], audience=CLIENT_ID, issuer=ISSUER)
    if not hmac.compare_digest(claims.get("nonce", ""), pending["nonce"]):
        abort(400, "nonce 不匹配")
    state.update(tokens=tokens, expires_at=time.time() + tokens["expires_in"], sub=claims["sub"], name=claims.get("name"))
    return redirect("/")


def access_token(state: dict) -> str:
    if time.time() < state["expires_at"] - 60:
        return state["tokens"]["access_token"]
    refresh = state["tokens"].get("refresh_token")
    if not refresh:
        abort(401, "未授予 offline_access，请重新登录")
    # 刷新令牌只能使用一次：同一用户的刷新必须串行，并原子替换
    response = requests.post(ISSUER + "/token", data=with_client({"grant_type": "refresh_token", "refresh_token": refresh}), timeout=30)
    response.raise_for_status()
    state["tokens"] = response.json()
    state["expires_at"] = time.time() + state["tokens"]["expires_in"]
    return state["tokens"]["access_token"]


def connect(state: dict, method: str, path: str, **kwargs) -> dict:
    response = requests.request(method, SERVICES + path, headers={"Authorization": "Bearer " + access_token(state)}, timeout=60, **kwargs)
    response.raise_for_status()
    return response.json()


@app.get("/")
def home():
    state = user_state()
    if "tokens" not in state:
        return '<a href="/login">用 TokenToSea 登录</a>'
    grant = connect(state, "GET", "/grant")
    return {"sub": state["sub"], "name": state.get("name"), "wallet": connect(state, "GET", "/wallet"), "remainingUsd": grant["remainingUsd"]}


@app.get("/tool")
def tool():
    state = user_state()
    if "tokens" not in state:
        return redirect("/login")
    action = next(a for a in connect(state, "GET", "/tools/public_hn/actions") if a["id"] == "hn.top")
    run = connect(state, "POST", "/tools/invoke", json={
        "actionId": action["id"],
        "input": {"limit": 3},
        "requestKey": str(uuid.uuid4()),
        "priceVersion": action["priceVersion"],
        "maxCostUsd": action["priceUsd"],
    })
    return {"status": run["status"], "chargedUsd": run["chargedUsd"], "output": run["output"], "errorCode": run["errorCode"]}