agent-studio-client

Agent Studio 运行时 API 的 Python 客户端，用于在你自己的服务中调用 Studio 中已发布的 Agent。

• 同步 StudioClient 与异步 AsyncStudioClient，接口一一对应
• 事件流（SSE）断线自动按游标续传，并检测事件丢失
• 只对可重试错误（429、5xx、网络错误）做有限重试；创建 Run 的重试复用同一个幂等键，不会重复执行
• 支持多轮对话、取消、人工审批、客户端工具结果回传
• 只依赖 httpx 与 pydantic v2，支持 Python 3.10+

SDK 只负责接入。Agent 的执行、工具调用和状态都在 Studio 服务端完成，SDK 不在本地运行 Agent。

安装

pip install agent-studio-client
# 或
uv add agent-studio-client

TODO(发布)：包仓库地址尚未确定（外部 SDK 需求文档 FR-D2）。发布前可从源码安装：pip install git+<本仓库地址>。

准备工作

1. 在 Studio 控制台中发布 Agent，记下 Agent ID（或 slug，需服务端支持，见版本兼容）。
2. 在项目的凭证页面创建 API Key，scope 至少勾选 runs:write（只读场景用 runs:read）。Key 只在创建时显示一次，请保存在密钥管理系统中。
3. 记下 Studio 地址与项目 ID。

快速开始

from agent_studio import StudioClient

with StudioClient(
    base_url="https://studio.example.com",
    project_id="0190a1b2-...",
    api_key="sk_studio_...",
    actor_ref="user-42",          # 你的终端用户标识，见下文“actor 与安全”
) as studio:
    agent = studio.agent("0190a1b2-...")   # Agent ID

    run = agent.run("帮我总结一下今天的笔记")
    for chunk in run.text():               # 流式文本
        print(chunk, end="", flush=True)

    final = run.refresh()
    print(final.status, final.result)

异步版本：

from agent_studio import AsyncStudioClient

async with AsyncStudioClient(base_url=..., project_id=..., api_key=..., actor_ref="user-42") as studio:
    run = await studio.agent(agent_id).run("你好")
    async for chunk in run.text():
        print(chunk, end="")
    final = await run.refresh()

更多示例见 examples/：

核心概念

使用指南

多轮对话

保存 thread_id，下一轮传回即可：

first = agent.run("我叫小王")
first.wait()
second = agent.run("我叫什么？", thread_id=first.thread_id)
print(second.wait().result)

同一个 Thread 同时只能有一个未结束的 Run，否则返回 409 thread_busy。需要并发时，为每个会话创建各自的 Thread。

接收事件

for event in run.events():
    match event.type:
        case "message.delta":
            print(event.payload["text"], end="")
        case "tool.started":
            print(f"\n调用工具 {event.payload['toolId']}")
        case "run.failed":
            print("失败：", event.payload.get("code"))

• 收到终态事件（run.completed、run.failed、run.cancelled）后迭代自动结束。
• 断线后 SDK 自动带上 Last-Event-ID 续传，不会重复也不会遗漏；如果检测到 seq 不连续，抛出 StudioStreamGapError。
• 进程重启后可以从保存的游标继续：client.stream_events(run_id, after=last_seq)。
• 不需要实时推送时，可以用 client.list_events(run_id, after, limit) 分页读取，或轮询 client.get_run(run_id)。

常见事件类型：

服务端在同一主版本内可能新增事件类型或 payload 字段。SDK 会原样透传未知类型，你的代码应忽略不认识的事件，不要对它报错。

结果

final = run.wait()            # 消费事件直到终态，再读取最终 Run
final.status                  # "completed" | "failed" | "cancelled"
final.result                  # 回复文本
final.structured_result       # Agent 配置了结果 Profile 时的结构化结果
final.error_code              # 失败原因
final.usage                   # 步骤、模型/工具调用次数、token、费用

取消

run.cancel()

取消是幂等操作，对已结束的 Run 调用会原样返回。异步客户端中取消调用方任务（task.cancel()）会立即关闭本地连接，但不会取消服务端的 Run，需要显式调用 cancel()。

幂等与重试

create_run 总会带 Idempotency-Key，SDK 内部重试时复用同一个键，服务端只会创建一个 Run。

如果需要跨进程重启续接同一个请求（例如处理队列消息），请自己生成并保存幂等键：

run = client.create_run(thread_id, text, idempotency_key=f"msg-{message.id}")

同一个键配上不同的请求内容会返回 409 idempotency_conflict。

create_thread 不带幂等键，网络重试时可能多建出一个空 Thread。空 Thread 不会执行，也不会计费。

人工审批

Agent 调用配置为“需审批”的工具时，Run 进入 waiting_client 并发出 run.waiting_client 事件。这时 Run 不占用服务端的执行资源，可以一直等到请求过期。

for event in run.events():
    if event.type == "run.waiting_client":
        for action in run.pending():
            if action.kind == "approval":
                decision = ask_user(action.tool_name, action.args)   # 由真实用户决定
                run.decide(action, decision)  # "approve" / "reject" / "edit"（edit 需传 args）

