codebuddy-cloud-agent-sdk 0.3.1.dev202605050405

CodeBuddy Cloud Agent SDK (Python) — 基于 ACP 的云端沙箱 Agent 客户端

codebuddy-cloud-agent-sdk (Python)

Python 版 Cloud Agent SDK — 与 @tencent-ai/cloud-agent-sdk（Node 版）功能等价。

功能清单 / 变更历史见 CHANGELOG.md 和 IMPLEMENTATION.md。

特性

• 🎯 与 Node 版严格对齐：REST 控制面 + ACP 订阅模型 + 21 条严格对齐要点
• 🐍 Pythonic：用 asyncio.Event / logging.Logger / task.cancel() 原生原语，不造 CancelToken、LeveledLogger 等二次封装
• 🔌 不重复发明：JSON-RPC 层 100% 复用官方 agent-client-protocol SDK（Pydantic schema + ClientSideConnection）
• 🚀 OpenTelemetry 友好：三个挂载点（http_client / headers / on_request/on_response），业务接 OTel 零侵入
• ⚡ 异步原生：基于 httpx.AsyncClient + asyncio
• 📋 类型完整：py.typed 标记 + Pydantic v2 + mypy strict 通过

安装

# 用 pip
pip install codebuddy-cloud-agent-sdk

# 用 uv（推荐）
uv add codebuddy-cloud-agent-sdk

运行时依赖：httpx + pydantic + agent-client-protocol（自动装）。

Python 3.10+（已验证 3.10 / 3.11 / 3.12 / 3.13 / 3.14）。

抢先体验 nightly 版本（对应 Node 版的 @next tag）：

pip install --pre codebuddy-cloud-agent-sdk
# 或锁定具体 dev 版本：
pip install codebuddy-cloud-agent-sdk==0.3.1.dev202605030807

Nightly 版本号格式 X.Y.Z.devYYYYMMDDHHMM（UTC 分钟级）—— 符合 PEP 440 dev release 规范。pip install 默认只装正式版，只有显式 --pre 才会装 nightly。

4 行代码到 prompt

import asyncio
from cloud_agent_sdk import CloudAgentClient

async def main():
    client = CloudAgentClient(api_key="ck_xxx")
    rt = await client.runtimes.create(runtime_name="demo")
    session = rt.sessions.default()
    response = await session.prompt("hello")
    print(response.stop_reason)

asyncio.run(main())

流式消费（订阅模型）

SDK 用 ACP 原生的两通道模型：

• prompt() 返回一轮终局（PromptResponse 含 stop_reason）
• subscribe() 注册监听器接收流式 notifications（session/update）

from cloud_agent_sdk import CloudAgentClient
from acp.schema import SessionNotification   # 直接用官方 SDK 的类型

async def main():
    client = CloudAgentClient(api_key="ck_xxx")
    rt = await client.runtimes.create(runtime_name="demo")
    session = rt.sessions.default()
    await session.connect()

    def on_chunk(n: SessionNotification) -> None:
        update = n.update
        if update.session_update == "agent_message_chunk":
            content = update.content
            if content.type == "text":
                print(content.text, end="", flush=True)

    unsub = session.subscribe(on_chunk)
    try:
        response = await session.prompt("write a poem")
        print(f"\n[done] stop_reason={response.stop_reason}")
    finally:
        unsub()
        await session.disconnect()

asyncio.run(main())

便利钩子：只关心本次 prompt 的流式内容时用 on_chunk，自动管理订阅生命周期：

from cloud_agent_sdk import PromptOptions

response = await session.prompt(
    "write a poem",
    PromptOptions(on_chunk=lambda n: print(n)),  # prompt 结束自动 unsub
)

取消 & 超时

import asyncio

# 方式 1：直接 task.cancel()（最简单，Pythonic）
task = asyncio.create_task(session.prompt("long task"))
await asyncio.sleep(5)
task.cancel()  # SDK 自动向服务端发 session/cancel 并传播 CancelledError

# 方式 2：用 asyncio.timeout（Python 3.11+）
async with asyncio.timeout(10):
    await session.prompt("long task")

# 方式 3：用 PromptOptions.cancel event（外部事件触发）
from cloud_agent_sdk import PromptOptions

cancel = asyncio.Event()
# ... 别处 cancel.set()
await session.prompt("long task", PromptOptions(cancel=cancel))

# 方式 4：用 PromptOptions.timeout_ms 硬超时
await session.prompt("long task", PromptOptions(timeout_ms=10_000))

