第 8 课：完整项目实战

devops-assistant/
├── main.py                 ← 主程序入口
├── tools/
│   ├── __init__.py
│   └── devops_tools.py     ← 自定义 DevOps 工具
├── hooks/
│   ├── __init__.py
│   └── safety.py           ← 安全 Hook
├── agents/
│   ├── __init__.py
│   └── definitions.py      ← Agent 定义
└── requirements.txt

"""DevOps 专用工具集"""

import json
import subprocess
from claude_agent_sdk import tool, create_sdk_mcp_server


@tool(
    "git_status",
    "获取当前 Git 仓库的状态摘要",
    {}
)
async def git_status() -> str:
    """获取 Git 状态"""
    try:
        result = subprocess.run(
            ["git", "status", "--short", "--branch"],
            capture_output=True, text=True, timeout=10,
        )
        if result.returncode != 0:
            return f"错误: {result.stderr}"

        output = result.stdout.strip()
        if not output:
            return "工作区干净，没有未提交的变更。"
        return output
    except FileNotFoundError:
        return "错误: git 命令未找到"
    except subprocess.TimeoutExpired:
        return "错误: git 命令超时"


@tool(
    "git_recent_commits",
    "获取最近的 Git 提交记录",
    {"count": {"type": "number", "description": "要获取的提交数量，默认 5"}}
)
async def git_recent_commits(count: int = 5) -> str:
    """获取最近的提交"""
    try:
        result = subprocess.run(
            ["git", "log", f"-{count}", "--oneline", "--decorate"],
            capture_output=True, text=True, timeout=10,
        )
        if result.returncode != 0:
            return f"错误: {result.stderr}"
        return result.stdout.strip() or "没有找到提交记录。"
    except Exception as e:
        return f"错误: {e}"


@tool(
    "service_health",
    "检查服务的健康状态（通过 HTTP 请求）",
    {"url": {"type": "string", "description": "健康检查的 URL"}}
)
async def service_health(url: str) -> str:
    """检查服务健康状态"""
    try:
        result = subprocess.run(
            ["curl", "-s", "-o", "/dev/null", "-w",
             '{"status_code": %{http_code}, "time_total": %{time_total}}',
             "-m", "5", url],
            capture_output=True, text=True, timeout=10,
        )
        data = json.loads(result.stdout)
        status = "✅ 健康" if data["status_code"] == 200 else f"❌ 异常 (HTTP {data['status_code']})"
        return f"{status} | 响应时间: {data['time_total']:.2f}s | URL: {url}"
    except Exception as e:
        return f"❌ 无法连接: {url} | 错误: {e}"


@tool(
    "disk_usage",
    "检查磁盘使用情况",
    {"path": {"type": "string", "description": "要检查的路径，默认 /"}}
)
async def disk_usage(path: str = "/") -> str:
    """检查磁盘使用情况"""
    try:
        result = subprocess.run(
            ["df", "-h", path],
            capture_output=True, text=True, timeout=10,
        )
        return result.stdout.strip()
    except Exception as e:
        return f"错误: {e}"


# 创建 MCP 服务器
devops_mcp = create_sdk_mcp_server(
    name="devops-tools",
    version="1.0.0",
    tools=[git_status, git_recent_commits, service_health, disk_usage],
)

"""DevOps 安全 Hook"""

import datetime
from claude_agent_sdk.types import HookInput, HookContext, HookJSONOutput

# 审计日志
audit_log: list[dict] = []

# 危险命令黑名单
DANGEROUS_PATTERNS = [
    "rm -rf /",
    "rm -rf ~",
    "rm -rf /*",
    "dd if=/dev/zero",
    "> /dev/sda",
    "chmod -R 777 /",
    ":(){ :|:& };:",
    "mkfs.",
    "shutdown",
    "reboot",
    "init 0",
    "init 6",
]

# 安全命令白名单（自动批准）
SAFE_PREFIXES = [
    "ls", "cat", "head", "tail", "wc",
    "grep", "find", "which", "whoami",
    "ps", "top", "df", "du", "free",
    "git status", "git log", "git diff", "git branch",
    "docker ps", "docker images", "docker logs",
    "pytest", "python -m pytest",
    "ruff check", "mypy",
    "curl -s",
    "echo",
]


