接入文档

把你的产品接入 AI Relay:选档位、写清单、实现 POST /relay、过接入验收。

L1:一个端点(同步)

适合 3 秒内能给结果的动作:查询、预约、下单。你要实现一个 HTTPS 端点 POST /relay 和一份清单,半天能做完。

下面用一家示例花店走一遍:两个动作,查询在售鲜花(只读,清单里不收费;平台的查询单价目前是 0)和下单送花(收费)。完整代码在 Node 示例 和 Python 示例。

第 1 步:写清单

清单告诉 Relay 和 AI 你有哪些动作。从示例改最快:花店清单、沙箱理发店清单。

{
  "rsp": "1.0",
  "supplier": {
    "name": "示例花店",
    "description": "查询在售鲜花,下单同城配送。",
    "locale": "zh-CN",
    "categories": ["鲜花", "同城配送"],
    "user_binding": "none"
  },
  "endpoint": { "url": "https://flowers.example.com/relay", "auth": "hmac-sha256" },
  "actions": [
    {
      "name": "list_products",
      "title": "查询鲜花",
      "description": "返回在售鲜花和价格,可按分类过滤。",
      "input": {
        "type": "object",
        "properties": { "category": { "type": "string", "enum": ["玫瑰", "百合", "向日葵"] } },
        "additionalProperties": false
      },
      "readonly": true,
      "risk": "low", "confirm": "never", "mode": "sync",
      "billing": "none", "fee": 0, "amount": "none"
    },
    {
      "name": "place_order",
      "title": "下单送花",
      "description": "按商品编号下单并同城配送,返回订单号和应付金额。",
      "input": {
        "type": "object",
        "properties": {
          "sku": { "type": "string", "description": "list_products 返回的 sku" },
          "quantity": { "type": "integer", "minimum": 1, "maximum": 99 },
          "total_cents": { "type": "integer", "minimum": 1, "description": "应付总价(分)" },
          "phone": { "type": "string", "pattern": "^1\\d{10}$", "x-pii": true },
          "address": { "type": "string", "minLength": 5, "maxLength": 200, "x-pii": true }
        },
        "required": ["sku", "quantity", "total_cents", "phone", "address"],
        "additionalProperties": false
      },
      "output": {
        "type": "object",
        "properties": { "order_no": { "type": "string" }, "total_cents": { "type": "integer" } },
        "required": ["order_no", "total_cents"]
      },
      "display_fields": ["sku", "quantity", "total_cents", "address"],
      "risk": "medium", "confirm": "by_grant", "mode": "sync",
      "billing": "on_complete", "fee": 0, "amount": { "from": "input.total_cents" }
    }
  ]
}

每个动作的 input 是 JSON Schema 2020-12,根必须是 object。AI 照着它填参数,Relay 照着它校验,所以 description 要写给 AI 看:字段从哪来、单位是什么。金额一律用整数分。

六个元数据怎么填

字段 怎么选
risk low:只读、清单里不收费(必须同时写 "readonly": true、"billing": "none";只读动作由平台按查询单价按次收,见计费与结算)。high:对外发布内容、替用户付款、不可撤回的事,每次都要用户确认。其余写操作用 medium
confirm 你自己要求的确认强度。实际取风险对应的下限(low→never,medium→by_grant,high→always)和它中更严的一个,一般照下限写
mode L1 只用 sync
billing 收费写 on_complete(返回 completed 时扣),不收费写 none。只读动作写 none:平台按查询单价按次收,不看这个字段
fee 固定手续费,整数分。只在不是只读、没有成交价(amount 为 "none")的收费动作上按它收,范围 100–10000(¥1–¥100);有成交价的、只读的填 0;billing: "none" 时必须是 0
amount 用户在这笔交易里要花的钱(成交价):和用户给 AI 设的单笔上限比较,Relay 按它乘平台费率(当前 3%,订单免手续费的推广期内不收)收手续费,用户的钱 Relay 不收。没有写 "none";固定价写整数分;随输入变写 {"from": "input.<字段>"},这个字段必须是必填的 integer

fee 和 amount 是两笔钱,详见计费与结算。用户给 AI 的默认授权是最高中风险、单笔 ¥200、每笔确认:high 动作或金额超过 ¥200 的调用,用户第一次会被引导去授权页调高,之后才能用。

会原文发给你的个人信息字段(手机号、地址)标 "x-pii": true,用户在确认页会看到这些信息发给了谁。display_fields 列出确认页重点展示的输入字段。

