扣子 Coze：扣子可以把你自己的 HTTP 服务导入为插件，方式是提供一份 OpenAPI 文档

扣子可以把你自己的 HTTP 服务导入为插件，方式是提供一份 OpenAPI 文档。导入失败时，提示往往不够具体。 这篇写一个小服务（文本统计、长度单位换算）和它的 openapi.yaml，再写一个检查器，分两层：

• 第 1 层：通用 OpenAPI 校验；
• 第 2 层：逐条照搬开源 Coze Studio v0.5.1 源码 openapi.go 中 Validate() 的额外规则，比如只能有一个 server、必须有 summary、响应只能是 200 + JSON 对象、path 参数不能是数组。

先看清成品（box 上真实运行）

• openapi.yaml 通过检查；反例 bad.yaml 报出 6 条错误。
• Coze Studio 自带的 18 个官方插件文档全部通过检查（用来反向验证规则移植得对不对）。
• 服务的实际响应用 jsonschema 对照文档中的 schema 校验通过。pytest 6 项通过。

合规提醒：扣子中国站（coze.cn）和国际站（coze.com）是两套独立的账号体系，个人访问令牌（PAT）也不能互用。请使用你所在地区对应的站点，不要用代理绕过地区限制，不要共享账号或 Token。Token 只放在环境变量里，不要写进代码或仓库。发给智能体和工作流的数据会由字节跳动的服务处理，涉及个人信息或公司数据时，先确认合规要求。本文不列价格和额度，以官网为准。

运行说明：box 上没有扣子账号，没有把插件真正导入 coze.cn，也没有把服务部署到公网。检查规则来自开源版源码，扣子云端的实现没有公开，可能有出入，最终以导入时的提示为准。 自部署 Coze Studio 也因为 minio 镜像拉不到而没有成功（上游 issue #2775）。服务、检查器和测试都在 box 上真实运行过。

做完之后你会得到什么？

需要准备哪些环境？

Python 3.9 以上；pip install pyyaml jsonschema openapi-spec-validator pytest（box 上 openapi-spec-validator 的版本是 0.9.0）。如果要测试官方插件，还需要 git clone --branch v0.5.1 https://github.com/coze-dev/coze-studio /tmp/coze-studio。

Coze Studio v0.5.1 的插件文档规则（源码原文）

文件结构

59-coze-C-src/
├── plugin_server.py
├── openapi.yaml
├── coze_plugin_lint.py
├── examples/bad.yaml
└── test_plugin.py

第 1 步：服务

"""一个可以做成扣子“基于已有服务创建插件”的小服务（标准库，教学用）。
POST /v1/text/stats    {"text": "..."} -> 字符数 / 汉字数 / 行数
GET  /v1/convert/length?value=1&from=m&to=cm -> 长度单位换算
请求头 X-Api-Key 校验（在扣子里对应“服务 Token / API Key”授权，密钥放环境变量）。"""
import json, os, re, sys, urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

KEY = os.environ.get("PLUGIN_API_KEY", "dev-key")
UNITS = {"mm": 0.001, "cm": 0.01, "m": 1.0, "km": 1000.0, "in": 0.0254, "ft": 0.3048}

class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass
    def _json(self, code, obj):
        raw = json.dumps(obj, ensure_ascii=False).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(raw))); self.end_headers(); self.wfile.write(raw)
    def _auth(self):
        if self.headers.get("X-Api-Key") != KEY:
            self._json(401, {"code": 401, "msg": "invalid api key"}); return False
        return True
    def do_POST(self):
        if not self._auth():
            return
        if urllib.parse.urlsplit(self.path).path != "/v1/text/stats":
            return self._json(404, {"code": 404, "msg": "not found"})
        try:
            body = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))) or b"{}")
            text = body["text"]
            assert isinstance(text, str)
        except Exception:
            return self._json(400, {"code": 400, "msg": "text (string) is required"})
        if len(text) > 100_000:
            return self._json(413, {"code": 413, "msg": "text too long (max 100000)"})
        self._json(200, {"code": 0, "msg": "ok", "data": {
            "chars": len(text), "han": len(re.findall(r"[\u4e00-\u9fff]", text)),
            "lines": text.count("\n") + 1 if text else 0}})
    def do_GET(self):
        if not self._auth():
            return
        u = urllib.parse.urlsplit(self.path)
        if u.path != "/v1/convert/length":
            return self._json(404, {"code": 404, "msg": "not found"})
        q = dict(urllib.parse.parse_qsl(u.query))
        try:
            v, f, t = float(q["value"]), q["from"], q["to"]
            r = v * UNITS[f] / UNITS[t]
        except (KeyError, ValueError):
            return self._json(400, {"code": 400, "msg": f"need value/from/to; units: {','.join(UNITS)}"})
        self._json(200, {"code": 0, "msg": "ok", "data": {"result": round(r, 6), "unit": t}})

