# SDK 源码 (/docs/plugins/helpers)

拷到插件 `lib/`。用法见 [模型](/docs/plugins/models)。不要改文件名里的导出。对照自己 `lib/` 里的 `SDK_VERSION`（Python 是 `__version__`）和文档母版，不一致就整文件再拷。工作台不提示过期。

## dex-buddy-sdk.js [#dex-buddy-sdkjs]

```js title="lib/dex-buddy-sdk.js"
// @ts-check
'use strict';

/**
 * Dex Buddy 插件 SDK 1.0.0。拷进插件 lib/ 后：
 * 工作台走宿主校验与 ctx.models；独立跑 requireEntitlement 直接通过，bindModels 用 DEX_BUDDY_API_KEY（开发者付钱）。
 * 对照文档母版的 SDK_VERSION，不一致就整文件再拷。不要把 Key 写进设置或 zip。
 */

const SDK_VERSION = '1.0.0';

const fs = require('node:fs');
const http = require('node:http');
const path = require('node:path');

const DEFAULT_BASE = 'https://www.dexbuddy.app';
const MISSING_KEY =
  '独立运行需要环境变量 DEX_BUDDY_API_KEY（账号里的用户 API Key，dbh_sk_ 开头）。写进本机 .env，不要放进插件设置，也不要打进 zip。装进 Dex Buddy 后会改走宿主，扣使用者的积分。';

class DexModelsError extends Error {
  /**
   * @param {string} code
   * @param {string} message
   * @param {number} [status]
   */
  constructor(code, message, status) {
    super(message);
    this.name = 'DexModelsError';
    this.code = code;
    this.status = typeof status === 'number' ? status : 0;
  }
}

/** @type {Promise<string> | null} */
let proxyUrlPromise = null;

function missingKey() {
  return new DexModelsError('UNAUTHORIZED', MISSING_KEY, 401);
}

/**
 * @typedef {{
 *   list: () => Promise<unknown>;
 *   billing: { quote: (body?: unknown, options?: { idempotencyKey?: string }) => Promise<unknown> };
 *   chat: { completions: { create: (body?: unknown, options?: { idempotencyKey?: string }) => Promise<unknown> } };
 *   images: { generate: (input?: unknown, options?: { idempotencyKey?: string }) => Promise<unknown> };
 * }} HostModels
 */

/**
 * @param {unknown} ctx
 * @returns {HostModels | null}
 */
function hostModels(ctx) {
  const models = ctx && typeof ctx === 'object' ? /** @type {{ models?: HostModels }} */ (ctx).models : null;
  const create = models && models.chat && models.chat.completions && models.chat.completions.create;
  return typeof create === 'function' && models ? models : null;
}

/**
 * @param {unknown} raw
 * @returns {string}
 */
function trimString(raw) {
  return typeof raw === 'string' ? raw.trim() : '';
}

/**
 * @param {Record<string, unknown> | null | undefined} [options]
 */
function resolvePluginId(options) {
  const fromOptions = trimString(options && /** @type {{ pluginId?: string }} */ (options).pluginId);
  if (fromOptions) return fromOptions;
  const explicit = trimString(options && /** @type {{ manifestPath?: string }} */ (options).manifestPath);
  const candidates = [
    explicit,
    path.join(__dirname, '..', 'plugin.manifest.json'),
    path.join(__dirname, 'plugin.manifest.json'),
  ];
  for (const file of candidates) {
    if (!file || !fs.existsSync(file)) continue;
    try {
      const parsed = JSON.parse(fs.readFileSync(file, 'utf8'));
      const id = parsed && typeof parsed.id === 'string' ? parsed.id.trim() : '';
      if (id) return id;
    } catch {
      // 清单读失败时不盖 id，网关头仍可选。
    }
  }
  return '';
}

/**
 * @param {unknown} body
 * @param {number} status
 */
function errorFromBody(body, status) {
  const record = body && typeof body === 'object' && !Array.isArray(body) ? /** @type {Record<string, unknown>} */ (body) : null;
  const nested = record && record.error && typeof record.error === 'object' ? /** @type {Record<string, unknown>} */ (record.error) : null;
  if (nested && typeof nested.message === 'string' && nested.message.trim() !== '') {
    const code = typeof nested.code === 'string' && nested.code.trim() !== '' ? nested.code : 'BAD_RESPONSE';
    return new DexModelsError(code, nested.message, status);
  }
  const code = record && typeof record.code === 'string' && record.code.trim() !== '' ? record.code : 'BAD_RESPONSE';
  const message =
    record && typeof record.error === 'string' && record.error.trim() !== ''
      ? record.error
      : `网关请求失败（${status}）`;
  return new DexModelsError(code, message, status);
}

/**
 * @param {string} filePath
 */
function mimeFromPath(filePath) {
  const ext = path.extname(filePath).toLowerCase();
  if (ext === '.png') return 'image/png';
  if (ext === '.jpg' || ext === '.jpeg') return 'image/jpeg';
  if (ext === '.webp') return 'image/webp';
  if (ext === '.gif') return 'image/gif';
  return 'application/octet-stream';
}

/**
 * @param {unknown} ctx
 * @param {Record<string, unknown> | null} [options]
 * @returns {HostModels}
 */
function bindModels(ctx, options) {
  const host = hostModels(ctx);
  if (host) return host;
  return createStandaloneModels(options || undefined);
}

/**
 * 工作台转交宿主买断确认；独立跑直接通过。正式作者不要传第二参。
 * @param {unknown} ctx
 * @param {unknown} [options]
 */
async function requireEntitlement(ctx, options) {
  const fn =
    ctx && typeof ctx === 'object'
      ? /** @type {{ requireEntitlement?: unknown }} */ (ctx).requireEntitlement
      : null;
  if (typeof fn === 'function') {
    if (arguments.length > 1) return fn(options);
    return fn();
  }
}

/**
 * @param {Record<string, unknown> | null | undefined} options
 * @returns {HostModels}
 */
function createStandaloneModels(options) {
  const env = /** @type {NodeJS.ProcessEnv} */ ((options && options.env) || process.env);
  const fetchImpl =
    /** @type {(input: string, init?: RequestInit) => Promise<Response>} */ (
      (options && options.fetch) || globalThis.fetch.bind(globalThis)
    );
  const baseUrl = String((options && options.baseUrl) || env.DEX_GATEWAY_BASE_URL || DEFAULT_BASE).replace(
    /\/+$/,
    '',
  );
  const pluginId = resolvePluginId(options);

  /**
   * @param {string} requestPath
   * @param {{ method: 'GET' | 'POST'; body?: unknown; headers?: Record<string, string> }} init
   */
  async function request(requestPath, init) {
    const key = trimString(env.DEX_BUDDY_API_KEY);
    if (!key) throw missingKey();
    /** @type {Record<string, string>} */
    const headers = {
      Accept: 'application/json',
      Authorization: `Bearer ${key}`,
      ...(pluginId ? { 'X-Plugin-Id': pluginId } : {}),
      ...(init.headers || {}),
    };
    if (init.body !== undefined) headers['Content-Type'] = 'application/json';
    const extra = init.headers && init.headers['Idempotency-Key'];
    if (extra) headers['Idempotency-Key'] = extra;
    let response;
    try {
      response = await fetchImpl(`${baseUrl}${requestPath.startsWith('/') ? requestPath : `/${requestPath}`}`, {
        method: init.method,
        headers,
        body: init.body === undefined ? undefined : JSON.stringify(init.body),
      });
    } catch (err) {
      throw new DexModelsError('NETWORK_ERROR', err instanceof Error ? err.message : '无法连接网关', 0);
    }
    let parsed = null;
    const text = await response.text();
    if (text.trim() !== '') {
      try {
        parsed = JSON.parse(text);
      } catch {
        throw new DexModelsError('BAD_RESPONSE', '网关返回了无法解析的响应', response.status);
      }
    }
    if (!response.ok) throw errorFromBody(parsed, response.status);
    return parsed;
  }

  /**
   * @param {{ idempotencyKey?: string }} [callOptions]
   * @returns {Record<string, string>}
   */
  function extraHeaders(callOptions) {
    const key = callOptions && typeof callOptions.idempotencyKey === 'string' ? callOptions.idempotencyKey.trim() : '';
    /** @type {Record<string, string>} */
    const headers = {};
    if (key) headers['Idempotency-Key'] = key;
    return headers;
  }

  /**
   * @param {string} filePath
   * @param {{ idempotencyKey?: string }} [callOptions]
   */
  async function uploadReference(filePath, callOptions) {
    const bytes = fs.readFileSync(filePath);
    if (!bytes || bytes.length === 0) {
      throw new DexModelsError('INVALID_BODY', `参考图是空的：${filePath}`, 400);
    }
    const contentType = mimeFromPath(filePath);
    const issued = await request('/v1/files', {
      method: 'POST',
      body: { contentType, contentLength: bytes.length },
      headers: extraHeaders(callOptions),
    });
    if (!issued || typeof issued.uploadUrl !== 'string' || typeof issued.token !== 'string' || typeof issued.key !== 'string') {
      throw new DexModelsError('BAD_RESPONSE', '网关 files 凭证无效', 502);
    }
    const form = new FormData();
    form.set('token', issued.token);
    form.set('key', issued.key);
    form.set('file', new Blob([bytes], { type: contentType }), path.basename(filePath) || 'image');
    let uploaded;
    try {
      uploaded = await fetchImpl(issued.uploadUrl, { method: 'POST', body: form });
    } catch (err) {
      throw new DexModelsError('NETWORK_ERROR', err instanceof Error ? err.message : '无法上传参考图', 0);
    }
    if (!uploaded.ok) {
      throw new DexModelsError('UPSTREAM_ERROR', `参考图上传失败（${uploaded.status}）`, uploaded.status);
    }
    if (typeof issued.id !== 'string' || issued.id.trim() === '') {
      throw new DexModelsError('BAD_RESPONSE', '网关 files 凭证无效', 502);
    }
    return issued.id;
  }

  return {
    async list() {
      return request('/v1/models', { method: 'GET' });
    },
    billing: {
      /**
       * @param {unknown} body
       * @param {{ idempotencyKey?: string }} [callOptions]
       */
      quote(body, callOptions) {
        return request('/v1/billing/quote', {
          method: 'POST',
          body,
          headers: extraHeaders(callOptions),
        });
      },
    },
    chat: {
      completions: {
        /**
         * @param {unknown} body
         * @param {{ idempotencyKey?: string }} [callOptions]
         */
        create(body, callOptions) {
          return request('/v1/chat/completions', {
            method: 'POST',
            body,
            headers: extraHeaders(callOptions),
          });
        },
      },
    },
    images: {
      /**
       * @param {unknown} input
       * @param {{ idempotencyKey?: string }} [callOptions]
       */
      async generate(input, callOptions) {
        const payload = input && typeof input === 'object' ? /** @type {Record<string, unknown>} */ (input) : {};
        const model = trimString(payload.model);
        const outputPath = trimString(payload.outputPath);
        const record = payload.input && typeof payload.input === 'object'
          ? /** @type {Record<string, unknown>} */ (payload.input)
          : {};
        const prompt = trimString(record.prompt);
        if (!model) throw new DexModelsError('INVALID_BODY', '缺少 model', 400);
        if (!prompt) throw new DexModelsError('INVALID_BODY', '缺少 prompt', 400);
        if (!outputPath || !path.isAbsolute(outputPath)) {
          throw new DexModelsError('INVALID_BODY', 'outputPath 必须是绝对路径', 400);
        }
        const rawImages = record.input_images;
        /** @type {string[]} */
        const localImages = [];
        if (Array.isArray(rawImages)) {
          for (const item of rawImages) {
            if (typeof item === 'string' && item.trim() !== '') localImages.push(item.trim());
          }
        }
        /** @type {string[]} */
        const fileIds = [];
        for (let i = 0; i < localImages.length; i += 1) {
          const fileKey = callOptions && callOptions.idempotencyKey
            ? { idempotencyKey: `${callOptions.idempotencyKey}:file:${i}` }
            : callOptions;
          fileIds.push(await uploadReference(localImages[i], fileKey));
        }
        /** @type {{ model: string; prompt: string; aspect_ratio?: string; fileIds?: string[] }} */
        const generateBody = { model, prompt };
        if (trimString(record.aspect_ratio)) generateBody.aspect_ratio = trimString(record.aspect_ratio);
        if (fileIds.length > 0) generateBody.fileIds = fileIds;
        const generated = await request('/v1/images/generate', {
          method: 'POST',
          body: generateBody,
          headers: extraHeaders(callOptions),
        });
        if (!generated || typeof generated.downloadUrl !== 'string' || generated.downloadUrl.trim() === '') {
          throw new DexModelsError('BAD_RESPONSE', '网关生图结果无效', 502);
        }
        let downloaded;
        try {
          downloaded = await fetchImpl(generated.downloadUrl);
        } catch (err) {
          throw new DexModelsError('NETWORK_ERROR', err instanceof Error ? err.message : '无法下载生成图', 0);
        }
        if (!downloaded.ok) {
          throw new DexModelsError('OUTPUT_EXPIRED', `生成图下载失败（${downloaded.status}）`, downloaded.status);
        }
        const bytes = Buffer.from(await downloaded.arrayBuffer());
        if (bytes.length === 0) throw new DexModelsError('BAD_RESPONSE', '生成图是空的', 502);
        fs.mkdirSync(path.dirname(outputPath), { recursive: true });
        fs.writeFileSync(outputPath, bytes);
        return {
          status: 'succeeded',
          output: outputPath,
          model: typeof generated.model === 'string' ? generated.model : model,
        };
      },
    },
  };
}

/**
 * @param {unknown} ctx
 * @param {http.IncomingMessage} req
 * @param {http.ServerResponse} res
 * @param {unknown} body
 */
async function handleProxy(ctx, req, res, body) {
  const models = hostModels(ctx);
  if (!models) {
    res.statusCode = 401;
    res.end(JSON.stringify({ code: 'UNAUTHORIZED', error: '宿主模型不可用' }));
    return;
  }
  const url = new URL(req.url || '/', 'http://127.0.0.1');
  let result;
  if (req.method === 'GET' && url.pathname === '/v1/models') {
    result = await models.list();
  } else if (req.method === 'POST' && url.pathname === '/v1/billing/quote') {
    result = await models.billing.quote(body);
  } else if (req.method === 'POST' && url.pathname === '/v1/chat/completions') {
    result = await models.chat.completions.create(body);
  } else if (req.method === 'POST' && url.pathname === '/images/generate') {
    result = await models.images.generate(body);
  } else {
    res.statusCode = 404;
    res.end(JSON.stringify({ code: 'BAD_RESPONSE', error: '未知路径' }));
    return;
  }
  res.statusCode = 200;
  res.setHeader('Content-Type', 'application/json');
  res.end(JSON.stringify(result));
}

/**
 * 工作台里给 Python 子进程用：指向本机代理，并删掉开发者 Key。
 * @param {unknown} ctx
 * @param {NodeJS.ProcessEnv} [baseEnv]
 */
function childEnv(ctx, baseEnv) {
  const env = { ...(baseEnv || process.env) };
  delete env.DEX_BUDDY_API_KEY;
  delete env.HTTP_PROXY;
  delete env.HTTPS_PROXY;
  delete env.ALL_PROXY;
  delete env.http_proxy;
  delete env.https_proxy;
  delete env.all_proxy;
  env.NO_PROXY = '127.0.0.1,localhost';
  env.no_proxy = '127.0.0.1,localhost';
  const host = hostModels(ctx);
  if (!host) return Promise.resolve(env);
  if (!proxyUrlPromise) {
    proxyUrlPromise = new Promise((resolve, reject) => {
      const server = http.createServer((req, res) => {
        /** @type {Buffer[]} */
        const chunks = [];
        req.on('data', (chunk) => {
          chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk)));
        });
        req.on('end', () => {
          let parsed = null;
          const raw = Buffer.concat(chunks).toString('utf8');
          if (raw.trim() !== '') {
            try {
              parsed = JSON.parse(raw);
            } catch {
              res.statusCode = 400;
              res.end(JSON.stringify({ code: 'INVALID_BODY', error: 'JSON 无效' }));
              return;
            }
          }
          handleProxy(ctx, req, res, parsed).catch((err) => {
            const code = err && typeof err.code === 'string' ? err.code : 'SERVER_ERROR';
            const status = err && typeof err.status === 'number' && err.status >= 400 ? err.status : 500;
            res.statusCode = status;
            res.setHeader('Content-Type', 'application/json');
            res.end(JSON.stringify({ code, error: err instanceof Error ? err.message : '代理失败' }));
          });
        });
      });
      server.listen(0, '127.0.0.1', () => {
        const address = server.address();
        if (!address || typeof address === 'string') {
          reject(new Error('无法监听本机代理'));
          return;
        }
        server.unref();
        resolve(`http://127.0.0.1:${address.port}`);
      });
      server.on('error', reject);
    });
  }
  return proxyUrlPromise.then((url) => {
    env.DEX_BUDDY_MODELS_URL = url;
    return env;
  });
}