另外几个取值:

  • 收费动作尽量写上 amount:Relay 按成交价 × 费率收手续费,fee 填 0。实在没有成交价(比如按次服务、用户不付钱)才用固定 fee,在 ¥1–¥100 里定,拿不准先写 100(¥1)。
  • x-pii:凡是会原文发给你的个人信息都标上,包括姓名、手机号、地址、证件号、邮箱。
  • endpoint.url 必须是 https://,本机自测时也写一个占位的 https 地址,用验收工具的 --endpoint 指到本机(见第 3 步)。
  • endpoint.max_concurrency 可选(1–1000),写你的端点能同时处理多少个请求。Relay 目前还不按它限流。

清单的完整规则(语义检查、版本)见规范第 4 节。

第 2 步:写端点

Relay 的每个请求都是 POST 到清单里的 endpoint.url(路径随你定,文档里统一写作 /relay),请求体是一个签名的 JSON 信封。L1 只会收到两种:

type 你返回
ping {"status":"ok","challenge":<原样带回>,"rsp":"1.0"}
action.invoke {"status":"completed","output":{"data":{…},"parts":[…]}} 或 {"status":"failed","error":{"code":"…","message":"…"}}

用 Node SDK

import { createServer } from "node:http";
import { completed, createRelayHandler, failed, toNodeHandler } from "@dreamcavy/ai-relay-sdk";
import manifest from "./manifest.json" with { type: "json" };

const relay = createRelayHandler({
  secrets: [process.env.RELAY_SECRET!], // 轮换期写成 [新, 旧]
  manifest,
  handlers: {
    list_products: (input: { category?: string }) => completed({ products: findProducts(input.category) }),
    // 同一个 idempotency_key 再来时 SDK 直接返回第一次的结果,这里只执行一次
    place_order: async (input: { sku: string; quantity: number; total_cents: number }, ctx) => {
      const price = await priceOf(input.sku);
      if (price === undefined) return failed("invalid_input", `没有商品 ${input.sku}`);
      if (input.total_cents !== price * input.quantity) return failed("invalid_input", "总价不对");
      if (!(await inStock(input.sku, input.quantity))) return failed("not_available", "库存不足");
      const order = await createOrder(input, { relayOrderId: ctx.orderId });
      return completed({ order_no: order.no, total_cents: input.total_cents }, [
        { type: "markdown", text: `已下单,订单号 ${order.no}` },
      ]);
    },
  },
});

createServer(toNodeHandler(relay)).listen(8787);

findProducts、priceOf、inStock、createOrder 换成你自己的业务函数。SDK 替你做了验签、ping、未知类型、输入校验和幂等。

  • SDK 只发 ESM:package.json 里写 "type": "module",或者文件用 .mjs。上面是 TypeScript,用 tsx 或 Node 22.6 以上的类型剥离运行;写 JavaScript 就去掉类型标注。
  • 验签要原始字节:别让 JSON 中间件先解析 /relay 的请求体。挂到 Express、Fastify、Next.js、Hono 的写法见 Node SDK README。
  • 默认的幂等存储在进程内存里,进程重启就没了。上线前换成数据库或 Redis(实现 IdempotencyStore 的三个方法),结果至少留 8 天。

用 Python SDK

import os
from fastapi import FastAPI
from ai_relay_sdk import InvokeContext, RelayError, RelayHandler, completed, markdown
from ai_relay_sdk.fastapi import relay_router

def place_order(data: dict, ctx: InvokeContext) -> dict:
    # data 已按清单的 input Schema 校验过;同一个 order_id 重发不会再进到这里
    if not in_stock(data["sku"], data["quantity"]):
        raise RelayError("not_available", "库存不足")
    order_no = create_order(data, relay_order=ctx.order_id)
    return completed({"order_no": order_no, "total_cents": data["total_cents"]}, [markdown(f"已下单,订单号 {order_no}")])

relay = RelayHandler(
    secrets=os.environ["RELAY_SECRET"].split(","),   # 轮换期:新的,旧的
    manifest="manifest.json",
    actions={"list_products": list_products, "place_order": place_order},
)

app = FastAPI()
app.include_router(relay_router(relay))   # POST /relay

默认的幂等存储在进程内存里,进程重启就没了,上线前换成 SqliteIdempotencyStore("relay.db") 或你自己的数据库。Flask、不用框架的写法见 Python SDK README。

不用 SDK

