YunQi AI DocsAPI ReferenceAgent API 文档
控制台 ↗
文档/API 接入
API 接入Anthropic

Anthropic Messages

Claude 原生 Messages API

接口

POSThttps://www.yunqiai.chat/v1/messages

所有请求使用 HTTPS，并通过请求头携带访问密钥。

请求参数

参数类型说明
model必填string从 GET /v1/models 返回结果中填写完整模型 ID
messages必填arrayuser / assistant 消息数组；图片作为 image 内容块
max_tokens必填integer最大输出 Token，填写正整数；上限随模型变化
systemstring | array顶层系统提示；Messages API 没有 system 角色
streamboolean是否以 SSE 流式返回，默认 false
temperaturenumber0–1；Claude 4.7 及以后模型建议省略采样参数
top_pnumber0–1；与 temperature 通常只设置一个
stop_sequencesstring[]自定义停止序列
toolsarrayAnthropic 工具定义

请求示例

cURL
curl --request POST \
  --url https://www.yunqiai.chat/v1/messages \
  --header "x-api-key: YOUR_API_KEY" \
  --header "anthropic-version: 2023-06-01" \
  --header "Content-Type: application/json" \
  --data '{
    "model": "claude-sonnet-4-6",
    "max_tokens": 1024,
    "messages": [
      {
        "role": "user",
        "content": "你好"
      }
    ]
  }'

工具调用闭环

先随请求发送工具定义；模型返回调用参数后，由客户端校验参数并执行本地函数，再用同一个调用 ID 回传结果。Responses 多轮由客户端保存原始 input 与完整 output[]（含 reasoning 项），追加全部 function_call_output；继续发送 tools，不依赖 previous_response_id。

JSON · define tools
{
  "model": "claude-sonnet-4-6",
  "max_tokens": 1024,
  "messages": [
    {
      "role": "user",
      "content": "上海现在几点？"
    }
  ],
  "tools": [
    {
      "name": "get_time",
      "description": "返回指定时区的当前时间",
      "input_schema": {
        "type": "object",
        "properties": {
          "timezone": {
            "type": "string",
            "description": "IANA 时区，例如 Asia/Shanghai"
          }
        },
        "required": [
          "timezone"
        ]
      }
    }
  ]
}
JSON · model returns tool_use
{
  "type": "tool_use",
  "id": "toolu_01JY7X",
  "name": "get_time",
  "input": {
    "timezone": "Asia/Shanghai"
  }
}
协议模型返回客户端回传
Anthropic Messagestool_use · id · inputtool_result · tool_use_id
JSON · return tool_result
{
  "model": "claude-sonnet-4-6",
  "max_tokens": 1024,
  "messages": [
    {
      "role": "user",
      "content": "上海现在几点？"
    },
    {
      "role": "assistant",
      "content": [
        {
          "type": "tool_use",
          "id": "toolu_01JY7X",
          "name": "get_time",
          "input": {
            "timezone": "Asia/Shanghai"
          }
        }
      ]
    },
    {
      "role": "user",
      "content": [
        {
          "type": "tool_result",
          "tool_use_id": "toolu_01JY7X",
          "content": "{\"time\":\"14:30\"}"
        }
      ]
    }
  ],
  "tools": [
    {
      "name": "get_time",
      "description": "返回指定时区的当前时间",
      "input_schema": {
        "type": "object",
        "properties": {
          "timezone": {
            "type": "string"
          }
        },
        "required": [
          "timezone"
        ]
      }
    }
  ]
}

Base64 图片输入

图片使用 image 内容块，source.type 设为 base64，并显式提供 media_type。

JSON · Base64
{
  "model": "claude-sonnet-4-6",
  "max_tokens": 256,
  "messages": [
    {
      "role": "user",
      "content": [
        {
          "type": "image",
          "source": {
            "type": "base64",
            "media_type": "image/png",
            "data": "BASE64_IMAGE_DATA"
          }
        },
        {
          "type": "text",
          "text": "描述这张图片"
        }
      ]
    }
  ]
}

PDF 文件输入

把 PDF 放在 document 内容块中，source.type 设为 base64，并在 source.data 中传入裸 Base64。提示词作为同一条消息里的 text 内容块。

Python · base64 PDF
import base64
import os
from pathlib import Path

import requests

pdf_path = Path("report.pdf")
pdf_data = base64.b64encode(pdf_path.read_bytes()).decode("ascii")
payload = {
    "model": "claude-opus-4-7",
    "max_tokens": 1024,
    "messages": [{
        "role": "user",
        "content": [
            {
                "type": "document",
                "source": {
                    "type": "base64",
                    "media_type": "application/pdf",
                    "data": pdf_data,
                },
            },
            {"type": "text", "text": "提取报告中的结论与关键数字"},
        ],
    }],
}

response = requests.post(
    "https://www.yunqiai.chat/v1/messages",
    headers={
        "x-api-key": os.environ["YUNQIAI_API_KEY"],
        "anthropic-version": "2023-06-01",
        "Content-Type": "application/json",
    },
    json=payload,
    timeout=(10, 180),
)
response.encoding = "utf-8"
if not response.ok:
    request_id = response.headers.get("x-request-id") or response.headers.get("request-id")
    raise RuntimeError(f"HTTP {response.status_code}; request_id={request_id}; body={response.text[:2000]}")