module.exports = {
  DEFAULT_BASE,
  MISSING_KEY,
  DexModelsError,
  bindModels,
  requireEntitlement,
  childEnv,
  SDK_VERSION,
};
```

## dex_buddy_sdk.py [#dex_buddy_sdkpy]

```python title="lib/dex_buddy_sdk.py"
"""Dex Buddy 插件 SDK 1.0.0。拷进插件 lib/ 后：
独立跑用 DEX_BUDDY_API_KEY 打 Dex Buddy 网关（开发者付钱）；
工作台里由 Node 注入 DEX_BUDDY_MODELS_URL，走宿主（谁用谁付钱）。
require_entitlement 在 Python 里始终直接返回；工作台真校验写在 index.js。
对照文档母版的 __version__，不一致就整文件再拷。不要把 Key 打进 zip。
"""

from __future__ import annotations

import json
import mimetypes
import os
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

__version__ = "1.0.0"
DEFAULT_BASE = "https://www.dexbuddy.app"
MISSING_KEY = (
    "独立运行需要环境变量 DEX_BUDDY_API_KEY（账号里的用户 API Key，dbh_sk_ 开头）。"
    "写进本机 .env，不要放进插件设置，也不要打进 zip。装进 Dex Buddy 后会改走宿主，扣使用者的积分。"
)