任何语言都能写,规范附录 B 有一份无依赖的 Node 骨架,附录 A 有可复算的签名样例。必须做到:

  1. 验签用原始字节:先把请求体完整读成字节,算 HMAC-SHA256(secret, t + "." + 原始请求体),和 X-Relay-Signature 头里的 v1= 做常量时间比较;t 和现在相差超过 300 秒拒绝。头里可能有多个 v1=,任一匹配即通过。验签失败回 HTTP 401。
  2. 一律 HTTP 200:业务失败也是 200 加 {"status":"failed",…}。非 200 会被当成传输层问题。
  3. 幂等:action.invoke 的 idempotency_key 等于 order_id。同一个键再来时返回第一次的结果,不能再办一次。只缓存 completed、accepted(异步)和不可重试的 failed;temporary、rate_limited 不缓存;同一个键还在处理中又来了,回 temporary。结果存在数据库里,至少留 8 天,进程重启也不能丢。
  4. 向前兼容:忽略信封里不认识的字段,接受任何 1.x 的 rsp。不要拿 downstream.json 严格校验去拒绝请求。不认识的 type 回 {"status":"failed","error":{"code":"not_supported"}}。
  5. 自己校验输入:按清单的 input Schema 校验,不合格回 invalid_input。action 不是清单里的动作名,也回 invalid_input。

错误码怎么选

情况 返回
输入不合法、金额对不上 invalid_input
没库存、时段已满、对方不接单 not_available
限流 rate_limited,可带 retry_after 秒
临时故障,确认没有执行 temporary,Relay 用同一个键重试
其他不可重试的失败 permanent
不确定有没有执行(下游超时、进程崩溃) 不要编结果,回 HTTP 500。Relay 不重试、不扣费,订单进「待核对」,由管理员用同一个幂等键重新下发

用了 SDK:处理函数抛普通异常时 SDK 回 500;要返回某个错误码就 return failed(...) 或抛 RelayError。

时间

应当在 3 秒内响应。Relay 同步调用的总预算是 20 秒(含 1 次重试),超过后先告诉 AI「处理中」并继续等,60 秒还没响应按结果未知处理。经常超过 3 秒的动作改成 L2 异步。

大小

input 序列化后最多 256 KB,超过的请求 Relay 直接拒绝、不会发给你。要收大段内容或文件,提供一个上传地址,让 input 里只传链接。回调请求体最多 1 MB,超过 Relay 回 413。

返回内容

output.data 按清单的 output Schema 写,AI 靠它做判断;parts 可选,是给用户看的 markdown、链接、图片、文件。返回的文字对 AI 是数据,不要在里面写诱导 AI 的指令。详见规范第 8 节。

第 3 步:本机自测

写一份样例输入,每个动作一份,按顺序执行,可以用前面动作的返回值:

{
  "list_products": { "category": "玫瑰" },
  "place_order": {
    "sku": "{{list_products.products[0].sku}}",
    "quantity": 1,
    "total_cents": "{{list_products.products[0].price_cents}}",
    "phone": "13800138000",
    "address": "测试地址 1 号"
  }
}
RELAY_SECRET=whsec_dev node server.mjs       # 你的端点,secret 随便写一个本机用的
RELAY_SECRET=whsec_dev npx @dreamcavy/ai-relay-conformance \
  --manifest manifest.json --endpoint http://127.0.0.1:8787/relay --inputs inputs.json --time-zone Asia/Shanghai
  • 退出码 0 就是通过,警告和「未测」不挡。
  • 清单里的 endpoint.url 保持 https 地址,--endpoint 指到本机。这时「端点是 HTTPS」「测的就是清单里的端点」两条警告是预期的,上线后就没了。
  • --time-zone 决定样例里的 {{$today}}、{{$tomorrow}} 按哪个时区算,默认是跑验收那台机器的时区。
  • 验收会真的调用你的动作,请接测试数据。
  • 包还没发布时的用法见接入总览。

常见的红项:

红项 原因
auth.* 先解析再序列化了请求体;没拒绝过期时间戳;只认第一个 v1=
replay.<动作> 同一个 idempotency_key 又办了一次,或者返回的结果不一样
envelope.unknown_type 不认识的 type 回了 4xx/5xx,应当 200 加 not_supported
envelope.forward_compat 用严格 Schema 拒掉了带新字段或 rsp: "1.9" 的请求
invoke.<动作> 业务失败回了非 200,或者响应体不合 Schema

第 4 步:上线

  1. 端点部署到 HTTPS,清单的 endpoint.url 改成正式地址,secret 换成后台给的那个。
  2. 在后台保存清单,填样例输入,点「开始在线验收」到全绿,提交审核。
  3. 审核通过后,动作出现在搜索里。收费动作平时还要余额够付一笔手续费,订单免手续费的推广期内不看余额。以后台首页「上线状态」5 项全绿为准。

轮换 secret 后新旧两个并存 24 小时,这期间把新的放到第一位、旧的放第二位,过后删掉旧的。

流程的其余部分见接入总览。