Skip to content

Python SDK 参考

版本要求:本文档针对 CodeBuddy Agent SDK v0.1.0 及以上版本。

本文档提供 Python SDK 的完整 API 参考。有关快速入门和使用示例,请参阅 SDK 概览

Requirements

依赖版本要求
Python>= 3.10
CodeBuddy CLI已安装

异步运行时:SDK 基于 asyncio,所有 API 都是异步的。

Installation

推荐使用 uv 进行依赖管理:

bash
uv add codebuddy-agent-sdk

或使用 pip:

bash
pip install codebuddy-agent-sdk

环境变量

变量名说明必需
CODEBUDDY_CODE_PATHCodeBuddy CLI 可执行文件路径可选

如果未设置,SDK 会按以下顺序查找 CLI:

  1. 环境变量 CODEBUDDY_CODE_PATH
  2. SDK 包内置的二进制文件
  3. 开发环境 monorepo 路径

认证配置

SDK 支持使用已有登录凭据、API Key 或 OAuth Client Credentials 认证,详见 [SDK 概览 - 认证配置](sdk.md#认证配置)。

Functions

query()

主要 API 入口,创建一个查询并返回消息异步迭代器。

python
async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: CodeBuddyAgentOptions | None = None,
    transport: Transport | None = None,
) -> AsyncIterator[Message]:

参数

参数类型说明
promptstr | AsyncIterable[dict]查询提示词或用户消息流
optionsCodeBuddyAgentOptions配置选项(可选)
transportTransport自定义传输层(可选)

返回值AsyncIterator[Message] - 消息异步迭代器

示例

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

Client Class

CodeBuddySDKClient

用于双向交互式对话的客户端类。支持多轮对话、中断和动态控制。

python
class CodeBuddySDKClient:
    def __init__(
        self,
        options: CodeBuddyAgentOptions | None = None,
        transport: Transport | None = None,
    ): ...

方法

connect()

连接到 CodeBuddy。

python
async def connect(
    self,
    prompt: str | AsyncIterable[dict[str, Any]] | None = None
) -> None:

query()

发送用户消息。

python
async def query(
    self,
    prompt: str | AsyncIterable[dict[str, Any]],
    session_id: str = "default",
) -> None:

receive_response()

接收消息直到收到 ResultMessage。

python
async def receive_response(self) -> AsyncIterator[Message]:

receive_messages()

接收所有消息(不会自动停止)。

python
async def receive_messages(self) -> AsyncIterator[Message]:

disconnect()

断开连接。

python
async def disconnect(self) -> None:

上下文管理器支持

python
async with CodeBuddySDKClient() as client:
    await client.query("Hello!")
    async for msg in client.receive_response():
        print(msg)

Unstable API

警告:以下 API 处于实验阶段,接口可能在未来版本中变更。

interrupt()

发送中断信号。

python
async def interrupt(self) -> None:

set_permission_mode()

动态修改权限模式。

python
async def set_permission_mode(self, mode: str) -> None:

set_model()

动态修改模型。

python
async def set_model(self, model: str | None = None) -> None:

Types

CodeBuddyAgentOptions

完整配置选项:

python
@dataclass
class 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
    agents: dict[str, AgentDefinition] | None = None
    setting_sources: list[SettingSource] | None = None
    can_use_tool: CanUseTool | None = None
字段类型说明
allowed_toolslist[str]自动允许的工具白名单
disallowed_toolslist[str]禁止使用的工具黑名单
system_promptstr | AppendSystemPrompt系统提示词配置
mcp_serversdict[str, McpServerConfig]MCP 服务器配置
permission_modePermissionMode权限模式
continue_conversationbool继续最近的会话
resumestr要恢复的会话 ID
max_turnsint最大对话轮数
modelstr指定模型
fallback_modelstr备用模型
cwdstr | Path工作目录
codebuddy_code_pathstr | PathCLI 可执行文件路径
envdict[str, str]环境变量
extra_argsdict[str, str | None]额外的 CLI 参数
stderrCallable[[str], None]stderr 回调
hooksdict[HookEvent, list[HookMatcher]]Hook 配置
include_partial_messagesbool包含部分消息
fork_sessionbool分叉会话
agentsdict[str, AgentDefinition]自定义 Agent
setting_sourceslist[SettingSource]设置来源
can_use_toolCanUseTool权限回调函数

PermissionMode

