Metadata-Version: 2.5
Name: agent-studio-client
Version: 0.0.1
Summary: Agent Studio 运行时 API 的 Python 客户端
Project-URL: Homepage, https://github.com/blublu1101/agent-studio-client
Project-URL: Repository, https://github.com/blublu1101/agent-studio-client
Project-URL: Issues, https://github.com/blublu1101/agent-studio-client/issues
License: MIT
License-File: LICENSE
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx<1,>=0.27
Requires-Dist: pydantic<3,>=2.6
Description-Content-Type: text/markdown

# 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。

## 安装

```bash
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。

## 快速开始

```python
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)
```

异步版本：

```python
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/`](examples)：

| 文件 | 内容 |
| --- | --- |
| `quickstart.py` | 最小调用 |
| `chat_with_approval.py` | 多轮对话，以及在终端中处理人工审批 |
| `fastapi_proxy.py` | 后端代理：前端调用你的服务，服务用 SDK 访问 Studio，并把文本以 SSE 转发给前端 |

## 核心概念

| 概念 | 说明 |
| --- | --- |
| **Project** | Studio 项目。API Key 属于某个项目，只能访问该项目的资源 |
| **Agent** | 在控制台配置并发布的 Agent。每次发布生成一个不可变版本，运行时默认使用 `production` 指针指向的版本 |
| **actor** | 你的应用中的终端用户，由 `actor_ref` 标识。Thread、Run 和事件都按 actor 隔离 |
| **Thread** | 一个会话。同一 Thread 中的多个 Run 共享上下文（记忆）|
| **Run** | 一次运行。状态：`queued → running → (waiting_client) → completed / failed / cancelled` |
| **Event** | Run 执行过程中产生的事件，带递增的 `seq`，可以从任意位置重放 |

## 使用指南

### 多轮对话

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

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

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

### 接收事件

```python
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 字段 |
| --- | --- | --- |
| `run.queued` / `run.started` | 已排队 / 开始执行 | `attempt` |
| `message.delta` | 流式文本片段 | `text`、`offset`、`messageId` |
| `message.completed` | 最终回复 | `text`、`result`（结构化结果）、`truncated` |
| `tool.started` / `tool.completed` / `tool.failed` | 工具调用 | `toolId`、`version`、`durationMs`、`code` |
| `model.started` / `model.completed` / `model.failed` | 模型调用 | `model`、`usage`、`durationMs` |
| `approval.requested` / `approval.decided` | 人工审批 | — |
| `run.waiting_client` | 等待审批或客户端工具 | — |
| `run.completed` / `run.failed` / `run.cancelled` | 终态 | `run.failed` 带 `code` |

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

### 结果

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

### 取消

```python
run.cancel()
```

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

### 幂等与重试

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

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

```python
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 **不占用**服务端的执行资源，可以一直等到请求过期。

```python
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 按服务端策略结束。

### 客户端工具

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

```python
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)
```

| 响应 | 含义 |
| --- | --- |
| `202` | 已接受；内容相同的重复提交返回首次的结果 |
| `409 tool_result_conflict` | 同一请求重复提交了不同内容 |
| `410 tool_request_gone` | 请求已过期或 Run 已取消 |
| `403 wrong_device` | 设备与创建 Run 时声明的不一致 |

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

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

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

### 删除用户数据

```python
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`：

| 类型 | 含义 |
| --- | --- |
| `StudioApiError` | 服务端返回的错误，带 `status`、`code`、`message`、`retryable` |
| `StudioConnectionError` | 网络错误或超时，已用完重试次数 |
| `StudioStreamGapError` | 事件流 `seq` 不连续 |
| `StudioFeatureUnavailableError` | 所连接的 Studio 实例不支持该能力 |
| `StudioCompatibilityError` | 服务端事件 Schema 主版本与 SDK 不一致 |
| `StudioProtocolError` | 响应不符合接口契约 |

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

常见错误码：

| 状态 | code | 处理建议 |
| --- | --- | --- |
| 400 | `invalid_request` | 检查参数，例如 `input` 为空或超过 20000 字 |
| 400 | `actor_required` | `actor_ref` 格式不合法 |
| 401 | `invalid_project_key` | Key 错误、已吊销或已过期 |
| 403 | `forbidden` | Key 缺少所需 scope |
| 403 | `resource_access_denied` | Agent 使用的资源未授权给该项目 |
| 404 | `agent_not_published` | Agent 不存在或尚未发布 |
| 404 | `thread_not_found` / `run_not_found` | 资源不存在，或属于其他 actor |
| 409 | `thread_busy` | 该 Thread 上一个 Run 还没结束 |
| 409 | `idempotency_conflict` | 同一个幂等键配了不同的请求内容 |
| 429 | `stream_limit` | 服务端 SSE 连接数已满，SDK 会退避重连 |

```python
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
```

## 配置

```python
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 不会替你关闭它：

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

## 版本兼容

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

| SDK | Studio 服务端 | 说明 |
| --- | --- | --- |
| 0.x / 1.x | 1.0（当前） | 用 Agent ID 调用 `production` 版本、事件流、取消、审批、客户端工具 |
| 0.x / 1.x | 1.1+（规划中） | 额外支持 slug 寻址（`agent("notes-assistant")`）与版本固定（`agent(id, version=3)`） |
| 任意 | 2.x | 抛出 `StudioCompatibilityError`，需要升级 SDK |

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

## 开发

```bash
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：待确定许可证。
