第 3 课：query() 一键自动化

import anyio
from claude_agent_sdk import query, AssistantMessage, TextBlock

async def main():
    async for msg in query(prompt="用 ls 命令列出当前目录的文件"):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)

anyio.run(main)

import anyio
from claude_agent_sdk import (
    query, ClaudeAgentOptions,
    AssistantMessage, ResultMessage, TextBlock, ToolUseBlock
)

async def run_tests():
    """让 AI 自动跑测试并分析结果"""

    options = ClaudeAgentOptions(
        system_prompt="""你是一个 CI/CD 助手。你的职责是：
1. 执行测试命令
2. 分析测试结果
3. 如果有失败，分析失败原因并给出修复建议
4. 输出简洁的测试报告""",

        allowed_tools=["Bash", "Read"],  # 只允许执行命令和读文件
        max_turns=10,
    )

    print("🔄 开始自动测试...\n")

    async for msg in query(
        prompt="运行 pytest tests/ -v，分析结果，给出测试报告",
        options=options,
    ):
        if isinstance(msg, AssistantMessage):
            for block in msg.content:
                if isinstance(block, TextBlock):
                    print(block.text)
                elif isinstance(block, ToolUseBlock):
                    print(f"  🔧 {block.name}: {block.input}")

        elif isinstance(msg, ResultMessage):
            if msg.total_cost_usd:
                print(f"\n💰 本次费用: ${msg.total_cost_usd:.4f}")

anyio.run(run_tests)

options = ClaudeAgentOptions(
    # === 必知参数 ===
    system_prompt="...",         # 系统提示词（角色设定）
    allowed_tools=["Bash"],      # 白名单：允许哪些工具
    max_turns=10,                # AI 最多操作几轮

    # === 常用参数 ===
    model="claude-sonnet-4-5-20250929",  # 指定模型
    cwd="/path/to/project",      # 工作目录
    permission_mode="default",   # 权限模式

    # === 高级参数（后面课程会讲）===
    hooks={...},                 # Hook 安全管控（第 5 课）
    mcp_servers={...},           # 自定义工具（第 4 课）
    agents={...},                # 多 Agent（第 7 课）
)

async def lint_check():
    options = ClaudeAgentOptions(
        system_prompt="你是代码质量检查工具。运行 lint 命令，分析问题，按严重程度排序。",
        allowed_tools=["Bash", "Read"],
        max_turns=5,
    )

    async for msg in query(
        prompt="运行 ruff check src/ 检查代码质量，列出所有问题",
        options=options,
    ):
        # 处理消息...
        pass

async def analyze_logs():
    options = ClaudeAgentOptions(
        system_prompt="""你是日志分析专家。分析日志文件：
1. 找出所有 ERROR 和 WARNING
2. 按时间线排列
3. 分析错误原因
4. 给出解决建议""",
        allowed_tools=["Bash", "Read", "Grep"],
        max_turns=15,
    )

    async for msg in query(
        prompt="分析 /var/log/app/error.log 最近 1 小时的错误",
        options=options,
    ):
        # 处理消息...
        pass

async def check_config():
    options = ClaudeAgentOptions(
        system_prompt="你是配置审计员。检查配置文件的安全性和正确性。特别注意敏感信息泄露。",
        allowed_tools=["Read", "Grep"],  # 注意：不给 Bash！只读不执行
        max_turns=5,
    )

    async for msg in query(
        prompt="检查 config/ 目录下所有配置文件，找出硬编码的密码和密钥",
        options=options,
    ):
        # 处理消息...
        pass

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

    if isinstance(msg, AssistantMessage):
        # AI 说话了 或者 要用工具
        for block in msg.content:
            if isinstance(block, TextBlock):
                # AI 说了一段话
                print(block.text)
            elif isinstance(block, ToolUseBlock):
                # AI 要用工具
                print(f"工具: {block.name}, 参数: {block.input}")

    elif isinstance(msg, ResultMessage):
        # 任务结束了
        print(f"费用: ${msg.total_cost_usd:.4f}")