uv add codebuddy-agent-sdk
# or
pip install codebuddy-agent-sdk

options = CodeBuddyAgentOptions(
    # Required only for Dedicated edition or self-hosted deployment users. Enter your enterprise service address:
    # endpoint="https://your-company.copilot.qq.com",
    env={
        "CODEBUDDY_API_KEY": os.environ.get("MY_API_KEY"),
        # Required for China edition users:
        # "CODEBUDDY_INTERNET_ENVIRONMENT": "internal"
        # Required for iOA edition users:
        # "CODEBUDDY_INTERNET_ENVIRONMENT": "ioa"
        # Required for Dedicated edition users:
        # "CODEBUDDY_INTERNET_ENVIRONMENT": "cloudhosted"
        # Required for self-hosted deployment users:
        # "CODEBUDDY_INTERNET_ENVIRONMENT": "selfhosted"
    }
)

import httpx
from codebuddy_agent_sdk import query, CodeBuddyAgentOptions

async def get_oauth_token(client_id: str, client_secret: str) -> str:
    async with httpx.AsyncClient() as client:
        response = await client.post(
            "https://copilot.tencent.com/oauth2/token",
            data={
                "grant_type": "client_credentials",
                "client_id": client_id,
                "client_secret": client_secret,
            },
        )
        return response.json()["access_token"]

# Obtain a token and call the SDK
token = await get_oauth_token("your-client-id", "your-client-secret")

options = CodeBuddyAgentOptions(
    env={"CODEBUDDY_AUTH_TOKEN": token}
)

async for msg in query(prompt="Hello", options=options):
    print(msg)

import asyncio
from codebuddy_agent_sdk import query, CodeBuddyAgentOptions
from codebuddy_agent_sdk import AssistantMessage, TextBlock

async def main():
    options = CodeBuddyAgentOptions(
        permission_mode="bypassPermissions"
    )

    async for message in query(prompt="Please explain what a recursive function is", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    print(block.text)

asyncio.run(main())

from codebuddy_agent_sdk import ResultMessage

async for message in query(prompt="...", options=options):
    if isinstance(message, ResultMessage):
        if message.subtype == "success":
            print(f"Done! Duration: {message.duration_ms} ms")
            print(f"Cost: {message.total_cost_usd} USD")
        else:
            print("Execution error")

from codebuddy_agent_sdk import (
    SystemMessage, AssistantMessage, ResultMessage,
    TextBlock, ToolUseBlock, ToolResultBlock
)

async for message in query(prompt="...", options=options):
    if isinstance(message, SystemMessage):
        # Session initialization message
        print(f"Session ID: {message.data.get('session_id')}")
        print(f"Available tools: {message.data.get('tools')}")

    elif isinstance(message, AssistantMessage):
        # AI assistant response
        for block in message.content:
            if isinstance(block, TextBlock):
                print(f"[Text] {block.text}")
            elif isinstance(block, ToolUseBlock):
                print(f"[Tool call] {block.name}: {block.input}")
            elif isinstance(block, ToolResultBlock):
                print(f"[Tool result] {block.content}")

    elif isinstance(message, ResultMessage):
        # Query completed
        print(f"Execution completed. Duration: {message.duration_ms} ms")

options = CodeBuddyAgentOptions(
    permission_mode="plan"  # Read-only mode
)
async for msg in query(prompt="Analyze the project structure", options=options):
    pass

options = CodeBuddyAgentOptions(
    cwd="/path/to/project"
)

options = CodeBuddyAgentOptions(
    model="deepseek-v3.1",
    fallback_model="deepseek-v3.1"
)

options = CodeBuddyAgentOptions(
    max_turns=20,        # Maximum number of conversation turns
)

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

    # Or load all configurations.
    # setting_sources=["user", "project", "local"]
)

# Load only the project configuration and ignore user and local configurations.
options = CodeBuddyAgentOptions(
    setting_sources=["project"],
    permission_mode="bypassPermissions"
)

# Default behavior: Do not load any configuration.
# All behaviors are explicitly defined through options.
options = CodeBuddyAgentOptions(
    agents={"reviewer": AgentDefinition(...)},
    mcp_servers={"db": {...}},
    allowed_tools=["Read", "Grep", "Glob"]
)

from codebuddy_agent_sdk import (
    query, CodeBuddyAgentOptions,
    CanUseToolOptions, PermissionResultAllow, PermissionResultDeny
)

async def can_use_tool(
    tool_name: str,
    input_data: dict,
    options: CanUseToolOptions
):
    # Only read-only tools are allowed.
    read_only_tools = ["Read", "Glob", "Grep"]

    if tool_name in read_only_tools:
        return PermissionResultAllow(updated_input=input_data)

    # Deny other tools.
    return PermissionResultDeny(
        message=f"Tool {tool_name} is not allowed."
    )

options = CodeBuddyAgentOptions(can_use_tool=can_use_tool)

dangerous_commands = ["rm -rf", "sudo", "chmod 777"]

async def can_use_tool(tool_name, input_data, options):
    if tool_name == "Bash":
        command = input_data.get("command", "")
        for dangerous in dangerous_commands:
            if dangerous in command:
                return PermissionResultDeny(
                    message=f"Dangerous command blocked: {dangerous}",
                    interrupt=True  # Interrupt the entire session.
                )
    return PermissionResultAllow(updated_input=input_data)