第 9 课：进阶扩展

第 1-2 课：基础认知 + 环境搭建
第 3 课：query() 一键自动化
第 4 课：自定义工具 MCP
第 5 课：Hook 安全管控
第 6 课：交互式会话 Client
第 7 课：多 Agent 协作
第 8 课：完整项目实战

from claude_agent_sdk import tool

@tool(
    "docker_ps",
    "列出运行中的 Docker 容器",
    {"all": {"type": "boolean", "description": "是否显示所有容器（包括已停止的）"}}
)
async def docker_ps(all: bool = False) -> str:
    import subprocess
    cmd = ["docker", "ps", "--format", "table {{.Names}}\t{{.Status}}\t{{.Ports}}"]
    if all:
        cmd.insert(2, "-a")
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
        return result.stdout.strip() or "没有运行中的容器"
    except FileNotFoundError:
        return "错误: docker 未安装"
    except Exception as e:
        return f"错误: {e}"


@tool(
    "docker_logs",
    "获取容器的最近日志",
    {
        "container": {"type": "string", "description": "容器名称或 ID"},
        "lines": {"type": "number", "description": "获取最近几行日志，默认 50"},
    }
)
async def docker_logs(container: str, lines: int = 50) -> str:
    import subprocess
    try:
        result = subprocess.run(
            ["docker", "logs", "--tail", str(lines), container],
            capture_output=True, text=True, timeout=15,
        )
        output = result.stdout + result.stderr
        return output.strip() or "没有日志输出"
    except Exception as e:
        return f"错误: {e}"


@tool(
    "docker_stats_snapshot",
    "获取容器资源使用快照（CPU/内存）",
    {}
)
async def docker_stats_snapshot() -> str:
    import subprocess
    try:
        result = subprocess.run(
            ["docker", "stats", "--no-stream",
             "--format", "table {{.Name}}\t{{.CPUPerc}}\t{{.MemUsage}}"],
            capture_output=True, text=True, timeout=15,
        )
        return result.stdout.strip() or "没有运行中的容器"
    except Exception as e:
        return f"错误: {e}"

你: 看看哪些容器在跑
🤖 调用 docker_ps → 列出容器

你: web-app 容器最近有什么错误？
🤖 调用 docker_logs(container="web-app") → 读日志
🤖 分析日志中的 ERROR → 给出诊断

你: 哪个容器吃资源最多？
🤖 调用 docker_stats_snapshot → 资源对比
🤖 分析 CPU/内存使用 → 给出优化建议

@tool(
    "run_lint",
    "运行代码格式检查（ruff）",
    {"path": {"type": "string", "description": "要检查的路径，默认 src/"}}
)
async def run_lint(path: str = "src/") -> str:
    import subprocess
    try:
        # 先检查
        check = subprocess.run(
            ["python", "-m", "ruff", "check", path],
            capture_output=True, text=True, timeout=30,
        )
        if check.returncode == 0:
            return f"✅ {path} 代码格式检查通过，没有问题！"

        # 有问题，统计数量
        lines = check.stdout.strip().split('\n')
        issue_count = len([l for l in lines if l.strip()])
        return f"⚠️ 发现 {issue_count} 个问题:\n{check.stdout[:1000]}"
    except Exception as e:
        return f"错误: {e}"


@tool(
    "run_typecheck",
    "运行类型检查（mypy）",
    {"path": {"type": "string", "description": "要检查的路径，默认 src/"}}
)
async def run_typecheck(path: str = "src/") -> str:
    import subprocess
    try:
        result = subprocess.run(
            ["python", "-m", "mypy", path],
            capture_output=True, text=True, timeout=60,
        )
        if result.returncode == 0:
            return f"✅ {path} 类型检查通过！"
        return f"⚠️ 类型检查发现问题:\n{result.stdout[:1000]}"
    except Exception as e:
        return f"错误: {e}"


@tool(
    "run_tests_with_coverage",
    "运行测试并生成覆盖率报告",
    {"test_path": {"type": "string", "description": "测试路径，默认 tests/"}}
)
async def run_tests_with_coverage(test_path: str = "tests/") -> str:
    import subprocess
    try:
        result = subprocess.run(
            ["python", "-m", "pytest", test_path,
             "--tb=short", "-q", "--co", "--no-header"],
            capture_output=True, text=True, timeout=120,
        )

        # 简单统计
        output = result.stdout
        if "passed" in output:
            return f"✅ 测试结果:\n{output[-500:]}"
        elif "failed" in output or "error" in output.lower():
            return f"❌ 测试失败:\n{output[-500:]}"
        else:
            return f"测试输出:\n{output[-500:]}"
    except Exception as e:
        return f"错误: {e}"

async def ci_pipeline():
    """一键 CI：lint → typecheck → test"""

    options = ClaudeAgentOptions(
        system_prompt="你是 CI 流水线管理员。按顺序运行 lint、typecheck、test，任何一步失败就停止。用中文回复。",
        mcp_servers={"ci": create_sdk_mcp_server(
            name="ci",
            version="1.0.0",
            tools=[run_lint, run_typecheck, run_tests_with_coverage],
        )},
        allowed_tools=[
            "mcp__ci__run_lint",
            "mcp__ci__run_typecheck",
            "mcp__ci__run_tests_with_coverage",
        ],
    )

    async for msg in query(
        prompt="依次运行 lint、类型检查和测试，给出综合报告",
        options=options,
    ):
        # 处理输出...
        pass

