Python SDK

本页说明如何安装、配置并使用官方 web-agent-sdk Python 包：起 session、跑 run、订阅事件。

pip install web-agent-sdk

需要 Python 3.10+。SDK 是 async-first（asyncio / anyio）。

一个入口：Client

用户面 API 都挂在同一个 Client 上：

from web_agent.v1 import Client

构造 Client 需要 api_key 和 project_id。两者都可省略——省略时分别自动读环境变量 $WEBAGENT_API_KEY / $WEBAGENT_PROJECT_ID。

资源	对应产品	用法
client.sessions / messages / events	DoAnything（开放型）	自由输入，agent 自己定路径
client.deep_research	DeepResearch（定型 — 研究→报告）	Standalone API
client.web_search	WebSearch（查询 → 结果）	默认 wait=true，等结果就绪再返回
client.track	Track（监控 → snapshot）	长寿命 monitor + webhook 通道

包名是 web-agent-sdk（带连字符），但 import 名是 web_agent——同 python-dateutil → dateutil 的惯例。

DoAnything —— 开放型任务

import asyncio
from web_agent.v1 import Client
from web_agent.v1.types import CreateSessionRequest

async def main():
    async with Client(
        api_key="wa_demo_xxxxxxxxxxxxxxxx",
        project_id="proj_demo_0001",
    ) as client:
        session = await client.sessions.create(CreateSessionRequest(
            instructions="找出 Hacker News 现在 Top 5 的故事，列表返回。",
        ))
        run = session.runs[0]   # session 创建会隐式排好首个 run
        async for event in client.events.stream(session.id, run.id):
            print(event.type, event.data)
            if event.type == "run.completed":
                break

asyncio.run(main())

「起新 run」 vs 「往运行中 run 塞消息」

# 1. 往当前 run 的对话队列里塞一条消息
#    （agent 在下个 ReAct 边界自己决定要不要消费）
await client.messages.send(
    session.id, run.id,
    content="顺便把每条的评论数也带上。",
)

# 2. 在同一个 session 里起一个新 run
#    （复用 browser / profile / workspace；上一个 run 必须已终态）
from web_agent.v1.types import CreateRunRequest
new_run = await client.sessions.create_run(
    session.id,
    CreateRunRequest(instructions="点进第一条，总结评论区。"),
)

回应 input request

await client.messages.intervene(
    session.id, run.id,
    input_request_id="ir_01HXX",
    response={"solved": True},
)

取消 / 停止 / 列表

await client.sessions.cancel_run(session.id, run.id, reason="user_cancelled")
await client.sessions.stop(session.id, force=False)            # 软停 session
listing = await client.sessions.list(status="running", limit=20)
for s in listing.items:
    print(s.id, s.status)

心跳事件

stream() 默认过滤心跳事件；做连接健康 UI 时传 include_heartbeats=True。续传支持 Last-Event-ID：

client.events.stream(session.id, run.id, last_event_id="142")

DeepResearch —— 研究 → 报告

DeepResearch 是 standalone API——路径不带 project 段（/v1/deep_research），项目租户从 Bearer token 解析。

async with Client(api_key="wa_...", project_id="proj_demo") as client:
    run = await client.deep_research.run(
        topic="2026 年开源向量数据库格局",
        depth="deep",                       # light / standard / deep
        require_outline_approval=True,      # 默认开 outline HITL gate
    )
    print(run["run_id"], run["status"])

订阅事件流（DR 走的是 DoAnything 的 SSE 通道）+ 响应 outline gate：

# run["run_id"] / run["session_id"] 都在创建响应里
async for event in client.events.stream(
    run["session_id"], run["run_id"],
):
    if event.type == "run.input_request":
        # outline 已生成，请求人工批准
        await client.deep_research.intervene(
            run["run_id"],
            request_id=event.data["request_id"],
            response="approve",  # 或 {"action": "approve_with_edits", "edits": [...]}
        )
    if event.type == "run.completed":
        break

# 拉取产物三件套（final.md / citations.json / confidence.json）
artifacts = await client.deep_research.list_artifacts(run["run_id"])
final = await client.deep_research.get_artifact(
    run["run_id"], artifacts[0]["id"],
)