async def safety_gate(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """安全门：拦截危险命令，放行安全命令"""

    tool_name = input_data["tool_name"]
    tool_input = input_data["tool_input"]

    # 只检查 Bash 命令
    if tool_name != "Bash":
        return {}

    command = tool_input.get("command", "")

    # 1. 检查危险命令
    for pattern in DANGEROUS_PATTERNS:
        if pattern in command:
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "deny",
                    "permissionDecisionReason": f"🚫 危险命令已拦截: {pattern}",
                }
            }

    # 2. 自动批准安全命令
    cmd_stripped = command.strip()
    for prefix in SAFE_PREFIXES:
        if cmd_stripped.startswith(prefix):
            return {
                "hookSpecificOutput": {
                    "hookEventName": "PreToolUse",
                    "permissionDecision": "allow",
                    "permissionDecisionReason": f"✅ 安全命令自动批准: {prefix}",
                }
            }

    # 3. 其他命令：交给默认权限机制（会提示用户确认）
    return {}


async def auto_approve_readonly(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """自动批准只读工具"""

    tool_name = input_data["tool_name"]

    if tool_name in ["Read", "Grep", "Glob"]:
        return {
            "hookSpecificOutput": {
                "hookEventName": "PreToolUse",
                "permissionDecision": "allow",
                "permissionDecisionReason": "只读操作，自动批准",
            }
        }

    return {}


async def audit_logger(
    input_data: HookInput,
    tool_use_id: str | None,
    context: HookContext
) -> HookJSONOutput:
    """审计日志：记录所有工具调用"""

    entry = {
        "time": datetime.datetime.now().isoformat(),
        "tool": input_data.get("tool_name", "unknown"),
        "input": str(input_data.get("tool_input", {}))[:200],
        "has_error": "error" in str(input_data.get("tool_response", "")).lower(),
    }
    audit_log.append(entry)

    # 如果出错了，给 AI 额外上下文
    if entry["has_error"]:
        return {
            "hookSpecificOutput": {
                "hookEventName": "PostToolUse",
                "additionalContext": "上一个操作出错了，请检查后再继续。",
            }
        }

    return {}


def get_audit_log() -> list[dict]:
    """获取审计日志"""
    return audit_log.copy()

"""DevOps Agent 定义"""

from claude_agent_sdk import AgentDefinition

# 代码审查员
reviewer = AgentDefinition(
    description="审查代码变更，找出 bug、安全问题和代码质量问题",
    prompt="""你是高级代码审查员。你的工作流程：
1. 用 git diff 查看最近的代码变更
2. 逐个文件审查，关注：
   - 逻辑错误和潜在 bug
   - 安全漏洞（SQL 注入、XSS、硬编码密钥等）
   - 性能问题
   - 代码风格和可读性
3. 给出具体的改进建议
4. 最终给出"通过"或"需要修改"的结论

请用中文回复。""",
    tools=["Bash", "Read", "Grep"],
    model="sonnet",
)

# 测试工程师
tester = AgentDefinition(
    description="运行测试、分析测试结果、报告测试覆盖率",
    prompt="""你是测试工程师。你的工作流程：
1. 找到项目的测试配置（pytest.ini、setup.cfg 等）
2. 运行测试套件
3. 分析测试结果：
   - 通过/失败数量
   - 失败测试的原因分析
   - 测试覆盖率（如果可用）
4. 给出"测试通过"或"测试失败"的结论

请用中文回复。""",
    tools=["Bash", "Read"],
    model="sonnet",
)

# 日志分析师
log_analyzer = AgentDefinition(
    description="分析应用日志，找出错误和异常模式",
    prompt="""你是日志分析专家。你的工作流程：
1. 读取指定的日志文件
2. 统计 ERROR、WARNING、INFO 的数量
3. 按时间线整理错误
4. 分析错误模式和根因
5. 给出解决建议

请用中文回复。""",
    tools=["Read", "Grep", "Bash"],
    model="sonnet",
)

# 部署工程师
deployer = AgentDefinition(
    description="执行部署操作并验证部署结果",
    prompt="""你是部署工程师。你的工作流程：
1. 确认代码审查和测试都已通过
2. 执行部署脚本
3. 验证部署结果（健康检查）
4. 如果部署失败，执行回滚
5. 报告部署状态

安全规则：
- 只在确认审查和测试通过后才部署
- 部署前先备份
- 部署后必须验证

请用中文回复。""",
    tools=["Bash", "Read"],
    model="sonnet",
)

# 导出所有 Agent
ALL_AGENTS = {
    "reviewer": reviewer,
    "tester": tester,
    "log-analyzer": log_analyzer,
    "deployer": deployer,
}

"""DevOps AI 助手 —— 主程序"""

import anyio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    AssistantMessage, ResultMessage, TextBlock, ToolUseBlock,
)
from claude_agent_sdk.types import HookMatcher

from tools.devops_tools import devops_mcp
from hooks.safety import safety_gate, auto_approve_readonly, audit_logger, get_audit_log
from agents.definitions import ALL_AGENTS

