第 4 课：自定义工具——MCP Server

内置工具能做的：
  ✅ 执行 shell 命令
  ✅ 读写文件

你需要的：
  ❌ 调用公司内部 API
  ❌ 查询数据库
  ❌ 操作 Docker 容器
  ❌ 读取 Kubernetes Pod 状态
  ❌ 发送告警通知

from claude_agent_sdk import tool

@tool("工具名", "工具描述", {"参数名": 参数类型})
async def 你的函数(args):
    # 你的逻辑
    return {"content": [{"type": "text", "text": "返回结果"}]}

from claude_agent_sdk import tool
import subprocess

@tool("git_status", "查看 Git 仓库状态", {})
async def git_status(args):
    """查看当前仓库的 git status"""
    result = subprocess.run(
        ["git", "status", "--porcelain"],
        capture_output=True, text=True
    )
    output = result.stdout if result.stdout else "工作区是干净的"
    return {"content": [{"type": "text", "text": output}]}


@tool("git_log", "查看最近的提交记录", {"count": int})
async def git_log(args):
    """查看最近 N 条提交记录"""
    count = args.get("count", 5)
    result = subprocess.run(
        ["git", "log", f"--oneline", f"-{count}"],
        capture_output=True, text=True
    )
    return {"content": [{"type": "text", "text": result.stdout}]}


@tool("git_diff", "查看文件改动", {"file": str})
async def git_diff(args):
    """查看指定文件的 diff"""
    file_path = args.get("file", "")
    cmd = ["git", "diff", file_path] if file_path else ["git", "diff"]
    result = subprocess.run(cmd, capture_output=True, text=True)
    return {"content": [{"type": "text", "text": result.stdout or "没有改动"}]}

from claude_agent_sdk import create_sdk_mcp_server

# 把工具打包成一个 MCP 服务器
git_server = create_sdk_mcp_server(
    name="git-tools",
    version="1.0.0",
    tools=[git_status, git_log, git_diff],
)

from claude_agent_sdk import query, ClaudeAgentOptions

options = ClaudeAgentOptions(
    mcp_servers={"git": git_server},  # 注册 MCP 服务器
    allowed_tools=[
        "mcp__git__git_status",   # 格式：mcp__服务器名__工具名
        "mcp__git__git_log",
        "mcp__git__git_diff",
    ],
    system_prompt="你是 Git 助手，帮用户管理代码仓库。",
)

async for msg in query(
    prompt="看看仓库状态，最近 3 条提交，以及有什么改动",
    options=options,
):
    # 处理消息...
    pass

mcp__<服务器名>__<工具名>

例子：
  mcp__git__git_status
  mcp__git__git_log
  mcp__docker__list_containers

@tool("list_containers", "列出 Docker 容器", {"all": bool})
async def list_containers(args):
    show_all = args.get("all", False)
    cmd = ["docker", "ps"]
    if show_all:
        cmd.append("-a")
    result = subprocess.run(cmd, capture_output=True, text=True)
    return {"content": [{"type": "text", "text": result.stdout}]}


@tool("container_logs", "查看容器日志", {"container": str, "lines": int})
async def container_logs(args):
    container = args["container"]
    lines = args.get("lines", 50)
    result = subprocess.run(
        ["docker", "logs", "--tail", str(lines), container],
        capture_output=True, text=True
    )
    output = result.stdout + result.stderr
    return {"content": [{"type": "text", "text": output}]}

import urllib.request
import json

@tool("health_check", "检查服务健康状态", {"url": str})
async def health_check(args):
    url = args["url"]
    try:
        req = urllib.request.urlopen(url, timeout=5)
        status = req.getcode()
        return {"content": [{"type": "text",
            "text": f"✅ {url} → HTTP {status}"}]}
    except Exception as e:
        return {"content": [{"type": "text",
            "text": f"❌ {url} → 失败: {str(e)}"}],
            "is_error": True}

@tool("send_alert", "发送告警通知", {"message": str, "level": str})
async def send_alert(args):
    message = args["message"]
    level = args.get("level", "info")

    # 这里可以接入你的告警系统：飞书、钉钉、Slack 等
    print(f"[{level.upper()}] 告警: {message}")

    return {"content": [{"type": "text",
        "text": f"告警已发送: [{level}] {message}"}]}

# 应用级别的状态
deployment_history = []

@tool("record_deployment", "记录部署", {"version": str, "env": str})
async def record_deployment(args):
    # 直接访问应用状态！
    deployment_history.append({
        "version": args["version"],
        "env": args["env"],
        "time": datetime.now().isoformat(),
    })
    return {"content": [{"type": "text",
        "text": f"已记录部署: {args['version']} → {args['env']}"}]}

@tool("list_deployments", "查看部署历史", {})
async def list_deployments(args):
    if not deployment_history:
        return {"content": [{"type": "text", "text": "没有部署记录"}]}
    history = "\n".join(
        f"  {d['version']} → {d['env']} @ {d['time']}"
        for d in deployment_history
    )
    return {"content": [{"type": "text", "text": history}]}