日志接入

按 Python 社区标准做法（对齐 urllib3 / httpx / openai）：SDK 只发日志，级别和输出 100% 由用户配置。SDK 永远不修改你的 logger 的级别、handler 或 formatter。

import logging

# 方式 1：用 SDK 默认 logger（不传 logger 参数）
logging.getLogger("cloud_agent_sdk").setLevel(logging.DEBUG)
logging.basicConfig()
CloudAgentClient(api_key="...")

# 方式 2：接入自己的 logging 基础设施（OTel Logs Bridge / JSON 格式器等）
my_logger = logging.getLogger("myapp.agent")
my_logger.setLevel(logging.DEBUG)
my_logger.addHandler(my_otel_handler)
CloudAgentClient(api_key="...", logger=my_logger)

# 方式 3：loguru / structlog
from loguru import logger
logger.add(sys.stderr, level="DEBUG")
CloudAgentClient(api_key="...", logger=logger)

OpenTelemetry 接入

SDK 不引入 opentelemetry 依赖，通过三个挂载点让业务零侵入接入：

import httpx
from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
from cloud_agent_sdk import CloudAgentClient

# 1. instrument 自己的 httpx client，注入 SDK 让控制面 + 数据面 HTTP span 自动产生
http = httpx.AsyncClient()
HTTPXClientInstrumentor().instrument_client(http)

client = CloudAgentClient(api_key="...", http_client=http)
# 所有 REST + ACP SSE/POST 都走这个 client，自动注入 traceparent、产生 span

其他场景（自定义 prompt span / RED metrics / traceparent 透传）见 IMPLEMENTATION.md § 9。

Manifest 构造

两种方式等价：

from cloud_agent_sdk import (
    AgentManifest, ManifestBuilder, ManifestSecret, ManifestEnv,
)

# 方式 A：ManifestBuilder fluent API
manifest = (
    ManifestBuilder()
        .id("my-agent")
        .name("My Agent")
        .version("1.0")
        .system_prompt("You are a helpful assistant.")
        .skills("code-reviewer", "test-writer")
        .secrets(ManifestSecret(key="OPENAI_API_KEY", value="..."))
        .envs(ManifestEnv(key="LOG_LEVEL", value="info"))
        .build()
)

# 方式 B：Pydantic 构造器
manifest = AgentManifest(
    id="my-agent",
    name="My Agent",
    manifest_version="1.0",
    system_prompt="You are a helpful assistant.",
    secrets=[ManifestSecret(key="OPENAI_API_KEY", value="...")],
)

# 用它创建 runtime
rt = await client.runtimes.create(
    RuntimeCreateOptions(runtime_name="demo", agent_manifest=manifest)
)

密钥脱敏：SDK 日志和 on_request 钩子里会把 agent_manifest.secrets[].value 自动替换成 '***'（真实发送给服务端的 body 不受影响）。

多模态输入

from acp.schema import TextContentBlock, ImageContentBlock

# 字符串会被自动包装成 TextContentBlock
response = await session.prompt("hello")

# 多模态直接传 block 列表
response = await session.prompt([
    TextContentBlock(type="text", text="What's in this image?"),
    ImageContentBlock(type="image", data="base64...", mime_type="image/png"),
])

与 Node 版对齐

所有公共 API 与 @tencent-ai/cloud-agent-sdk 语义一致：

项	Node	Python
命名	camelCase	snake_case（Pydantic 透明 alias 回 camelCase 契约）
取消	AbortSignal	asyncio.CancelledError + asyncio.Event
日志	Logger interface + LeveledLogger	logging.Logger（标准 Python）
错误类	TimeoutError 等	APITimeoutError 等（API 前缀避免遮蔽内建名，对齐 OpenAI / Anthropic SDK）
DEFAULT_RETRY_ON	全大写	default_retry_on（Python snake_case）

详细差异 + 严格对齐要点见 IMPLEMENTATION.md § 7-8。

开发

# 安装开发依赖
pip install -e '.[dev]'

# 代码检查
ruff check src examples
mypy src/

# 跑测试（91 用例，mock 场景，~4s）
pytest examples/ --timeout=15

# 跑真实后端测试（需要 CLOUD_AGENT_API_KEY）
CLOUD_AGENT_API_KEY=ck_xxx python examples/test_smoke.py

文档

• IMPLEMENTATION.md — 完整设计 + Phase 路线图 + 验证矩阵
• Node 版源码：packages/cloud-agent-sdk — 严格对齐基准