版本要求：本文档针对 CodeBuddy Agent SDK v0.1.0 及以上版本。
本文档提供 Python SDK 的完整 API 参考。有关快速入门和使用示例，请参阅 SDK 概览。
依赖 | 版本要求 |
Python | >= 3.10 |
CodeBuddy CLI | 已安装 |
异步运行时：SDK 基于 asyncio，所有 API 都是异步的。
推荐使用 uv 进行依赖管理：
uv add codebuddy-agent-sdk
或使用 pip：
pip install codebuddy-agent-sdk
变量名 | 说明 | 必需 |
| CodeBuddy CLI 可执行文件路径 | 可选 |
如果未设置，SDK 会按以下顺序查找 CLI：
环境变量 CODEBUDDY_CODE_PATH
SDK 包内置的二进制文件
开发环境 monorepo 路径
SDK 支持使用已有登录凭据、API Key 或 OAuth Client Credentials 认证，详见 SDK 概览 - 认证配置。
主要 API 入口，创建一个查询并返回消息异步迭代器。
async def query( *, prompt: str | AsyncIterable[dict[str, Any]], options: CodeBuddyAgentOptions | None = None, transport: Transport | None = None,) -> AsyncIterator[Message]:
参数：
参数 | 类型 | 说明 |
|
| 查询提示词或用户消息流 |
|
| 配置选项（可选） |
|
| 自定义传输层（可选） |
返回值：AsyncIterator[Message] - 消息异步迭代器
示例：
from codebuddy_agent_sdk import query, AssistantMessage, TextBlock
async for message in query(prompt="What is 2+2?"):
if isinstance(message, AssistantMessage):
for block in message.content:
if isinstance(block, TextBlock):
print(block.text)
用于双向交互式对话的客户端类。支持多轮对话、中断和动态控制。
class CodeBuddySDKClient: def __init__( self, options: CodeBuddyAgentOptions | None = None, transport: Transport | None = None, ): ...
方法：
连接到 CodeBuddy。
async def connect( self, prompt: str | AsyncIterable[dict[str, Any]] | None = None) -> None:
发送用户消息。
async def query( self, prompt: str | AsyncIterable[dict[str, Any]], session_id: str = "default",) -> None:
接收消息直到收到 ResultMessage。
async def receive_response(self) -> AsyncIterator[Message]:
接收所有消息（不会自动停止）。
async def receive_messages(self) -> AsyncIterator[Message]:
断开连接。
async def disconnect(self) -> None:
上下文管理器支持：
async with CodeBuddySDKClient() as client: await client.query("Hello!") async for msg in client.receive_response(): print(msg)
获取 MCP 服务器连接状态。
async def mcp_server_status(self) -> list[McpServerStatus]:
SDK 提供独立的认证 API，采用 two-phase 设计：先获取登录 URL，再等待用户完成认证。
启动认证流程，返回 AuthFlow 对象。
async def authenticate( *, method_id: str = "external", environment: str | None = None, endpoint: str | None = None, codebuddy_code_path: str | None = None, env: dict[str, str] | None = None, timeout: float = 300.0,) -> AuthFlow:
参数：
参数 | 类型 | 说明 |
|
| 认证方法标识（默认 |
|
| 预定义环境名 |
|
| 自定义端点 URL（与 environment 互斥） |
|
| CLI 可执行文件路径 |
|
| 额外环境变量 |
|
| 用户完成登录的超时时间（秒，默认 300） |
返回值：AuthFlow — 携带登录 URL 的可等待对象
示例：
from codebuddy_agent_sdk import authenticate
# Two-phase: 获取 URL → 展示给用户 → 等待完成
auth = await authenticate()
if auth.auth_url:
print(f"请访问: {auth.auth_url}")
result = await auth
print(f"欢迎, {result.userinfo.user_name}")
# 已登录时 auth.auth_url 为空，await 立即返回
auth = await authenticate()
result = await auth # 已登录则立即返回
# 自定义超时
auth = await authenticate()
result = await auth.wait(timeout=60)
认证流程对象，由 authenticate() 返回。实现了 __await__ 协议，可直接 await。
属性：
属性 | 类型 | 说明 |
|
| 登录 URL（已登录时为空字符串） |
|
| 认证方法标识 |
方法：
等待用户完成认证。
async def wait(self, timeout: float | None = None) -> AuthenticateResponse:
取消认证流程并释放资源。
async def cancel(self) -> None:
登出并清除缓存的认证令牌。
async def logout( *, environment: str | None = None, endpoint: str | None = None, codebuddy_code_path: str | None = None, env: dict[str, str] | None = None,) -> None:
示例：
from codebuddy_agent_sdk import logout
await logout()
警告：
以下 API 处于实验阶段，接口可能在未来版本中变更。
发送中断信号。
async def interrupt(self) -> None:
动态修改权限模式。
async def set_permission_mode(self, mode: str) -> None:
动态修改模型。
async def set_model(self, model: str | None = None) -> None:
完整配置选项：
@dataclassclass CodeBuddyAgentOptions: allowed_tools: list[str] = field(default_factory=list) disallowed_tools: list[str] = field(default_factory=list) system_prompt: str | AppendSystemPrompt | None = None mcp_servers: dict[str, McpServerConfig] | str | Path = field(default_factory=dict) permission_mode: PermissionMode | None = None continue_conversation: bool = False resume: str | None = None max_turns: int | None = None model: str | None = None fallback_model: str | None = None cwd: str | Path | None = None codebuddy_code_path: str | Path | None = None env: dict[str, str] = field(default_factory=dict) extra_args: dict[str, str | None] = field(default_factory=dict) stderr: Callable[[str], None] | None = None hooks: dict[HookEvent, list[HookMatcher]] | None = None include_partial_messages: bool = False fork_session: bool = False persist_session: bool = True agents: dict[str, AgentDefinition] | None = None setting_sources: list[SettingSource] | None = None can_use_tool: CanUseTool | None = None
字段 | 类型 | 说明 |
|
| 自动允许的工具白名单 |
|
| 禁止使用的工具黑名单 |
|
| 系统提示词配置 |
|
| MCP 服务器配置 |
|
| 权限模式 |
|
| 继续最近的会话 |
|
| 要恢复的会话 ID |
|
| 最大对话轮数 |
|
| 指定模型 |
|
| 备用模型 |
|
| 工作目录 |
|
| CLI 可执行文件路径 |
|
| 环境变量 |
|
| 额外的 CLI 参数 |
|
| stderr 回调 |
|
| Hook 配置 |
|
| 包含部分消息 |
|
| 分叉会话 |
|
| 是否持久化会话记录，默认 |
|
| 自定义 Agent |
|
| 设置来源 |
|
| 权限回调函数 |
|
| 最大思考 token 数（已废弃，请使用 |
|
| 思考模式配置： |
|
| 模型推理努力程度 |
PermissionMode = Literal["default", "acceptEdits", "plan", "bypassPermissions"]
值 | 说明 |
| 默认模式，所有操作需确认 |
| 自动批准文件编辑 |
| 规划模式，仅允许读取 |
| 跳过所有权限检查 |
PermissionResult = PermissionResultAllow | PermissionResultDeny
@dataclass
class PermissionResultAllow:
updated_input: dict[str, Any]
behavior: Literal["allow"] = "allow"
updated_permissions: list[dict[str, Any]] | None = None
@dataclass
class PermissionResultDeny:
message: str
behavior: Literal["deny"] = "deny"
interrupt: bool = False
CanUseTool = Callable[
[str, dict[str, Any], CanUseToolOptions],
Awaitable[PermissionResult],
]
@dataclass
class CanUseToolOptions:
tool_use_id: str
signal: Any | None = None
agent_id: str | None = None
suggestions: list[dict[str, Any]] | None = None
blocked_path: str | None = None
decision_reason: str | None = None
@dataclassclass AgentDefinition: description: str # Agent 描述 prompt: str # 系统提示词 tools: list[str] | None = None # 允许的工具 disallowed_tools: list[str] | None = None # 禁止的工具 model: str | None = None # 使用的模型
class McpStdioServerConfig(TypedDict):
type: NotRequired[Literal["stdio"]]
command: str
args: NotRequired[list[str]]
env: NotRequired[dict[str, str]]
McpServerConfig = McpStdioServerConfig
HookEvent = ( Literal["PreToolUse"] | Literal["PostToolUse"] | Literal["UserPromptSubmit"] | Literal["Stop"] | Literal["SubagentStop"] | Literal["PreCompact"] | Literal["WorktreeCreate"] | Literal["WorktreeRemove"])
@dataclassclass HookMatcher: matcher: str | None = None # 匹配模式（支持正则） hooks: list[HookCallback] = field(default_factory=list) timeout: float | None = None # 超时时间（秒）
HookCallback = Callable[
[Any, str | None, HookContext],
Awaitable[HookJSONOutput],
]
class HookContext(TypedDict):
signal: Any | None
class SyncHookJSONOutput(TypedDict):
continue_: NotRequired[bool]
suppressOutput: NotRequired[bool]
stopReason: NotRequired[str]
decision: NotRequired[Literal["block"]]
reason: NotRequired[str]
控制 SDK 从哪些文件系统位置加载配置。
SettingSource = Literal["user", "project", "local"]
值 | 说明 | 位置 |
| 全局用户设置 |
|
| 项目共享设置 |
|
| 项目本地设置 |
|
默认行为：当 setting_sources 未指定时，SDK 不加载任何文件系统配置。这提供了完全干净的运行环境。
# 默认：不加载任何配置（干净环境）
async for msg in query(prompt="..."):
pass
# 加载项目配置
options = CodeBuddyAgentOptions(setting_sources=["project"])
# 加载所有配置（类似 CLI 行为）
options = CodeBuddyAgentOptions(setting_sources=["user", "project", "local"])
@dataclassclass AppendSystemPrompt: append: str # 追加到默认系统提示词的内容
所有消息类型的联合：
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent
说明：
TaskStartedMessage / TaskNotificationMessage 是 SystemMessage 的子类，已被联合覆盖（isinstance / case SystemMessage() 继续匹配），并单独导出以便在调用处做精确类型判断。
@dataclassclass SystemMessage: subtype: str data: dict[str, Any]
后台任务（Bash / PowerShell / Workflow / Agent，run_in_background: true）进入运行态时发出。是 SystemMessage 的子类，isinstance(msg, SystemMessage) 仍然成立。
@dataclassclass TaskStartedMessage(SystemMessage): task_id: str description: str uuid: str session_id: str tool_use_id: str | None = None task_type: str | None = None # "Bash" / "PowerShell" / "Workflow" / "Agent"
task_progress / task_notification 携带的用量统计（对齐 Claude Code 的 TaskUsage）。sub-agent（task_type == "Agent"）后台任务有值；后台 shell 任务通常省略。
class TaskUsage(TypedDict): total_tokens: int tool_uses: int duration_ms: int
后台任务进度事件。事件驱动（非定时）：每完成一次 tool_use 推一条，携带累计 usage 与最近工具名 last_tool_name。后台 shell 任务不发 progress。
@dataclassclass TaskProgressMessage(SystemMessage): task_id: str description: str usage: TaskUsage uuid: str session_id: str tool_use_id: str | None = None last_tool_name: str | None = None
后台任务状态变迁事件。patch 携带本次变更字段（至少 status，终态补 end_time）。
生命周期提示：后台任务的终态有时只来 task_updated（patch.status 为终态）而没有配套的 task_notification。跟踪"活跃任务"的消费方应对二者的终态 status 一视同仁——用 TERMINAL_TASK_STATUSES frozenset 判定（{"completed", "failed", "stopped", "killed"}）。
@dataclassclass TaskUpdatedMessage(SystemMessage): task_id: str patch: dict[str, Any] status: TaskUpdatedStatus | None = None # pending/running/paused/completed/failed/killed session_id: str | None = None uuid: str | None = None
后台任务完成 / 失败 / 被停止时发出。在 stdio stream-json 长连接模式下，任务若在触发它的那轮 result 之后才完成，该消息会被主动推回同一输出流——消费方需用 receive_messages() 持续读取才能收到（query() 与 receive_response() 在首个 ResultMessage 处停止，会错过后台完成事件）。usage 在 sub-agent 任务上携带，shell 任务省略。
query() 已自动禁用后台任务：由于 query() 在首个 ResultMessage 处停止并关闭子进程、无法接收跨轮回推事件，SDK 在 query() 路径下会自动注入 CODEBUDDY_CODE_DISABLE_BACKGROUND_TASKS=1（Bash / PowerShell / Agent 的 run_in_background 被隐藏/降级为前台）。若确需在 query() 下保留后台任务，显式在 options.env 或进程环境里设置该变量（任意值，包括 "0"）即可覆盖。CodeBuddySDKClient + receive_messages() 路径不受影响。
@dataclassclass TaskNotificationMessage(SystemMessage): task_id: str status: Literal["completed", "failed", "stopped"] summary: str uuid: str session_id: str tool_use_id: str | None = None output_file: str | None = None # 后台任务 stdout 落盘路径（文件模式） output_stderr_file: str | None = None usage: TaskUsage | None = None
多个并发后台任务靠 task_id 区分；tool_use_id 关联回模型发起该任务的 tool_use。示例（持续读取以接收整体完成后的通知）：
from codebuddy_agent_sdk import (
CodeBuddySDKClient,
CodeBuddyAgentOptions,
TaskStartedMessage,
TaskNotificationMessage,
)
async def run_background_tasks():
options = CodeBuddyAgentOptions(permission_mode="bypassPermissions")
async with CodeBuddySDKClient(options=options) as client:
await client.query("并行跑两个后台命令，完成后告诉我结果")
# 用 receive_messages 持续读——不要用 receive_response（它在首个 result 处停）
async for msg in client.receive_messages():
if isinstance(msg, TaskStartedMessage):
print(f"[started] {msg.task_id} ({msg.task_type}): {msg.description}")
elif isinstance(msg, TaskNotificationMessage):
print(f"[done] {msg.task_id} status={msg.status} output={msg.output_file}")
# 收齐关心的任务后自行 break
@dataclassclass UserMessage: content: str | list[ContentBlock] uuid: str | None = None parent_tool_use_id: str | None = None
@dataclassclass AssistantMessage: content: list[ContentBlock] model: str parent_tool_use_id: str | None = None error: str | None = None
@dataclassclass ResultMessage: subtype: str duration_ms: int duration_api_ms: int is_error: bool num_turns: int session_id: str total_cost_usd: float | None = None usage: dict[str, Any] | None = None result: str | None = None errors: list[str] | None = None # Structured error info aligned with `errors` by index. # - Length matches `errors` when present; entry is None if no structured dimension available. # - Field is absent entirely when every entry would be None (backward compatible). # - Each entry may carry: status (HTTP), code (SDK/business), category (network/quota/auth/model_service/...), details (message). errors_info: list[dict[str, Any] | None] | None = None
@dataclassclass StreamEvent: uuid: str session_id: str event: dict[str, Any] parent_tool_use_id: str | None = None
ContentBlock = TextBlock | ThinkingBlock | ToolUseBlock | ToolResultBlock
@dataclass
class TextBlock:
text: str
@dataclass
class ThinkingBlock:
thinking: str
signature: str
@dataclass
class ToolUseBlock:
id: str
name: str
input: dict[str, Any]
@dataclass
class ToolResultBlock:
tool_use_id: str
content: str | list[dict[str, Any]] | None = None
is_error: bool | None = None
@dataclassclass AskUserQuestionInput: questions: list[AskUserQuestionQuestion] answers: dict[str, str] | None = None
@dataclassclass AskUserQuestionQuestion: question: str # 完整问题文本（应以 ? 结尾） header: str # 简短标签（最多 12 个字符） options: list[AskUserQuestionOption] multi_select: bool # 是否允许多选
@dataclassclass AskUserQuestionOption: label: str # 显示文本（1-5 个单词） description: str # 选项说明
所有异常都继承自 CodeBuddySDKError。
class CodeBuddySDKError(Exception): """Base exception for CodeBuddy SDK errors.""" pass
当连接到 CLI 失败或未建立连接时抛出。
class CLIConnectionError(CodeBuddySDKError): pass
当找不到 CLI 可执行文件时抛出。
class CLINotFoundError(CodeBuddySDKError): def __init__( self, message: str, platform: str | None = None, arch: str | None = None, ): ...
属性：
属性 | 类型 | 说明 |
|
| 当前平台 |
|
| 当前架构 |
当 CLI 输出的 JSON 解码失败时抛出。
class CLIJSONDecodeError(CodeBuddySDKError): pass
当 CLI 进程遇到错误时抛出。
class ProcessError(CodeBuddySDKError): pass
当 CLI 进程在启动阶段崩溃或未产生任何输出时抛出。
class CLIStartupError(CodeBuddySDKError): def __init__( self, message: str, stderr: str = "", exit_code: int | None = None, ): ...
属性：
属性 | 类型 | 说明 |
|
| CLI 进程的 stderr 输出 |
|
| 进程退出码 |
当执行失败时抛出（如认证错误、API 错误）。包含 ResultMessage 中的 errors 数组。
class ExecutionError(CodeBuddySDKError): def __init__(self, errors: list[str], subtype: str): ...
属性：
属性 | 类型 | 说明 |
|
| 错误消息列表 |
|
| 错误子类型 |
当认证失败时抛出。
class AuthenticationError(CodeBuddySDKError): def __init__(self, error_type: str, message: str): ...
属性：
属性 | 类型 | 说明 |
|
| 错误类型（如 |
@dataclass(slots=True)class AuthenticateResponse: userinfo: UserInfo
@dataclass(slots=True)class UserInfo: user_id: str user_name: str = "" user_nickname: str = "" token: str = "" enterprise_id: str | None = None enterprise: str | None = None
@dataclass(slots=True)class McpServerStatus: name: str status: Literal["connected", "failed", "needs-auth", "pending"] server_info: dict[str, Any] | None = None
SDK 概览 - 快速入门和使用示例
TypeScript SDK 参考 - TypeScript 版本 API
Hook 参考指南 - 详细的 Hook 配置说明
MCP 集成 - MCP 服务器配置指南
