快速开始

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

options = CodeBuddyAgentOptions(
    # 仅专享版或私有化部署用户需要设置，填写您的企业服务地址：
    # endpoint="https://your-company.copilot.qq.com",
    env={
        "CODEBUDDY_API_KEY": os.environ.get("MY_API_KEY"),
        # 中国版用户需要设置：
        # "CODEBUDDY_INTERNET_ENVIRONMENT": "internal"
        # iOA 版用户需要设置：
        # "CODEBUDDY_INTERNET_ENVIRONMENT": "ioa"
        # 专享版用户需要设置：
        # "CODEBUDDY_INTERNET_ENVIRONMENT": "cloudhosted"
        # 私有化部署用户需要设置：
        # "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"]

# 获取 token 并调用 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="请解释什么是递归函数", 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"完成！耗时： {message.duration_ms} ms")
            print(f"费用： {message.total_cost_usd} USD")
        else:
            print("执行出错")

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

async for message in query(prompt="...", options=options):
    if isinstance(message, SystemMessage):
        # 会话初始化消息
        print(f"会话 ID: {message.data.get('session_id')}")
        print(f"可用工具： {message.data.get('tools')}")

    elif isinstance(message, AssistantMessage):
        # AI 助手响应
        for block in message.content:
            if isinstance(block, TextBlock):
                print(f"[文本] {block.text}")
            elif isinstance(block, ToolUseBlock):
                print(f"[工具调用] {block.name}: {block.input}")
            elif isinstance(block, ToolResultBlock):
                print(f"[工具结果] {block.content}")

    elif isinstance(message, ResultMessage):
        # 查询完成
        print(f"执行完成，耗时： {message.duration_ms} ms")

options = CodeBuddyAgentOptions(
    permission_mode="plan"  # 只读模式
)
async for msg in query(prompt="分析项目结构", 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,        # 最大对话轮数
)

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

    # 或加载全部配置
    # setting_sources=["user", "project", "local"]
)

# 只加载项目配置，忽略用户和本地配置
options = CodeBuddyAgentOptions(
    setting_sources=["project"],
    permission_mode="bypassPermissions"
)

# 默认行为：不加载任何配置
# 所有行为通过 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
):
    # 只允许读取类工具
    read_only_tools = ["Read", "Glob", "Grep"]

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

    # 拒绝其他工具
    return PermissionResultDeny(
        message=f"工具 {tool_name} 不允许使用"
    )

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}",
                    interrupt=True  # 中断整个会话
                )
    return PermissionResultAllow(updated_input=input_data)