def make(port=0):
    return ThreadingHTTPServer(("127.0.0.1", port), H)

if __name__ == "__main__":
    s = make(int(sys.argv[1]) if len(sys.argv) > 1 else 8787)
    print(f"listening on http://127.0.0.1:{s.server_address[1]}"); s.serve_forever()

第 2 步：OpenAPI 文档

openapi: 3.0.1
info:
  title: 文本小工具
  description: 统计文本的字符数、汉字数、行数，以及常用长度单位换算。
  version: v1
servers:
  - url: https://plugin.example.com
paths:
  /v1/text/stats:
    post:
      operationId: text_stats
      summary: 统计一段文本的字符数、汉字数和行数
      requestBody:
        content:
          application/json:
            schema:
              type: object
              required: [text]
              properties:
                text:
                  type: string
                  description: 要统计的文本，最多 100000 字符
      responses:
        "200":
          description: 成功
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/StatsResp"
  /v1/convert/length:
    get:
      operationId: convert_length
      summary: 长度单位换算（mm/cm/m/km/in/ft）
      parameters:
        - {name: value, in: query, required: true, description: 数值, schema: {type: number}}
        - {name: from, in: query, required: true, description: 原单位, schema: {type: string, enum: [mm, cm, m, km, in, ft]}}
        - {name: to, in: query, required: true, description: 目标单位, schema: {type: string, enum: [mm, cm, m, km, in, ft]}}
      responses:
        "200":
          description: 成功
          content:
            application/json:
              schema:
                $ref: "#/components/schemas/ConvertResp"
components:
  schemas:
    StatsResp:
      type: object
      required: [code, msg, data]
      properties:
        code: {type: integer}
        msg: {type: string}
        data:
          type: object
          required: [chars, han, lines]
          properties:
            chars: {type: integer, description: 字符数}
            han: {type: integer, description: 汉字数}
            lines: {type: integer, description: 行数}
    ConvertResp:
      type: object
      required: [code, msg, data]
      properties:
        code: {type: integer}
        msg: {type: string}
        data:
          type: object
          required: [result, unit]
          properties:
            result: {type: number}
            unit: {type: string}

反例：

openapi: 3.0.1
info: {title: 坏例子, version: v1}
servers:
  - url: https://a.example.com
  - url: https://b.example.com
paths:
  /items/{ids}:
    post:
      operationId: add_items
      parameters:
        - {name: ids, in: path, required: true, schema: {type: array, items: {type: string}}}
      requestBody:
        content:
          text/plain:
            schema: {type: string}
      responses:
        "201":
          description: created
          content:
            application/json:
              schema: {type: object}

第 3 步：检查器

"""检查 OpenAPI 文档能否被扣子插件导入。
第 1 层：openapi-spec-validator（通用 OpenAPI 3.0 校验）
第 2 层：照搬 coze-studio v0.5.1 backend/crossdomain/plugin/model/openapi.go 中 Validate() 的额外规则。
注意：这是开源版的规则，扣子云端（coze.cn/coze.com）的实现未公开，可能有出入。"""
import sys, urllib.parse
import yaml
from openapi_spec_validator import validate
from openapi_spec_validator.readers import read_from_filename

REQ_TYPES = ["application/json", "application/problem+json", "application/x-www-form-urlencoded",
             "application/x-yaml", "application/yaml"]
