Python SDK 参考

uv add codebuddy-agent-sdk

pip install codebuddy-agent-sdk

async def query(
    *,
    prompt: str | AsyncIterable[dict[str, Any]],
    options: CodeBuddyAgentOptions | None = None,
    transport: Transport | None = None,
) -> 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,
    ): ...

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:

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)

async def mcp_server_status(self) -> list[McpServerStatus]:

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:

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)

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

async def interrupt(self) -> None:

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

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

@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
    persist_session: bool = True
    agents: dict[str, AgentDefinition] | None = None
    setting_sources: list[SettingSource] | None = None
    can_use_tool: CanUseTool | None = None

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

@dataclass
class 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"]
)

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

SettingSource = Literal["user", "project", "local"]

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

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

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

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

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

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

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

class TaskUsage(TypedDict):
    total_tokens: int
    tool_uses: int
    duration_ms: int

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

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

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

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