Python SDK Reference

Version Requirement: This document is for CodeBuddy Agent SDK v0.1.0 and above.

This document provides a complete API reference for the Python SDK. For quick start and usage examples, please refer to SDK Overview.

Requirements

Dependency	Version Requirement
Python	>= 3.10
CodeBuddy CLI	Installed

Async Runtime: The SDK is based on asyncio, all APIs are asynchronous.

Installation

Recommended to use uv for dependency management:

bash

uv add codebuddy-agent-sdk

Or using pip:

bash

pip install codebuddy-agent-sdk

Environment Variables

Variable Name	Description	Required
`CODEBUDDY_CODE_PATH`	CodeBuddy CLI executable path	Optional

If not set, the SDK will look for CLI in the following order:

Environment variable CODEBUDDY_CODE_PATH
Built-in binary within SDK package
Development environment monorepo path

Authentication Configuration

The SDK supports authentication using existing login credentials, API Key, or OAuth Client Credentials. See SDK Overview - Authentication Configuration for details.

Functions

query()

Main API entry point, creates a query and returns an async iterator of messages.

python

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

Parameters:

Parameter	Type	Description
`prompt`	`str \| AsyncIterable[dict]`	Query prompt or user message stream
`options`	`CodeBuddyAgentOptions`	Configuration options (optional)
`transport`	`Transport`	Custom transport layer (optional)

Returns: AsyncIterator[Message] - Async iterator of messages

Example:

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

Client class for bidirectional interactive conversations. Supports multi-turn dialogues, interruptions, and dynamic control.

python

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

Methods:

connect()

Connect to CodeBuddy.

python

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

query()

Send user message.

python

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

receive_response()

Receive messages until a ResultMessage is received.

python

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

receive_messages()

Receive all messages (does not auto-stop).

python

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

disconnect()

Disconnect.

python

async def disconnect(self) -> None:

Context Manager Support:

python

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

mcp_server_status()

Get MCP server connection status.

python

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

Authentication

The SDK provides an independent authentication API with a two-phase design: first obtain the login URL, then wait for the user to complete authentication.

authenticate()

Start the authentication flow, returns an AuthFlow object.

python

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:

Parameters:

Parameter	Type	Description
`method_id`	`str`	Authentication method identifier (default `"external"`)
`environment`	`str \| None`	Predefined environment name
`endpoint`	`str \| None`	Custom endpoint URL (mutually exclusive with environment)
`codebuddy_code_path`	`str \| None`	CLI executable path
`env`	`dict[str, str] \| None`	Additional environment variables
`timeout`	`float`	Timeout for user to complete login (seconds, default 300)

Returns: AuthFlow — Awaitable object carrying the login URL

Example:

python

from codebuddy_agent_sdk import authenticate

# Two-phase: get URL → show to user → wait for completion
auth = await authenticate()
if auth.auth_url:
    print(f"Please visit: {auth.auth_url}")
result = await auth
print(f"Welcome, {result.userinfo.user_name}")

# When already logged in, auth.auth_url is empty and await returns immediately
auth = await authenticate()
result = await auth  # Returns immediately if already logged in

# Custom timeout
auth = await authenticate()
result = await auth.wait(timeout=60)

AuthFlow

Authentication flow object returned by authenticate(). Implements the __await__ protocol and can be directly awaited.

Attributes:

Attribute	Type	Description
`auth_url`	`str`	Login URL (empty string when already logged in)
`method_id`	`str \| None`	Authentication method identifier

Methods:

wait()

Wait for the user to complete authentication.

python

async def wait(self, timeout: float | None = None) -> AuthenticateResponse:

cancel()

Cancel the authentication flow and release resources.

python

async def cancel(self) -> None:

logout()

Log out and clear cached authentication tokens.

python

async def logout(
    *,
    environment: str | None = None,
    endpoint: str | None = None,
    codebuddy_code_path: str | None = None,
    env: dict[str, str] | None = None,
) -> None:

Example:

python

from codebuddy_agent_sdk import logout

await logout()

Unstable API

