开发者接入 — REST API 与 MCP 协议

接入方式

REST API

异步任务模型：提交 → 拿 job_id → 轮询 → 下载结果。支持幂等键与回调。

MCP Server

开源连接器，在 Cursor / Claude Desktop 里直接调用出图能力，走一次性设备配对，不在客户端存长期密钥。

飞书中继

运营在飞书多维表格填图片链接，机器人自动排队跑图，结果自动回填到表格。

Python 批量调用示例

包含幂等、退避轮询、并发控制三个生产环境必需的设计。

import os, hashlib, time, requests

BASE = os.environ["SHANGTU_BASE_URL"]
HEADERS = {
    "Authorization": f"Bearer {os.environ['SHANGTU_DEVICE_SECRET']}",
    "X-Shangtu-Pair-Id": os.environ["SHANGTU_PAIR_ID"],
    "X-Shangtu-Source": "api",
    "Content-Type": "application/json",
}

def idempotency_key(sku, feature, payload):
    raw = f"{sku}|{feature}|{sorted(payload.items())}"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def submit(sku, feature_id, inputs, options):
    resp = requests.post(
        f"{BASE}/bridge/jobs",
        headers={**HEADERS,
                 "Idempotency-Key": idempotency_key(sku, feature_id, {**inputs, **options})},
        json={"feature_id": feature_id, "inputs": inputs, "options": options},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["bridge_job_id"]

def poll(job_id, timeout=300):
    interval, elapsed = 2.0, 0
    while elapsed < timeout:
        r = requests.get(f"{BASE}/bridge/jobs/{job_id}", headers=HEADERS, timeout=30)
        data = r.json()
        if data["status"] in ("succeeded", "failed"):
            return data
        time.sleep(interval)
        elapsed += interval
        interval = min(interval * 1.6, 15)
    raise TimeoutError(job_id)

job = submit("A1023", "template_transfer",
             {"product_images": ["https://cdn.example.com/a1023.jpg"],
              "reference_images": ["https://cdn.example.com/ref.jpg"]},
             {"resolution": "2k", "replace_target": "garment"})
print(poll(job))

MCP 客户端配置

加进 mcp.json 后重启客户端，工具列表里会出现 connect / submit / get 三个方法。

{
  "mcpServers": {
    "shangtuce": {
      "command": "node",
      "args": ["/absolute/path/to/dist/packages/mcp-server/src/index.js"],
      "env": {
        "SHANGTU_BASE_URL": "https://<your-app>.runninghub.cn/__pb/api/shangtu/v1",
        "SHANGTU_UPLOAD_ROOTS": "/Users/me/ecommerce-assets"
      }
    }
  }
}

生产环境四个坑

幂等键

不设这个，一次网络抖动就是一次重复扣费。用 SKU + 功能 + 参数摘要哈希。

轮询退避

2 秒起步，每次乘 1.6，封顶 15 秒。别用固定 200ms 轮询，会被限流。

并发上限

3—5 并发是稳态。超过开始 429，总耗时反而更长。

结果落盘

返回 URL 有有效期，必须下载到自己的对象存储再引用。

常见问题

幂等键必须自己生成吗？

必须。API 支持 Idempotency-Key 但不会替你生成。重试时如果 key 变了就会重复建单、重复扣费。建议用 SKU + 功能 ID + 参数摘要做哈希。

并发能开多大？

实测 3—5 并发是稳态上限，超过会开始返回 429，反而更慢。建议用线程池控制在 3。

返回的图片 URL 能长期用吗？

不能。结果 URL 有有效期，生产环境必须下载到自己的对象存储再引用。

把出图接进你自己的流水线