class DexModelsError(RuntimeError):
    def __init__(self, code: str, message: str, status: int = 0) -> None:
        super().__init__(message)
        self.code = code
        self.status = status


def _env_map(env: dict[str, str] | None) -> dict[str, str]:
    return dict(os.environ if env is None else env)


def _trim(value: Any) -> str:
    return value.strip() if isinstance(value, str) else ""


class _Completions:
    def __init__(self, client: "Models") -> None:
        self._client = client

    def create(self, body: Any, options: dict[str, Any] | None = None) -> Any:
        return self._client._post("/v1/chat/completions", body, options)


class _Chat:
    def __init__(self, client: "Models") -> None:
        self.completions = _Completions(client)


class _Billing:
    def __init__(self, client: "Models") -> None:
        self._client = client

    def quote(self, body: Any, options: dict[str, Any] | None = None) -> Any:
        return self._client._post("/v1/billing/quote", body, options)


class _Images:
    def __init__(self, client: "Models") -> None:
        self._client = client

    def generate(self, payload: dict[str, Any], options: dict[str, Any] | None = None) -> Any:
        return self._client._generate_image(payload, options)


class Models:
    def __init__(self, env: dict[str, str] | None = None) -> None:
        self._env = _env_map(env)
        self.chat = _Chat(self)
        self.billing = _Billing(self)
        self.images = _Images(self)

    def list(self) -> Any:
        return self._request("GET", "/v1/models")

    def _models_url(self) -> str:
        return _trim(self._env.get("DEX_BUDDY_MODELS_URL")).rstrip("/")

    def _gateway(self) -> str:
        return _trim(self._env.get("DEX_GATEWAY_BASE_URL")) or DEFAULT_BASE

    def _headers(self, options: dict[str, Any] | None, json_body: bool) -> dict[str, str]:
        headers = {"Accept": "application/json"}
        if json_body:
            headers["Content-Type"] = "application/json"
        models_url = self._models_url()
        if not models_url:
            key = _trim(self._env.get("DEX_BUDDY_API_KEY"))
            if not key:
                raise DexModelsError("UNAUTHORIZED", MISSING_KEY, 401)
            headers["Authorization"] = f"Bearer {key}"
        if options and isinstance(options.get("idempotencyKey"), str):
            token = options["idempotencyKey"].strip()
            if token:
                headers["Idempotency-Key"] = token
        return headers

    def _request(self, method: str, path: str, body: Any = None, options: dict[str, Any] | None = None) -> Any:
        base = self._models_url() or self._gateway().rstrip("/")
        url = f"{base}{path if path.startswith('/') else '/' + path}"
        data = None if body is None else json.dumps(body).encode("utf-8")
        req = urllib.request.Request(url, data=data, method=method, headers=self._headers(options, body is not None))
        try:
            with _urlopen(req) as response:
                raw = response.read().decode("utf-8")
                return json.loads(raw) if raw.strip() else None
        except urllib.error.HTTPError as err:
            raw = err.read().decode("utf-8") if err.fp else ""
            parsed: Any = None
            if raw.strip():
                try:
                    parsed = json.loads(raw)
                except json.JSONDecodeError:
                    parsed = None
            raise _error_from_body(parsed, err.code) from err
        except urllib.error.URLError as err:
            raise DexModelsError("NETWORK_ERROR", str(err.reason or err), 0) from err

    def _post(self, path: str, body: Any, options: dict[str, Any] | None) -> Any:
        return self._request("POST", path, body, options)

    def _generate_image(self, payload: dict[str, Any], options: dict[str, Any] | None) -> Any:
        if self._models_url():
            return self._post("/images/generate", payload, options)
        model = _trim(payload.get("model"))
        output_path = _trim(payload.get("outputPath"))
        record = payload.get("input") if isinstance(payload.get("input"), dict) else {}
        prompt = _trim(record.get("prompt"))
        if not model:
            raise DexModelsError("INVALID_BODY", "缺少 model", 400)
        if not prompt:
            raise DexModelsError("INVALID_BODY", "缺少 prompt", 400)
        if not output_path or not os.path.isabs(output_path):
            raise DexModelsError("INVALID_BODY", "outputPath 必须是绝对路径", 400)
        local_images = [
            item.strip()
            for item in (record.get("input_images") or [])
            if isinstance(item, str) and item.strip()
        ]
        file_ids = []
        for index, file_path in enumerate(local_images):
            file_options = options
            if options and isinstance(options.get("idempotencyKey"), str) and options["idempotencyKey"].strip():
                file_options = {"idempotencyKey": f"{options['idempotencyKey'].strip()}:file:{index}"}
            file_ids.append(self._upload_reference(file_path, file_options))
        body: dict[str, Any] = {"model": model, "prompt": prompt}
        aspect = _trim(record.get("aspect_ratio"))
        if aspect:
            body["aspect_ratio"] = aspect
        if file_ids:
            body["fileIds"] = file_ids
        generated = self._post("/v1/images/generate", body, options)
        download_url = generated.get("downloadUrl") if isinstance(generated, dict) else None
        if not isinstance(download_url, str) or not download_url.strip():
            raise DexModelsError("BAD_RESPONSE", "网关生图结果无效", 502)
        try:
            with _urlopen(download_url) as response:
                data = response.read()
        except urllib.error.URLError as err:
            raise DexModelsError("NETWORK_ERROR", str(err.reason or err), 0) from err
        if not data:
            raise DexModelsError("BAD_RESPONSE", "生成图是空的", 502)
        out = Path(output_path)
        out.parent.mkdir(parents=True, exist_ok=True)
        out.write_bytes(data)
        model_out = generated.get("model") if isinstance(generated, dict) else model
        return {"status": "succeeded", "output": output_path, "model": model_out if isinstance(model_out, str) else model}

    def _upload_reference(self, file_path: str, options: dict[str, Any] | None) -> str:
        data = Path(file_path).read_bytes()
        if not data:
            raise DexModelsError("INVALID_BODY", f"参考图是空的：{file_path}", 400)
        content_type = mimetypes.guess_type(file_path)[0] or "application/octet-stream"
        issued = self._post(
            "/v1/files",
            {"contentType": content_type, "contentLength": len(data)},
            options,
        )
        if not isinstance(issued, dict):
            raise DexModelsError("BAD_RESPONSE", "网关 files 凭证无效", 502)
        upload_url = issued.get("uploadUrl")
        token = issued.get("token")
        key = issued.get("key")
        file_id = issued.get("id")
        if not all(isinstance(item, str) and item.strip() for item in (upload_url, token, key, file_id)):
            raise DexModelsError("BAD_RESPONSE", "网关 files 凭证无效", 502)
        boundary = "----DexBuddyForm"
        filename = os.path.basename(file_path) or "image"
        chunks = [
            f"--{boundary}\r\nContent-Disposition: form-data; name=\"token\"\r\n\r\n{token}\r\n".encode("utf-8"),
            f"--{boundary}\r\nContent-Disposition: form-data; name=\"key\"\r\n\r\n{key}\r\n".encode("utf-8"),
            (
                f"--{boundary}\r\nContent-Disposition: form-data; name=\"file\"; filename=\"{filename}\"\r\n"
                f"Content-Type: {content_type}\r\n\r\n"
            ).encode("utf-8")
            + data
            + b"\r\n",
            f"--{boundary}--\r\n".encode("utf-8"),
        ]
        req = urllib.request.Request(
            str(upload_url),
            data=b"".join(chunks),
            method="POST",
            headers={"Content-Type": f"multipart/form-data; boundary={boundary}"},
        )
        try:
            with _urlopen(req) as response:
                response.read()
        except urllib.error.HTTPError as err:
            raise DexModelsError("UPSTREAM_ERROR", f"参考图上传失败（{err.code}）", err.code) from err
        except urllib.error.URLError as err:
            raise DexModelsError("NETWORK_ERROR", str(err.reason or err), 0) from err
        return str(file_id)