SYSTEM_PROMPT = """你是 DevOps AI 助手。你能帮用户完成以下工作：

🔍 代码管理：审查代码、分析 Git 历史、检查代码质量
🧪 测试：运行测试、分析测试结果、报告覆盖率
📊 日志分析：查找错误、分析模式、追踪问题
🚀 部署：执行部署、验证结果、健康检查
🔧 系统管理：检查磁盘、监控服务、查看进程

你有以下专业 Agent 可以调度：
- reviewer：代码审查员
- tester：测试工程师
- log-analyzer：日志分析师
- deployer：部署工程师

你还有以下自定义工具：
- git_status：Git 状态摘要
- git_recent_commits：最近的提交记录
- service_health：服务健康检查
- disk_usage：磁盘使用情况

请用中文回复。根据用户的需求选择合适的 Agent 或工具。"""


async def main():
    options = ClaudeAgentOptions(
        system_prompt=SYSTEM_PROMPT,
        allowed_tools=[
            "Bash", "Read", "Write", "Edit", "Grep", "Glob",
            # 自定义工具
            "mcp__devops-tools__git_status",
            "mcp__devops-tools__git_recent_commits",
            "mcp__devops-tools__service_health",
            "mcp__devops-tools__disk_usage",
        ],
        agents=ALL_AGENTS,
        mcp_servers={"devops-tools": devops_mcp},
        hooks={
            "PreToolUse": [
                # 拦截危险命令
                HookMatcher(matcher="Bash", hooks=[safety_gate]),
                # 自动批准只读操作
                HookMatcher(matcher=None, hooks=[auto_approve_readonly]),
            ],
            "PostToolUse": [
                # 审计日志
                HookMatcher(matcher=None, hooks=[audit_logger]),
            ],
        },
    )

    print("=" * 50)
    print("🤖 DevOps AI 助手")
    print("=" * 50)
    print("输入任务开始工作，输入 'quit' 退出")
    print("输入 'log' 查看审计日志")
    print()

    async with ClaudeSDKClient(options=options) as client:
        while True:
            user_input = input("你: ").strip()

            if user_input.lower() in ['quit', 'exit', 'q']:
                # 退出前打印审计日志摘要
                log = get_audit_log()
                if log:
                    print(f"\n📋 本次会话共执行了 {len(log)} 次工具调用")
                    errors = sum(1 for e in log if e["has_error"])
                    if errors:
                        print(f"  ⚠️ 其中 {errors} 次出现错误")
                print("👋 再见！")
                break

            if user_input.lower() == 'log':
                log = get_audit_log()
                if not log:
                    print("📋 暂无操作记录")
                else:
                    print(f"\n📋 审计日志（共 {len(log)} 条）:")
                    for entry in log[-10:]:  # 最近 10 条
                        status = "❌" if entry["has_error"] else "✅"
                        print(f"  {status} [{entry['time'][:19]}] {entry['tool']}")
                print()
                continue

            if not user_input:
                continue

            await client.query(user_input)

            async for msg in client.receive_response():
                if isinstance(msg, AssistantMessage):
                    for block in msg.content:
                        if isinstance(block, TextBlock):
                            print(f"\n🤖 {block.text}")
                        elif isinstance(block, ToolUseBlock):
                            print(f"  🔧 {block.name}")
                elif isinstance(msg, ResultMessage):
                    if msg.total_cost_usd and msg.total_cost_usd > 0:
                        print(f"  💰 ${msg.total_cost_usd:.4f}")
            print()


if __name__ == "__main__":
    anyio.run(main)

# 1. 安装依赖
pip install claude-agent-sdk

# 2. 确保 API Key 已设置
export ANTHROPIC_API_KEY="sk-ant-你的密钥"

# 3. 运行
python main.py

你: 看看 Git 状态
你: 检查一下磁盘使用情况
你: 用 reviewer 审查最近的代码变更
你: 跑一下测试
你: log
你: quit

用户输入
  │
  ▼
主程序 (ClaudeSDKClient)
  │
  ├── 系统提示词（角色设定）
  │
  ├── Hook 层（安全管控）
  │   ├── PreToolUse: safety_gate → 拦截危险命令
  │   ├── PreToolUse: auto_approve_readonly → 自动批准只读
  │   └── PostToolUse: audit_logger → 审计日志
  │
  ├── 工具层
  │   ├── 内置工具: Bash, Read, Write, Edit, Grep, Glob
  │   └── 自定义工具: git_status, git_recent_commits,
  │                    service_health, disk_usage
  │
  └── Agent 层（角色协作）
      ├── reviewer → 代码审查
      ├── tester → 运行测试
      ├── log-analyzer → 日志分析
      └── deployer → 部署操作