WebSearch —— 查询 → 结果

WebSearch 是 project-scoped API。run() 默认 wait=true：服务器最多阻塞 30s，等搜索完成后直接返回完整结果；超过 30s 则返回 202，再调 get(run_id) 轮询。

# 同步形态（默认）
result = await client.web_search.run(
    queries=["best Python ORM 2026"],
    engines=["tavily"],
    summarize=True,
)
for hit in result["results"]["results"]:
    print(hit["title"], hit["url"])

# 异步形态
pending = await client.web_search.run_async(queries=["best Python ORM 2026"])
detail = await client.web_search.get(pending["run_id"])

# Refine（同 run 内重搜）
await client.web_search.refine(
    pending["run_id"],
    text="再加 site:reddit.com 限制重搜",
)

Track —— 长寿命监控

Track 会创建一个长期运行的 monitor。它按 schedule（cron / interval / event）反复检查目标页面，每次检查保存一份 snapshot；当变化满足触发条件时，通过你配置的通道（如 webhook）通知你。

mon = await client.track.create(
    intent="apple.com 上 iPhone 17 Pro 跌破 $999 时通知我",
    schedule={"kind": "interval", "interval_seconds": 3600},
    notify_channel={"kind": "callback_url", "url": "https://hooks.example.com/track"},
)

# 生命周期 —— 通过 patch 暂停 / 恢复 / refine：
await client.track.pause(mon["id"], reason="人工核对")
await client.track.resume(mon["id"])
await client.track.refine(mon["id"], trigger_dsl={"op": "lt", "field": "price", "value": 999})

# 手动触发一次检查（绕开 schedule）；返回这一次检查的结果：
outcome = await client.track.run_now(mon["id"])

# 拉 snapshot 历史（最新优先）：
snapshots = await client.track.list_snapshots(mon["id"])
snap = await client.track.get_snapshot(mon["id"], snapshots["items"][0]["id"])

# webhook 投递历史 + 重投失败的投递：
deliveries = await client.track.list_deliveries(mon["id"], include_payload=True)
await client.track.retry_delivery(mon["id"], deliveries["items"][0]["id"])

# 取消 monitor（终态）：
await client.track.cancel(mon["id"])  # 等价 await client.track.delete(mon["id"])

Alignment HITL（可选）

Supervisor 需要让你澄清意图（如 "你说的是 SKU A 还是 SKU B?"）时，monitor 进入 pending_clarification，发出 alignment.input_request 事件。用 intervene() 回答：

await client.track.intervene(
    mon["id"],
    request_id="req_align_1",
    response="SKU A",
)

也可以随时往 alignment 队列推自由文本：client.track.message(mon_id, content="…")。

错误

SDK 用 typed exception，按类 catch：

from web_agent.v1 import (
    UnauthorizedError, InsufficientCreditsError, RateLimitedError,
)

try:
    await client.sessions.create(CreateSessionRequest(instructions="…"))
except InsufficientCreditsError as e:
    print("top up:", e.detail, e.extra)

每个 exception 都继承 ApiError，带 code / detail / extra，与 API 错误信封 对应。

异常类	HTTP	code
UnauthorizedError	401	unauthorized
ForbiddenError	403	forbidden、safety_boundary_violated
NotFoundError	404	*_not_found
ConflictError	409	conflict
ValidationError	422	validation_error
RateLimitedError	429	rate_limit_exceeded
InsufficientCreditsError	402	insufficient_credits
BudgetExceededError	402	budget_exceeded

类型

Session、Run、Event 等 DoAnything 资源都是 dataclass，从 web_agent.v1 重新导出：

from web_agent.v1 import Session, Run, Event, RunStatus

DR / DS / WS 的响应直接是 dict[str, Any]（按 OpenAPI 信封原样回传），用键名访问即可（run["run_id"] / run["status"]）。mypy --strict 通过。

接下来

• TypeScript SDK —— JS / TS 同样的接口
• 错误码与重试 —— 推荐重试策略 + 幂等键
• DoAnything —— 生命周期 / profile / workspace