ArkClaw 企业版
ArkClaw 企业版
文档指南
请输入
- 文档首页
ArkClaw 企业版ArkClaw 企业版最佳实践ArkClaw A2A 接口集成最佳实践ArkClaw A2A 接口集成基础调用说明
复制全文
下载 pdfArkClaw A2A 接口集成基础调用说明
文档反馈
问问助手本文介绍如何通过 A2A（Agent-to-Agent）JSON-RPC 标准接口远程调用 ArkClaw 智能体能力。支持同步阻塞、流式推送、异步轮询三种调用模式，可适配简单问答、实时输出、长耗时任务等各类业务场景。开发者可通过 curl、Python、Postman 等任意 HTTP 客户端完成对接调试与业务集成。
本文将以curl命令和Python代码为例介绍接口调用方式。
ArkClaw A2A 接口能做什么
ArkClaw A2A（Agent-to-Agent）接口基于 JSON-RPC 2.0 协议，将 ArkClaw 智能体的能力开放给外部系统调用。通过该接口可以实现：
- 业务系统集成智能问答：将智能体接入业务后端，提供知识问答、语义搜索等能力。
- 长耗时任务异步处理：适合报告生成、复杂检索等场景，避免同步等待。
- 流式交互体验：前端实时展示智能体输出，提升用户体验。
- 多轮上下文对话：通过会话保持支持连续追问、上下文推理。
请求地址与公共参数说明
开启并获取 Webhook 请求地址
开启 ArkClaw 实例基于 A2A 协议的外部接入能力后，平台将自动生成 Endpoint URL 和 API Key，供外部业务系统通过公网访问调用当前 ArkClaw。
- 在目标 ArkClaw 实例详情页面，切换至“设置”页签后，单击
图标，开启 Webhook。了解更多
-
- 单击
图标，复制生成的公网访问链接，以备后续调用时使用。了解更多
-
公共参数说明
- 接口统一请求地址
- 所有 A2A 接口请求统一访问以下公网地址，链接中占位参数请替换为当前 ArkClaw 的实际配置信息：
- https://{GATEWAY_HOST}/a2a/jsonrpc?apikey={API_KEY}&clawId={CLAW_ID}
说明
- 公网地址参数说明
-
公共请求头
所有接口请求统一固定请求头：Content-Type: application/json。
通过curl命令调用 A2A 接口
curl调用适合快速调试、临时验证接口，无需开发环境，复制命令直接在终端执行，多用于测试排错、快速看返回结构。您可以按需选择以下任意方式进行 A2A 接口调试。
适用于需要实时输出结果、长文本分段返回或希望在任务处理中持续获取中间状态的交互场景。
说明
该方式基于 SSE（Server-Sent Events）协议持续推送数据，服务端会在同一连接内按事件帧逐步返回处理状态和生成内容，便于客户端实时展示输出结果。
- 发起流式请求：以询问“你是谁”为例，执行下方命令，通过流式接口发送用户对话消息。
注意
仅下述参数需结合业务实际信息替换，其余参数为固定配置，无需改动。
- {URL}： 接口调用地址，为必填替换项，请替换为 ArkClaw 控制台提供的公网 Endpoint，详情请参见 开启并获取Webhook访问地址。
- id：单次请求唯一标识，支持自定义字符串或 UUID，用于区分每一次接口调用请求。
- messageId：本条对话消息唯一 ID，支持自定义字符串或 UUID，用于区分会话内不同消息。
- text：用户提问内容，可自定义业务问题，按需替换为实际对话内容。
- curl '{URL}' \-H 'content-type: application/json' \-d '{"jsonrpc": "2.0","id": "unique-request-id","method": "message/stream","params": {"message": {"parts": [{"kind": "text", "text": "你是谁"}],"messageId": "unique-message-id","role": "user"}}}'
- 查看流式响应：请求发起后，流式接口基于 SSE 协议返回多条data:事件帧，分为两类：
- working 处理中帧：仅推送任务运行状态、contextId会话标识，无完整回复文本，用于实时展示加载状态。
- completed 完成帧：推理结束后推送，包含完整 AI 回复消息、会话上下文、输出产物 artifacts，代表本轮对话结束，停止监听流式数据
使用Python命令调用 A2A 接口
通过 Python 的 requests 库调用 A2A 接口，适用于在服务端脚本、自动化任务或后端应用中接入 ArkClaw 的 A2A 能力。调用方式包括同步阻塞调用、流式调用，以及异步任务轮询三种模式。
步骤一：安装依赖环境
- 执行如下命令，安装网络请求依赖库。
- pip install requests
- 打开终端，执行以下命令，展示 Version 版本信息、Location 安装路径，即为安装完成。
- pip show requests
步骤二：配置接口地址并封装公共请求方法
将接口调用地址替换为 ArkClaw 控制台提供的公网 Endpoint，并封装消息构造方法与通用请求方法。便于后续的同步调用、流式调用和异步任务调用复用，可避免重复编写请求构造。
import json
import time
import uuid
import requests
URL = f"https://sd8ojs9tiq5rjni****.apigateway-cn-beijing.volceapi.com/a2a/jsonrpc?apikey=9ac252facbad4e519ddd6****&clawId=ci-yeok2q4dmo3pe****"
HEADERS = {"Content-Type": "application/json"}
def build_message(text):
return {
"parts": [{"kind": "text", "text": text}],
"messageId": str(uuid.uuid4()),
"role": "user",
}
def post_jsonrpc(method, params, timeout=120):
payload = {
"jsonrpc": "2.0",
"id": str(uuid.uuid4()),
"method": method,
"params": params,
}
response = requests.post(URL, headers=HEADERS, json=payload, timeout=timeout)
response.raise_for_status()
data = response.json()
if "error" in data:
raise RuntimeError(data["error"])
return data["result"]
参数说明如下表所示，请根据实际值替换。
步骤三：按业务场景选择调用方式
将基础封装代码与同步阻塞、流式或异步轮询调用示例拼接至同一个 Python 脚本，直接运行该文件，程序将自动输出智能体返回的完整对话结果。
同步阻塞调用适用于适用于普通问答、短文本生成、结果可在单次请求内返回的场景，例如单轮问答或轻量级指令执行。
调用通用请求函数 post_jsonrpc，传入方法名 message/send 和构造好的对话消息体，接口会在服务端处理完成后一次性返回最终结果。
说明
按需将send_sync("你是谁") 括号内的文本，替换成您想要提问的内容即可。
def send_sync(text):
result = post_jsonrpc(
"message/send",
{"message": build_message(text)},
)
parts = result.get("status", {}).get("message", {}).get("parts", [])
return "".join(part.get("text", "") for part in parts if part.get("kind") == "text")
answer = send_sync("你是谁")
print(answer)
适用于长文本生成、分段返回和需要实时展示结果的场景。
message/stream基于 SSE（Server-Sent Events）协议返回事件流，Python 侧通过 stream=True 持续读取服务端返回的增量内容。
将方法名改为 message/stream，通过 stream=True 持续读取 SSE 事件流，逐帧输出增量内容。当返回状态 state 变为 completed、failed 或 canceled 时，表示本次流式任务已结束，客户端应停止读取后续事件。
说明
- 按需将send_sync("你是谁") 括号内的文本，替换成您想要提问的内容即可。
- 若业务存在长耗时推理或生成场景，可根据接口响应时长适当调大timeout参数取值，以避免因等待时间不足导致请求提前结束。
def send_stream(text):
payload = {
"jsonrpc": "2.0",
"id": str(uuid.uuid4()),
"method": "message/stream",
"params": {"message": build_message(text)},
}
with requests.post(URL, headers=HEADERS, json=payload, stream=True, timeout=120) as response:
response.raise_for_status()
response.encoding = "utf-8"
for line in response.iter_lines(decode_unicode=True):
if not line or not line.startswith("data:"):
continue
event = json.loads(line[len("data:"):].strip())
result = event.get("result", {})
state = result.get("status", {}).get("state")
parts = result.get("status", {}).get("message", {}).get("parts", [])
text_delta = "".join(part.get("text", "") for part in parts if part.get("kind") == "text")
if text_delta:
print(text_delta, end="", flush=True)
if state in {"completed", "failed", "canceled"}:
break
send_stream("你是谁")
任务执行时间较长、客户端不希望持续阻塞等待的场景。
说明
- 按需将send_sync("你是谁") 括号内的文本，替换成您想要提问的内容即可。
- 若业务存在长耗时推理或生成场景，可根据接口响应时长适当调大timeout参数取值，以避免因等待时间不足导致请求提前结束。
- 先提交异步任务，获取 task_id。
- 调用 start_async_task 函数，通过 message/send 接口并配置 configuration.blocking=False，告知服务端开启异步非阻塞模式。
- 服务端接收请求后立即返回 task_id，不等待模型推理。
- 该模式可显著缩短接口响应时间，避免客户端长连接阻塞。
- def start_async_task(text):result = post_jsonrpc("message/send",{"message": build_message(text),"configuration": {"blocking": False},},)return result["id"]
- 调用 get_task 函数，通过 tasks/get 按 task_id 查询任务执行状态，并设置固定轮询间隔。
- 当 state 进入 completed、failed 或canceled时停止轮询。
- 建议设置合理的轮询间隔（如 1～5 秒），避免高频请求造成资源浪费。
- def get_task(task_id):return post_jsonrpc("tasks/get", {"id": task_id})def wait_task(task_id, interval=2, timeout=300):deadline = time.time() + timeoutwhile time.time() < deadline:result = get_task(task_id)state = result.get("status", {}).get("state")if state in {"completed", "failed", "canceled"}:return resulttime.sleep(interval)raise TimeoutError(f"task timeout: {task_id}")task_id = start_async_task("你是谁")result = wait_task(task_id)parts = result.get("status", {}).get("message", {}).get("parts", [])print("".join(part.get("text", "") for part in parts if part.get("kind") == "text"))
最近更新时间：2026.07.16 20:01:02
这个页面对您有帮助吗？
有用
有用
无用
无用