python
PermissionMode = Literal["default", "acceptEdits", "plan", "bypassPermissions"]
说明
"default"默认模式,所有操作需确认
"acceptEdits"自动批准文件编辑
"plan"规划模式,仅允许读取
"bypassPermissions"跳过所有权限检查

PermissionResult

python
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

python
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

AgentDefinition

python
@dataclass
class AgentDefinition:
    description: str        # Agent 描述
    prompt: str             # 系统提示词
    tools: list[str] | None = None           # 允许的工具
    disallowed_tools: list[str] | None = None  # 禁止的工具
    model: str | None = None                 # 使用的模型

McpServerConfig

python
class McpStdioServerConfig(TypedDict):
    type: NotRequired[Literal["stdio"]]
    command: str
    args: NotRequired[list[str]]
    env: NotRequired[dict[str, str]]

McpServerConfig = McpStdioServerConfig

HookEvent

python
HookEvent = (
    Literal["PreToolUse"]
    | Literal["PostToolUse"]
    | Literal["UserPromptSubmit"]
    | Literal["Stop"]
    | Literal["SubagentStop"]
    | Literal["PreCompact"]
)

HookMatcher

python
@dataclass
class HookMatcher:
    matcher: str | None = None      # 匹配模式(支持正则)
    hooks: list[HookCallback] = field(default_factory=list)
    timeout: float | None = None    # 超时时间(秒)

HookCallback

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

SettingSource

控制 SDK 从哪些文件系统位置加载配置。

python
SettingSource = Literal["user", "project", "local"]
说明位置
"user"全局用户设置~/.codebuddy/settings.json
"project"项目共享设置.codebuddy/settings.json
"local"项目本地设置.codebuddy/settings.local.json

默认行为:当 setting_sources 未指定时,SDK 不加载任何文件系统配置。这提供了完全干净的运行环境。

python
# 默认:不加载任何配置(干净环境)
async for msg in query(prompt="..."):
    pass

# 加载项目配置
options = CodeBuddyAgentOptions(setting_sources=["project"])

# 加载所有配置(类似 CLI 行为)
options = CodeBuddyAgentOptions(setting_sources=["user", "project", "local"])

AppendSystemPrompt

python
@dataclass
class AppendSystemPrompt:
    append: str  # 追加到默认系统提示词的内容

Message Types

Message

所有消息类型的联合:

python
Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

SystemMessage

python
@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

UserMessage

python
@dataclass
class UserMessage:
    content: str | list[ContentBlock]
    uuid: str | None = None
    parent_tool_use_id: str | None = None

AssistantMessage

python
@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None
    error: str | None = None

ResultMessage

python
@dataclass
class 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

StreamEvent

python
@dataclass
class StreamEvent:
    uuid: str
    session_id: str
    event: dict[str, Any]
    parent_tool_use_id: str | None = None

ContentBlock

python
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

Input Types

AskUserQuestionInput

python
@dataclass
class AskUserQuestionInput:
    questions: list[AskUserQuestionQuestion]
    answers: dict[str, str] | None = None

AskUserQuestionQuestion

python
@dataclass
class AskUserQuestionQuestion:
    question: str     # 完整问题文本(应以 ? 结尾)
    header: str       # 简短标签(最多 12 个字符)
    options: list[AskUserQuestionOption]
    multi_select: bool  # 是否允许多选

AskUserQuestionOption

python
@dataclass
class AskUserQuestionOption:
    label: str        # 显示文本(1-5 个单词)
    description: str  # 选项说明

Errors

所有异常都继承自 CodeBuddySDKError

CodeBuddySDKError

python
class CodeBuddySDKError(Exception):
    """Base exception for CodeBuddy SDK errors."""
    pass

CLIConnectionError

当连接到 CLI 失败或未建立连接时抛出。

python
class CLIConnectionError(CodeBuddySDKError):
    pass

CLINotFoundError

当找不到 CLI 可执行文件时抛出。

python
class CLINotFoundError(CodeBuddySDKError):
    def __init__(
        self,
        message: str,
        platform: str | None = None,
        arch: str | None = None,
    ): ...

属性

属性类型说明
platformstr | None当前平台
archstr | None当前架构

CLIJSONDecodeError

当 CLI 输出的 JSON 解码失败时抛出。

python
class CLIJSONDecodeError(CodeBuddySDKError):
    pass

ProcessError

当 CLI 进程遇到错误时抛出。

python
class ProcessError(CodeBuddySDKError):
    pass

相关文档