Warning: The following APIs are experimental and may change in future versions.

interrupt()

Send interrupt signal.

python

async def interrupt(self) -> None:

set_permission_mode()

Dynamically modify permission mode.

python

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

set_model()

Dynamically modify model.

python

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

Types

CodeBuddyAgentOptions

Complete configuration options:

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

Field	Type	Description
`allowed_tools`	`list[str]`	Tool whitelist to auto-allow
`disallowed_tools`	`list[str]`	Tool blacklist to disallow
`system_prompt`	`str \| AppendSystemPrompt`	System prompt configuration
`mcp_servers`	`dict[str, McpServerConfig]`	MCP server configuration
`permission_mode`	`PermissionMode`	Permission mode
`continue_conversation`	`bool`	Continue recent session
`resume`	`str`	Session ID to resume
`max_turns`	`int`	Maximum conversation turns
`model`	`str`	Specify model
`fallback_model`	`str`	Fallback model
`cwd`	`str \| Path`	Working directory
`codebuddy_code_path`	`str \| Path`	CLI executable path
`env`	`dict[str, str]`	Environment variables
`extra_args`	`dict[str, str \| None]`	Extra CLI arguments
`stderr`	`Callable[[str], None]`	stderr callback
`hooks`	`dict[HookEvent, list[HookMatcher]]`	Hook configuration
`include_partial_messages`	`bool`	Include partial messages
`fork_session`	`bool`	Fork session
`agents`	`dict[str, AgentDefinition]`	Custom agents
`setting_sources`	`list[SettingSource]`	Setting sources
`can_use_tool`	`CanUseTool`	Permission callback function
`max_thinking_tokens`	`int`	Maximum thinking tokens (deprecated, use `thinking` instead)
`thinking`	`ThinkingConfig`	Thinking mode configuration: `{"type": "adaptive"}`, `{"type": "enabled", "budget_tokens": N}`, or `{"type": "disabled"}`
`effort`	`'low' \| 'medium' \| 'high' \| 'xhigh'`	Model reasoning effort level

PermissionMode

python

PermissionMode = Literal["default", "acceptEdits", "plan", "bypassPermissions"]

Value	Description
`"default"`	Default mode, all operations require confirmation
`"acceptEdits"`	Auto-approve file edits
`"plan"`	Planning mode, only allow reads
`"bypassPermissions"`	Skip all permission checks

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 description
    prompt: str             # System prompt
    tools: list[str] | None = None           # Allowed tools
    disallowed_tools: list[str] | None = None  # Disallowed tools
    model: str | None = None                 # Model to use

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"]
    | Literal["WorktreeCreate"]
    | Literal["WorktreeRemove"]
)

HookMatcher

python

@dataclass
class HookMatcher:
    matcher: str | None = None      # Match pattern (supports regex)
    hooks: list[HookCallback] = field(default_factory=list)
    timeout: float | None = None    # Timeout in seconds

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

Controls which filesystem locations the SDK loads configuration from.

python

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

Value	Description	Location
`"user"`	Global user settings	`~/.codebuddy/settings.json`
`"project"`	Project shared settings	`.codebuddy/settings.json`
`"local"`	Project local settings	`.codebuddy/settings.local.json`

Default Behavior: When setting_sources is not specified, the SDK does not load any filesystem configuration. This provides a completely clean runtime environment.

python

# Default: no configuration loaded (clean environment)
async for msg in query(prompt="..."):
    pass

# Load project configuration
options = CodeBuddyAgentOptions(setting_sources=["project"])

# Load all configurations (similar to CLI behavior)
options = CodeBuddyAgentOptions(setting_sources=["user", "project", "local"])

AppendSystemPrompt

python

@dataclass
class AppendSystemPrompt:
    append: str  # Content to append to default system prompt

Message Types

Message

Union of all message types:

python

Message = UserMessage | AssistantMessage | SystemMessage | ResultMessage | StreamEvent

TaskStartedMessage / TaskNotificationMessage are subclasses of SystemMessage and are already covered by the union (isinstance / case SystemMessage() continues to match). They are exported separately for precise type checking at call sites.