result = response.json()
print("".join(block.get("text", "") for block in result.get("content", []) if block.get("type") == "text"))
来源document.source 写法适用条件
内联 Base64type: base64 · media_type · data本地文件可直接编码后发送

流式输出

客户端以 SSE 逐行读取 data: 事件。不要按网络数据块直接解码 JSON；同一事件可能被拆成多个传输片段。

阶段事件读取内容
开始message_start消息元数据
内容content_block_start → content_block_delta → content_block_stopdelta.text
工具调用content_block_start / input_json_delta.partial_json保存 tool_use.id 与 name，按内容块 index 拼接参数
完成message_delta → message_stop停止原因与最终用量
错误errorerror.type / error.message / request_id
Python · Messages SSE
import json
import os

import requests

def iter_sse_data(response):
    data_lines = []
    for line in response.iter_lines(decode_unicode=True):
        if line == "":
            if data_lines:
                data = "\n".join(data_lines)
                data_lines = []
                if data:
                    yield data
            continue
        if line.startswith(":"):
            continue
        field, _, value = line.partition(":")
        if value.startswith(" "):
            value = value[1:]
        if field == "data":
            data_lines.append(value)
    # A frame without its terminating blank line is incomplete.


payload = {
    "model": "claude-sonnet-4-6",
    "max_tokens": 1024,
    "messages": [{"role": "user", "content": "用三句话解释量子计算"}],
    "stream": True,
}

with requests.post(
    "https://www.yunqiai.chat/v1/messages",
    headers={
        "x-api-key": os.environ["YUNQIAI_API_KEY"],
        "anthropic-version": "2023-06-01",
        "Content-Type": "application/json",
    },
    json=payload,
    stream=True,
    timeout=(10, 180),
) as response:
    response.encoding = "utf-8"
    if not response.ok:
        request_id = response.headers.get("x-request-id") or response.headers.get("request-id")
        raise RuntimeError(f"HTTP {response.status_code}; request_id={request_id}; body={response.text[:2000]}")
    message_stopped = False
    tool_calls = {}

    for data in iter_sse_data(response):
        event = json.loads(data)
        if event.get("type") == "content_block_start":
            block = event.get("content_block") or {}
            if block.get("type") == "tool_use":
                tool_calls[event.get("index", 0)] = {
                    "id": block.get("id", ""),
                    "name": block.get("name", ""),
                    "input": block.get("input") or {},
                    "arguments": "",
                }
        elif event.get("type") == "content_block_delta":
            delta = event.get("delta", {})
            if delta.get("type") == "text_delta":
                print(delta.get("text", ""), end="", flush=True)
            elif delta.get("type") == "input_json_delta":
                index = event.get("index", 0)
                current = tool_calls.setdefault(index, {"id": "", "name": "", "input": {}, "arguments": ""})
                current["arguments"] += delta.get("partial_json", "")
        elif event.get("type") == "message_stop":
            message_stopped = True
        elif event.get("type") == "error":
            error = event.get("error", {})
            raise RuntimeError(f"{error.get('type')}: {error.get('message')}")

if not message_stopped:
    raise RuntimeError("stream ended before message_stop")

print()
for call in tool_calls.values():
    arguments = json.loads(call["arguments"]) if call["arguments"] else call["input"]
    tool_result = {"type": "tool_result", "tool_use_id": call["id"], "content": "工具执行结果"}
    print("tool", call["id"], call["name"], arguments, tool_result)

错误、重试与超时

情况客户端处理
连接超时单独设置连接超时，例如 10 秒；确认网络与 Base URL
读取超时文本请求可从 180 秒起设置；图像等耗时任务可设为 300 秒
408、5xx、网络超时POST 结果可能未知；记录请求 ID，不自动重放。已取得 MJ id 时只重试 GET 查询
429先区分限流与额度不足；限流按 Retry-After 等待，额度不足先处理额度
400、401、403、404不要自动重试；先检查错误正文、鉴权头、接口路径、模型与参数
JSON · error response body
{
  "type": "error",
  "error": {
    "type": "invalid_request_error",
    "message": "请求参数无效"
  },
  "request_id": "req_01JY7X"
}
协议错误正文中的请求 ID响应头中的请求 ID
Anthropic Messagesrequest_idrequest-id
Python · single attempt and timeout
import os
import requests

url = "https://www.yunqiai.chat/v1/messages"
headers = {"x-api-key": os.environ["YUNQIAI_API_KEY"], "anthropic-version": "2023-06-01"}
payload = {"model": "claude-sonnet-4-6", "max_tokens": 1024, "messages": [{"role": "user", "content": "你好"}]}

# A POST may have reached the server even when the client times out.
# Do not automatically replay it or switch to a different model.
response = requests.post(url, headers=headers, json=payload, timeout=(10, 180))
request_id = response.headers.get("x-request-id") or response.headers.get("request-id")
if not response.ok:
    raise RuntimeError(
        f"HTTP {response.status_code}; request_id={request_id or 'unavailable'}; "
        "check the redacted error response before starting a new request"
    )
print(response.json())

响应

JSON
{
  "id": "msg_01JY7X",
  "type": "message",
  "role": "assistant",
  "model": "claude-sonnet-4-6",
  "content": [
    {
      "type": "text",
      "text": "你好！我是一个 AI 助手。"
    }
  ],
  "stop_reason": "end_turn"
}
YunQi AI 开放平台文档 · 客户接入与参数参考文档更新于 2026-09-14 · v1.0.11