• 决定必须来自终端用户的真实操作。SDK 不提供“自动批准”选项，请不要在代码中替用户批准有副作用的工具。
• 请求过期后返回 410 request_gone，Run 按服务端策略结束。

客户端工具

客户端工具在你的环境中执行，例如访问本地文件或调用你内网的服务。创建客户端时声明设备与可执行的工具：

from agent_studio import ClientCapabilities

studio = StudioClient(..., capabilities=ClientCapabilities(device_id="device-123", tools=["save_note@1"]))
...
for action in run.pending():
    if action.kind == "client_tool":
        output = execute_locally(action.tool_name, action.args)
        studio.submit_client_tool_result(run.id, action, approved=True, result=output)

一个后端服务代理多个用户

with_actor 返回一个共享连接池、但代表另一个用户的客户端：

studio = AsyncStudioClient(..., actor_ref="service")   # 进程内只创建一个
user_client = studio.with_actor(current_user.id)

删除用户数据

studio.with_actor("user-42").delete_actor_data()

这会删除该 actor 在项目中的全部 Thread、检查点、评测记录与文件。如果还有未结束的 Run，返回 409 actor_busy，请先取消并等待 Run 结束。

actor 与安全

• API Key 只能放在后端。 Studio 不开放浏览器跨域访问。前端和桌面端应调用你自己的后端，由后端使用 SDK。
• actor_ref 必须来自你后端已认证的用户。 不要直接使用前端传来的值，否则用户可以冒充他人，读取别人的会话。actor 的真实性由 Key 的持有方负责。
• actor_ref 只能包含字母、数字和 :._-，长度不超过 128 个字符。不要把邮箱、手机号等个人信息直接作为 actor，可以用 HMAC 或内部 ID 代替。
• 生产环境必须使用 HTTPS。如果 base_url 不是 HTTPS 且不是本机地址，SDK 会记录一条警告。
• SDK 不记录 API Key、请求体和事件内容。日志使用标准库的 logging.getLogger("agent_studio")，默认不输出。
• 网页、搜索、工具返回的内容由 Agent 处理，你在展示 result 时仍应把它当作不可信文本，例如进行 HTML 转义。

错误处理

所有错误都继承 StudioError：

状态码分类由服务端固定：400/401/403/404/409/410 是终止类，SDK 直接抛出，不重试；429/5xx 是可重试类，SDK 会自动重试，默认最多 3 次，遵守 Retry-After。

常见错误码：

from agent_studio import StudioApiError

try:
    run = agent.run(text, thread_id=thread_id)
except StudioApiError as e:
    if e.code == "thread_busy":
        ...  # 提示用户等待上一条回复
    else:
        raise

配置

StudioClient(
    base_url, project_id, api_key, actor_ref,
    timeout=30.0,            # 单次请求超时（秒）
    max_retries=3,           # 可重试错误的最大重试次数
    max_reconnects=5,        # 事件流连续失败的最大重连次数
    heartbeat_timeout=45.0,  # 事件流超过该时间没有任何数据即重连（服务端每 15 秒发一次心跳）
    capabilities=None,       # 客户端工具声明
    http_client=None,        # 自定义 httpx.Client，用于代理、自定义 CA、连接池大小等
)

需要自定义代理或证书时，传入自己的 httpx.Client。这种情况下 SDK 不会替你关闭它：

http = httpx.Client(verify="/path/to/ca.pem", proxy="http://proxy:3128", timeout=30)
studio = StudioClient(..., http_client=http)

版本兼容

SDK 主版本与 Studio 事件 Schema 主版本对齐，次版本和修订号独立演进。

使用 slug 或固定版本时，SDK 会先读取服务端的 GET /v1/meta 判断是否支持。旧版服务端会抛出 StudioFeatureUnavailableError，而不是返回一个看不懂的 404。只用 Agent ID、不固定版本的调用不会触发这次探测。

开发

uv sync                     # 安装依赖（含开发依赖）
uv run pytest               # 单元测试（使用 httpx.MockTransport，不需要 Studio 实例）
uv run ruff check .
uv run mypy src
uv build                    # 构建 wheel 与 sdist

目录结构：

src/agent_studio/
  __init__.py        公开导出
  client.py          同步客户端、AgentHandle、RunHandle
  async_client.py    异步客户端
  _api.py            端点定义（同步与异步共用）
  _transport.py      配置、请求头、错误映射、重试退避
  _sse.py            SSE 解析、事件序号与版本检查
  _meta.py           能力发现
  models.py          数据模型
  errors.py          错误类型
tests/               单元测试
examples/            示例

契约

服务端契约以 Agent Studio 仓库的 packages/contracts 为唯一来源。目前 models.py 按该契约手写了运行时 API 所需的子集，所有模型都允许未知字段，以兼容服务端新增字段。

TODO(契约包)：Studio 发布版本化的契约包（JSON Schema、OpenAPI、契约样本，见外部 SDK 技术设计 §7.2）之后，改为 scripts/fetch_contracts.py 按 contracts.lock 下载并校验摘要，再用 datamodel-code-generator 生成模型，在 CI 中检查生成结果有无漂移；同时接入契约样本测试，以及基于锁定版本服务镜像的集成测试。触发条件：契约包首次发布。

许可

TODO：待确定许可证。