def _error_from_body(body: Any, status: int) -> DexModelsError:
    record = body if isinstance(body, dict) else None
    nested = record.get("error") if record else None
    if isinstance(nested, dict) and isinstance(nested.get("message"), str) and nested["message"].strip():
        code = nested["code"] if isinstance(nested.get("code"), str) and nested["code"].strip() else "BAD_RESPONSE"
        return DexModelsError(code, nested["message"], status)
    if record and isinstance(record.get("error"), str) and record["error"].strip():
        code = record["code"] if isinstance(record.get("code"), str) and record["code"].strip() else "BAD_RESPONSE"
        return DexModelsError(code, record["error"], status)
    return DexModelsError("BAD_RESPONSE", f"网关请求失败（{status}）", status)


def _urlopen(req: urllib.request.Request | str, timeout: int = 120):
    url = req.full_url if isinstance(req, urllib.request.Request) else req
    handlers = []
    if "127.0.0.1" in url or "localhost" in url:
        # 本机代理不要走系统 HTTP_PROXY，否则工作台里的 Python 会卡住。
        handlers.append(urllib.request.ProxyHandler({}))
    opener = urllib.request.build_opener(*handlers)
    return opener.open(req, timeout=timeout)


def bind_models(env: dict[str, str] | None = None) -> Models:
    return Models(env)


def require_entitlement() -> None:
    return None
```