METHODS = {"get", "put", "post", "delete", "options", "head", "patch", "trace"}

def deref(doc, node):
    while isinstance(node, dict) and "$ref" in node:
        cur = doc
        for part in node["$ref"].lstrip("#/").split("/"):
            cur = cur[part]
        node = cur
    return node

def check_op(doc, where, op):
    errs = []
    E = lambda m: errs.append(f"{where}: {m}")
    if not op.get("operationId"):
        E("operationID is required")
    if not op.get("summary"):
        E("summary is required")
    rb = op.get("requestBody")
    if rb is not None:
        content = (deref(doc, rb) or {}).get("content") or {}
        if not content:
            E("request body is required")
        elif len(content) != 1:
            E("request body only supports one media type")
        else:
            mt = next((content[c] for c in REQ_TYPES if c in content), None)
            if mt is None:
                E(f"invalid media type, request body only the following types: [{', '.join(REQ_TYPES)}]")
            elif not mt.get("schema"):
                E("request body schema is required")
            elif deref(doc, mt["schema"]).get("type") != "object":
                E("request body only supports 'object' type")
    for p in op.get("parameters") or []:
        p = deref(doc, p)
        sc = deref(doc, p.get("schema")) if p.get("schema") else None
        if not sc:
            E("parameter schema is required"); continue
        t = sc.get("type", "")
        if p.get("in") == "body":
            E(f"the location of parameter '{p.get('name')}' cannot be 'body'")
        if not t:
            E(f"the type of  parameter '{p.get('name')}' is required")
        if t == "object":
            E(f"the type of parameter '{p.get('name')}' cannot be 'object'")
        if p.get("in") == "path" and t == "array":
            E(f"the type of parameter '{p.get('name')}' cannot be 'array'")
    resps = op.get("responses") or {}
    if resps:
        keys = set(map(str, resps))
        if not (keys == {"200"} or keys == {"200", "default"}):
            E("response only supports '200' status")
        else:
            r = deref(doc, resps.get("200", resps.get(200)))
            content = r.get("content") or {}
            if len(content) != 1 or "application/json" not in content:
                E("response only supports 'application/json' media type")
            elif not content["application/json"].get("schema"):
                E("the media type schema of response is required")
            elif deref(doc, content["application/json"]["schema"]).get("type") != "object":
                E("response body only supports 'object' type")
    return errs

def lint(path):
    try:
        spec, _ = read_from_filename(path)
        validate(spec)
    except Exception as e:
        msg = (str(e).strip().splitlines() or [type(e).__name__])[0]   # 有的异常 str() 为空
        return [f"OpenAPI 校验失败：{msg}"]
    doc = yaml.safe_load(open(path, encoding="utf-8"))
    errs = []
    info = doc.get("info") or {}
    if not info.get("title"):
        errs.append("the title of info is required")
    if not info.get("description"):
        errs.append("the description of info is required")
    servers = doc.get("servers") or []
    if len(servers) != 1:
        errs.append("server is required and only one server is allowed")
    else:
        u = servers[0].get("url", "")
        if not urllib.parse.urlsplit(u).hostname:
            errs.append(f"invalid server url '{u}'")
        if len(u) > 512:
            errs.append(f"server url '{u}' is too long")
    for p, item in (doc.get("paths") or {}).items():
        for m, op in item.items():
            if m in METHODS:
                errs += check_op(doc, f"{m.upper()} {p}", op)
    return errs

if __name__ == "__main__":
    bad = 0
    for f in sys.argv[1:]:
        errs = lint(f)
        bad += bool(errs)
        print(f"{'✗' if errs else '✓'} {f}")
        for e in errs:
            print(f"    - {e}")
    sys.exit(1 if bad else 0)

第 4 步：测试

import glob, json, threading, urllib.error, urllib.request
import jsonschema, pytest, yaml
import coze_plugin_lint as L, plugin_server

SPEC = yaml.safe_load(open("openapi.yaml", encoding="utf-8"))

