Metadata-Version: 2.4
Name: codebuddy-cloud-agent-sdk
Version: 0.3.3.dev202605140414
Summary: CodeBuddy Cloud Agent SDK (Python) — 基于 ACP 的云端沙箱 Agent 客户端
Project-URL: Homepage, https://cnb.cool/codebuddy/codebuddy-code
Project-URL: Repository, https://cnb.cool/codebuddy/codebuddy-code
Project-URL: Issues, https://cnb.cool/codebuddy/codebuddy-code/-/issues
Project-URL: Changelog, https://cnb.cool/codebuddy/codebuddy-code/-/blob/main/packages/cloud-agent-sdk-python/CHANGELOG.md
Author-email: Tencent CodeBuddy <codebuddy@tencent.com>
License: MIT
License-File: LICENSE
Keywords: acp,agent,cloud-agent,codebuddy,llm,sandbox
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: <3.15,>=3.10
Requires-Dist: agent-client-protocol<1.0,>=0.9
Requires-Dist: httpx>=0.27
Requires-Dist: pydantic>=2.6
Provides-Extra: dev
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Requires-Dist: respx>=0.21; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Requires-Dist: twine>=5; extra == 'dev'
Requires-Dist: uvicorn>=0.30; extra == 'dev'
Description-Content-Type: text/markdown

# codebuddy-cloud-agent-sdk (Python)

Python 版 Cloud Agent SDK — 与 [`@tencent-ai/cloud-agent-sdk`](https://www.npmjs.com/package/@tencent-ai/cloud-agent-sdk)（Node 版）功能等价。

> 功能清单 / 变更历史见 [`CHANGELOG.md`](./CHANGELOG.md) 和 [`IMPLEMENTATION.md`](./IMPLEMENTATION.md)。

## 特性

- 🎯 **与 Node 版严格对齐**：REST 控制面 + ACP 订阅模型 + 21 条严格对齐要点
- 🐍 **Pythonic**：用 `asyncio.Event` / `logging.Logger` / `task.cancel()` 原生原语，不造 `CancelToken`、`LeveledLogger` 等二次封装
- 🔌 **不重复发明**：JSON-RPC 层 100% 复用官方 [`agent-client-protocol`](https://pypi.org/project/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 通过

## 安装

```bash
# 用 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）：

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

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

```python
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`，自动管理订阅生命周期：

```python
from cloud_agent_sdk import PromptOptions

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

## 取消 & 超时

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

```python
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` 依赖，通过三个挂载点让业务零侵入接入：

```python
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`](./IMPLEMENTATION.md#9-opentelemetry--可观测性接入)。

## Manifest 构造

两种方式等价：

```python
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 不受影响）。

## 多模态输入

```python
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`](https://github.com/Tencent/codebuddy/tree/main/packages/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`](./IMPLEMENTATION.md)。

## 开发

```bash
# 安装开发依赖
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`](./IMPLEMENTATION.md) — 完整设计 + Phase 路线图 + 验证矩阵
- Node 版源码：[`packages/cloud-agent-sdk`](../cloud-agent-sdk/) — 严格对齐基准