SystemMessage

python

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

TaskStartedMessage

Emitted when a background task (Bash / PowerShell / Workflow / Agent with run_in_background: true) enters the running state. This is a subclass of SystemMessage; isinstance(msg, SystemMessage) still holds.

python

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

TaskUsage

Usage statistics carried by task_progress / task_notification (aligned with Claude Code's TaskUsage). Sub-agent (task_type == "Agent") background tasks have values; background shell tasks typically omit this.

python

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

TaskProgressMessage

Background task progress event. Event-driven (not timer-based): one message per completed tool_use, carrying cumulative usage and the most recent tool name last_tool_name. Background shell tasks do not emit progress.

python

@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

TaskUpdatedMessage

Background task status transition event. patch carries the changed fields for this update (at minimum status; terminal states also include end_time).

Lifecycle note: A background task's terminal state sometimes arrives via task_updated only (with patch.status being a terminal state) without a corresponding task_notification. Consumers tracking "active tasks" should treat the terminal status from both messages equally — use the TERMINAL_TASK_STATUSES frozenset to check ({"completed", "failed", "stopped", "killed"}).

python

@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

TaskNotificationMessage

Emitted when a background task completes / fails / is stopped. In stdio stream-json long-lived connection mode, if the task completes after the result message of the turn that triggered it, this message is proactively pushed back to the same output stream — consumers must use receive_messages() to continuously read and receive it (query() and receive_response() stop at the first ResultMessage and will miss background completion events). usage is present for sub-agent tasks and omitted for shell tasks.

query() automatically disables background tasks: Since query() stops at the first ResultMessage and closes the subprocess — unable to receive cross-turn push events — the SDK automatically injects CODEBUDDY_CODE_DISABLE_BACKGROUND_TASKS=1 under the query() path (Bash / PowerShell / Agent run_in_background is hidden/degraded to foreground). If you need to keep background tasks under query(), explicitly set this variable (any value, including "0") in options.env or the process environment to override. The CodeBuddySDKClient + receive_messages() path is unaffected.

python

@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          # Background task stdout file path (file mode)
    output_stderr_file: str | None = None
    usage: TaskUsage | None = None

Multiple concurrent background tasks are distinguished by task_id; tool_use_id links back to the tool_use that the model used to initiate the task. Example (continuously read to receive notifications after overall completion):

python

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("Run two background commands in parallel and tell me the results when done")
        # Use receive_messages to continuously read — do not use receive_response (it stops at the first 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 on your own once you've collected the tasks you care about

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

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     # Complete question text (should end with ?)
    header: str       # Short label (max 12 characters)
    options: list[AskUserQuestionOption]
    multi_select: bool  # Whether to allow multiple selections

AskUserQuestionOption

python

@dataclass
class AskUserQuestionOption:
    label: str        # Display text (1-5 words)
    description: str  # Option description

Errors

All exceptions inherit from CodeBuddySDKError.

CodeBuddySDKError

python

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

CLIConnectionError

Thrown when connection to CLI fails or no connection is established.

python

class CLIConnectionError(CodeBuddySDKError):
    pass

CLINotFoundError

Thrown when CLI executable is not found.

python

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

Attributes:

Attribute	Type	Description
`platform`	`str \| None`	Current platform
`arch`	`str \| None`	Current architecture

CLIJSONDecodeError

Thrown when JSON decoding of CLI output fails.

python

class CLIJSONDecodeError(CodeBuddySDKError):
    pass

ProcessError

Thrown when CLI process encounters an error.

python

class ProcessError(CodeBuddySDKError):
    pass

CLIStartupError

Thrown when the CLI process crashes or produces no output during the startup phase.

python

class CLIStartupError(CodeBuddySDKError):
    def __init__(
        self,
        message: str,
        stderr: str = "",
        exit_code: int | None = None,
    ): ...

Attributes:

Attribute	Type	Description
`stderr`	`str`	stderr output from CLI process
`exit_code`	`int \| None`	Process exit code

ExecutionError

Thrown when execution fails (e.g., authentication error, API error). Contains the errors array from ResultMessage.

python

class ExecutionError(CodeBuddySDKError):
    def __init__(self, errors: list[str], subtype: str): ...

Attributes:

Attribute	Type	Description
`errors`	`list[str]`	List of error messages
`subtype`	`str`	Error subtype

AuthenticationError

Thrown when authentication fails.

python

class AuthenticationError(CodeBuddySDKError):
    def __init__(self, error_type: str, message: str): ...

Attributes:

Attribute	Type	Description
`error_type`	`str`	Error type (e.g., `"timeout"`, `"auth_failed"`)

Auth Types

AuthenticateResponse

python

@dataclass(slots=True)
class AuthenticateResponse:
    userinfo: UserInfo

UserInfo

python

@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

McpServerStatus

python

@dataclass(slots=True)
class McpServerStatus:
    name: str
    status: Literal["connected", "failed", "needs-auth", "pending"]
    server_info: dict[str, Any] | None = None

SDK Overview - Quick start and usage examples
TypeScript SDK Reference - TypeScript version API
Hook Reference Guide - Detailed Hook configuration instructions
MCP Integration - MCP server configuration guide

Python SDK Reference ​

Requirements ​

Installation ​

Environment Variables ​

Authentication Configuration ​

Functions ​

query() ​

Client Class ​

CodeBuddySDKClient ​

connect() ​

query() ​

receive_response() ​

receive_messages() ​

disconnect() ​

mcp_server_status() ​

Authentication ​

authenticate() ​

AuthFlow ​

wait() ​

cancel() ​

logout() ​

Unstable API ​

interrupt() ​

set_permission_mode() ​

set_model() ​

Types ​

CodeBuddyAgentOptions ​

PermissionMode ​

PermissionResult ​

CanUseTool ​

AgentDefinition ​

McpServerConfig ​

HookEvent ​

HookMatcher ​

HookCallback ​

SettingSource ​

AppendSystemPrompt ​

Message Types ​

Message ​

SystemMessage ​

TaskStartedMessage ​

TaskUsage ​

TaskProgressMessage ​

TaskUpdatedMessage ​

TaskNotificationMessage ​

UserMessage ​

AssistantMessage ​

ResultMessage ​

StreamEvent ​

ContentBlock ​

Input Types ​

AskUserQuestionInput ​

AskUserQuestionQuestion ​

AskUserQuestionOption ​

Errors ​

CodeBuddySDKError ​

CLIConnectionError ​

CLINotFoundError ​

CLIJSONDecodeError ​

ProcessError ​

CLIStartupError ​

ExecutionError ​

AuthenticationError ​

Auth Types ​

AuthenticateResponse ​

UserInfo ​

McpServerStatus ​

Related Documentation ​

Python SDK Reference

Requirements

Installation

Environment Variables

Authentication Configuration

Functions

query()

Client Class

CodeBuddySDKClient

connect()

query()

receive_response()

receive_messages()

disconnect()

mcp_server_status()

Authentication

authenticate()

AuthFlow

wait()

cancel()

logout()

Unstable API

interrupt()

set_permission_mode()

set_model()

Types

CodeBuddyAgentOptions

PermissionMode

PermissionResult

CanUseTool

AgentDefinition

McpServerConfig

HookEvent

HookMatcher

HookCallback

SettingSource

AppendSystemPrompt

Message Types

Message

SystemMessage

TaskStartedMessage

TaskUsage

TaskProgressMessage

TaskUpdatedMessage

TaskNotificationMessage

UserMessage

AssistantMessage

ResultMessage

StreamEvent

ContentBlock

Input Types

AskUserQuestionInput

AskUserQuestionQuestion

AskUserQuestionOption

Errors

CodeBuddySDKError

CLIConnectionError

CLINotFoundError

CLIJSONDecodeError

ProcessError

CLIStartupError

ExecutionError

AuthenticationError

Auth Types

AuthenticateResponse

UserInfo

McpServerStatus

Related Documentation