@pytest.fixture(scope="module")
def base():
    s = plugin_server.make(); threading.Thread(target=s.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{s.server_address[1]}"; s.shutdown()

def call(base, path, body=None, key="dev-key"):
    req = urllib.request.Request(base + path, data=json.dumps(body).encode() if body is not None else None,
                                 headers={"X-Api-Key": key, "Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req) as r:
            return r.status, json.load(r)
    except urllib.error.HTTPError as e:
        return e.code, json.load(e)

def schema_for(path, method):
    s = SPEC["paths"][path][method]["responses"]["200"]["content"]["application/json"]["schema"]
    return {**s, "components": SPEC["components"]}

def test_spec_passes_lint():
    assert L.lint("openapi.yaml") == []

def test_bad_example_errors():
    errs = "\n".join(L.lint("examples/bad.yaml"))
    for m in ["description of info", "only one server", "summary is required", "invalid media type",
              "cannot be 'array'", "only supports '200'"]:
        assert m in errs

def test_official_plugins_pass():
    files = [f for f in glob.glob("/tmp/coze-studio/backend/conf/plugin/pluginproduct/*.yaml")
             if not f.endswith("plugin_meta.yaml")]
    if not files:
        pytest.skip("未克隆 coze-studio")
    assert all(L.lint(f) == [] for f in files), [f for f in files if L.lint(f)]

def test_stats_matches_schema(base):
    code, r = call(base, "/v1/text/stats", {"text": "你好 world\n第二行"})
    assert code == 200 and r["data"] == {"chars": 12, "han": 5, "lines": 2}
    jsonschema.validate(r, schema_for("/v1/text/stats", "post"))

def test_convert_matches_schema(base):
    code, r = call(base, "/v1/convert/length?value=3&from=ft&to=m")
    assert code == 200 and r["data"]["result"] == 0.9144
    jsonschema.validate(r, schema_for("/v1/convert/length", "get"))

def test_errors(base):
    assert call(base, "/v1/text/stats", {"text": 1})[0] == 400
    assert call(base, "/v1/convert/length?value=1&from=m&to=li")[0] == 400
    assert call(base, "/v1/text/stats", {"text": "x"}, key="nope")[0] == 401

第 5 步：运行

$ python coze_plugin_lint.py openapi.yaml examples/bad.yaml
✓ openapi.yaml
✗ examples/bad.yaml
    - the description of info is required
    - server is required and only one server is allowed
    - POST /items/{ids}: summary is required
    - POST /items/{ids}: invalid media type, request body only the following types: [application/json, application/problem+json, application/x-www-form-urlencoded, application/x-yaml, application/yaml]
    - POST /items/{ids}: the type of parameter 'ids' cannot be 'array'
    - POST /items/{ids}: response only supports '200' status
$ python -m pytest -q
6 passed

要在扣子中导入，需要先把服务部署到一个公网 HTTPS 地址，并把 servers[0].url 改成这个地址，然后在扣子里选择基于已有服务创建插件，导入 openapi.yaml，授权方式选 API Key（Header X-Api-Key）。这一步我没有执行。

调试示例

1. 检查器自己崩溃了。 我第一次用检查器跑官方插件目录时，得到的不是检查结果，而是 IndexError: list index out of range。原因是 openapi-spec-validator 抛出的 ValidatorDetectError 的 str() 是空字符串，而我写的是 str(e).splitlines()[0]。改成在消息为空时使用异常类名之后，报错正常显示为 OpenAPI 校验失败：ValidatorDetectError。

2. 有一个“官方插件”没有通过检查。 修好检查器之后，只有 plugin_meta.yaml 不通过，原因是它没有 openapi: 版本字段。看过文件内容后发现，它是插件的索引清单，本身不是 OpenAPI 文档。另外 5 个飞书插件的 openapi: 字段不在文件开头，但能正常识别。所以测试中排除了 plugin_meta.yaml，其余 18 个全部通过，这也说明移植的规则没有误伤官方文档。

常见问题

模板规则

• 每个操作都写 operationId 和中文 summary（模型靠它来选工具）。
• 响应统一使用 200 + JSON 对象。
• 用测试保证服务的实际响应符合文档。
• 密钥只通过扣子的授权配置传入。

项目资源

插件文档写对了，导入时就不用反复试错，模型也更容易选对工具。