@tool(
    "env_check",
    "检查开发环境配置是否齐全",
    {}
)
async def env_check() -> str:
    import subprocess
    import shutil

    checks = []

    # 检查常用工具
    tools = {
        "python3": "Python",
        "node": "Node.js",
        "git": "Git",
        "docker": "Docker",
        "npm": "npm",
        "pip": "pip",
    }

    for cmd, name in tools.items():
        path = shutil.which(cmd)
        if path:
            try:
                ver = subprocess.run(
                    [cmd, "--version"],
                    capture_output=True, text=True, timeout=5,
                )
                version = ver.stdout.strip().split('\n')[0]
                checks.append(f"✅ {name}: {version}")
            except Exception:
                checks.append(f"✅ {name}: 已安装 ({path})")
        else:
            checks.append(f"❌ {name}: 未安装")

    # 检查环境变量
    import os
    env_vars = ["ANTHROPIC_API_KEY", "HOME", "PATH"]
    for var in env_vars:
        value = os.environ.get(var)
        if value:
            # 隐藏敏感值
            display = value[:8] + "..." if "KEY" in var or "SECRET" in var else "已设置"
            checks.append(f"✅ ${var}: {display}")
        else:
            checks.append(f"❌ ${var}: 未设置")

    return "\n".join(checks)

import json
import datetime

# 告警记录
alerts: list[dict] = []

async def alert_on_failure(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """当工具执行失败时触发告警"""

    tool_name = input_data.get("tool_name", "unknown")
    tool_response = str(input_data.get("tool_response", ""))

    # 检测错误关键词
    error_keywords = ["error", "failed", "exception", "timeout", "denied", "not found"]
    is_error = any(kw in tool_response.lower() for kw in error_keywords)

    if is_error:
        alert = {
            "time": datetime.datetime.now().isoformat(),
            "tool": tool_name,
            "severity": "HIGH" if "denied" in tool_response.lower() else "MEDIUM",
            "message": tool_response[:200],
        }
        alerts.append(alert)

        # 打印告警
        print(f"\n🚨 告警 [{alert['severity']}]: {tool_name} 执行出错")
        print(f"   {tool_response[:100]}")

        # 给 AI 更多上下文
        return {
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "additionalContext":
                    f"⚠️ {tool_name} 执行出现错误，请谨慎处理。"
                    f"错误次数: {len(alerts)}",
            }
        }

    return {}

import json
from pathlib import Path

STATE_FILE = Path("devops_state.json")

def save_state(state: dict) -> None:
    """保存会话状态"""
    STATE_FILE.write_text(json.dumps(state, indent=2, ensure_ascii=False))

def load_state() -> dict:
    """加载上次的会话状态"""
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {"tasks_completed": [], "last_check": None}

async def stateful_devops():
    """带状态管理的 DevOps 助手"""
    state = load_state()

    options = ClaudeAgentOptions(
        system_prompt=f"""你是 DevOps AI 助手。
上次会话信息：
- 已完成任务: {state.get('tasks_completed', [])}
- 上次检查时间: {state.get('last_check', '无')}
请根据历史信息继续工作。用中文回复。""",
        # ... 其他配置
    )

    async with ClaudeSDKClient(options=options) as client:
        # 使用 client 进行交互...
        pass

    # 退出时保存状态
    state["last_check"] = datetime.datetime.now().isoformat()
    save_state(state)

@tool(
    "k8s_pods",
    "列出 Kubernetes Pod 状态",
    {"namespace": {"type": "string", "description": "命名空间，默认 default"}}
)
async def k8s_pods(namespace: str = "default") -> str:
    import subprocess
    try:
        result = subprocess.run(
            ["kubectl", "get", "pods", "-n", namespace,
             "-o", "wide", "--no-headers"],
            capture_output=True, text=True, timeout=15,
        )
        if result.returncode != 0:
            return f"错误: {result.stderr}"
        return result.stdout.strip() or "该命名空间没有 Pod"
    except FileNotFoundError:
        return "错误: kubectl 未安装"


@tool(
    "k8s_events",
    "获取 Kubernetes 集群最近的事件（用于排错）",
    {"namespace": {"type": "string", "description": "命名空间，默认 default"}}
)
async def k8s_events(namespace: str = "default") -> str:
    import subprocess
    try:
        result = subprocess.run(
            ["kubectl", "get", "events", "-n", namespace,
             "--sort-by=.lastTimestamp",
             "--field-selector", "type!=Normal"],
            capture_output=True, text=True, timeout=15,
        )
        output = result.stdout.strip()
        if not output or "No resources found" in output:
            return "✅ 没有异常事件"
        return output[:2000]
    except FileNotFoundError:
        return "错误: kubectl 未安装"

1. 最小权限原则
   ├── 每个 Agent 只给必要的工具
   ├── 审查员不能写代码
   └── 测试员不能部署

2. 命令管控
   ├── 危险命令黑名单（rm -rf /、shutdown 等）
   ├── 安全命令白名单（ls、git status 等）
   └── 未知命令需要人工确认

3. 审计追踪
   ├── 每次工具调用都记录日志
   ├── 错误操作触发告警
   └── 会话结束输出摘要

4. 环境隔离
   ├── 开发环境和生产环境分开
   ├── API Key 用环境变量，不硬编码
   └── 敏感操作需要二次确认

5. 渐进式信任
   ├── 新工具先在测试环境验证
   ├── 逐步放开权限
   └── 定期审查权限配置