SDK：可观测性 Agent
监控系统指标并自动响应
🌐 查看英文原文 | 源码 Notebook
import os
import shutil
import subprocess
from typing import Any
from dotenv import load_dotenv
from IPython.display import Markdown, display
from utils.agent_visualizer import (
display_agent_response,
print_activity,
reset_activity_context,
visualize_conversation,
)
from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient02 - 可观测性 Agent
在之前的 notebook 中，我们构建了一个基本的研究 Agent 和一个参谋长多 Agent 框架。虽然我们构建的 Agent 已经很强大，但它们的能力仍然有限：网络搜索 Agent 仅限于搜索互联网，而我们的参谋长 Agent 仅限于与其自身的文件系统交互。
这是一个严重的限制：现实世界中的 Agent 通常需要与数据库、API、文件系统和其他专业服务等其他系统交互。MCP（Model Context Protocol）是一个用于 AI 工具集成的开源标准，允许我们的 Agent 与这些外部系统之间轻松连接。在本 notebook 中，我们将探索如何将 MCP 服务器连接到我们的 Agent。
需要更多关于 MCP 的详细信息？有关全面的设置说明、配置最佳实践和故障排除技巧，请参阅 Claude Code MCP 文档。
MCP 服务器介绍
1. Git MCP 服务器
让我们首先赋予我们的 Agent 理解和使用 Git 仓库的能力。通过将 Git MCP 服务器添加到我们的 Agent 中，它获得了 13 个 Git 专用工具的访问权限，可以检查提交历史、查看文件变更、创建分支，甚至进行提交。这将我们的 Agent 从被动观察者转变为开发工作流中的主动参与者。在此示例中，我们将配置 Agent 仅使用 Git 工具来探索仓库的历史。这很简单，但知道这一点后，不难想象 Agent 可以自动创建 pull request、分析代码演化模式，或帮助管理跨多个仓库的复杂 Git 工作流。
# Get the git repository root (mcp_server_git requires a valid git repo path)
# os.getcwd() may return a subdirectory, so we find the actual repo root
git_executable = shutil.which("git")
if git_executable is None:
raise RuntimeError("Git executable not found in PATH")
git_repo_root = subprocess.run( # noqa: S603
[git_executable, "rev-parse", "--show-toplevel"],
capture_output=True,
text=True,
check=True,
).stdout.strip()
# Define our git MCP server (installed via uv sync from pyproject.toml)
git_mcp: dict[str, Any] = {
"git": {
"command": "uv",
"args": ["run", "python", "-m", "mcp_server_git", "--repository", git_repo_root],
}
}messages = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model="claude-opus-4-6",
mcp_servers=git_mcp,
allowed_tools=["mcp__git"],
# disallowed_tools ensures the agent ONLY uses MCP tools, not Bash with git commands
disallowed_tools=["Bash", "Task", "WebSearch", "WebFetch"],
permission_mode="acceptEdits",
)
) as agent:
await agent.query(
"Explore this repo's git history and provide a brief summary of recent activity."
)
async for msg in agent.receive_response():
print_activity(msg)
messages.append(msg)🤖 Using: mcp__git__git_log()
🤖 Using: mcp__git__git_status()
🤖 Using: mcp__git__git_branch()
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: mcp__git__git_log()
🤖 Using: mcp__git__git_status()
🤖 Using: mcp__git__git_branch()
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...display(Markdown(f"\nResult:\n{messages[-1].result}"))<IPython.core.display.Markdown object>2. GitHub MCP 服务器
现在让我们从本地 Git 操作升级到完整的 GitHub 平台集成。通过切换到 官方 GitHub MCP 服务器，我们的 Agent 获得了超过 100 个与 GitHub 整个生态系统交互的工具——从管理 issues 和 pull request 到监控 CI/CD 工作流和分析代码安全告警。该服务器可以与公共和私有仓库配合使用，使你的 Agent 能够自动化通常需要多个手动步骤的复杂 GitHub 工作流。
步骤 1：设置你的 GitHub Token
你需要一个 GitHub Personal Access Token。在此处获取一个，并放入 .env 文件中：GITHUB_TOKEN="<token>" > 注意：获取 token 时，选择”Fine-grained” token 并使用默认选项（即公共仓库，无账户权限），这将是让此演示运行的最简单方式。
此外，在此示例中，你的机器上需要运行 Docker。需要 Docker 是因为 GitHub MCP 服务器在容器化环境中运行，以实现安全性和隔离性。
Docker 快速设置： - 从 docker.com 安装 Docker Desktop - 确保 Docker 正在运行（你会在系统托盘中看到 Docker 图标） - 在终端中使用 docker --version 验证 - 故障排除： 如果 Docker 无法启动，请检查 BIOS 中是否启用了虚拟化。有关详细的设置说明，请参阅 Docker 文档
步骤 2：定义 MCP 服务器并启动 Agent 循环！
# define our github mcp server
load_dotenv(override=True)
github_mcp: dict[str, Any] = {
"github": {
"command": "docker",
"args": [
"run",
"-i",
"--rm",
"-e",
"GITHUB_PERSONAL_ACCESS_TOKEN",
"ghcr.io/github/github-mcp-server",
],
"env": {"GITHUB_PERSONAL_ACCESS_TOKEN": os.environ.get("GITHUB_TOKEN")},
}
}# run our agent
messages = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model="claude-opus-4-6",
mcp_servers=github_mcp,
allowed_tools=["mcp__github"],
# disallowed_tools ensures the agent ONLY uses MCP tools, not Bash with gh CLI
disallowed_tools=["Bash", "Task", "WebSearch", "WebFetch"],
permission_mode="acceptEdits",
)
) as agent:
await agent.query(
"Search for the anthropics/claude-agent-sdk-python repository and give me a few key facts about it."
)
async for msg in agent.receive_response():
print_activity(msg)
messages.append(msg)🤖 Using: mcp__github__search_repositories()
✓ Tool completed
🤖 Thinking...display(Markdown(f"\nResult:\n{messages[-1].result}"))<IPython.core.display.Markdown object>实际用例：可观测性 Agent
现在，通过如此简单的设置，我们就已经有了一个充当自愈软件系统的 Agent！
load_dotenv(override=True)
prompt = """Analyze the CI health for facebook/react repository.
Examine the most recent runs of the 'CI' workflow and provide:
1. Current status and what triggered the run (push, PR, schedule, etc.)
2. If failing: identify the specific failing jobs/tests and assess severity
3. If passing: note any concerning patterns (long duration, flaky history)
4. Recommended actions with priority (critical/high/medium/low)
Provide a concise operational summary suitable for an on-call engineer.
Do not create issues or PRs - this is a read-only analysis."""
github_mcp: dict[str, Any] = {
"github": {
"command": "docker",
"args": [
"run",
"-i",
"--rm",
"-e",
"GITHUB_PERSONAL_ACCESS_TOKEN",
"ghcr.io/github/github-mcp-server",
],
"env": {"GITHUB_PERSONAL_ACCESS_TOKEN": os.environ.get("GITHUB_TOKEN")},
}
}
messages = []
async with ClaudeSDKClient(
options=ClaudeAgentOptions(
model="claude-opus-4-6",
mcp_servers=github_mcp,
allowed_tools=["mcp__github"],
# IMPORTANT: disallowed_tools is required to actually RESTRICT tool usage.
# Without this, allowed_tools only controls permission prompting, not availability.
# The agent would still have access to Bash (and could use `gh` CLI instead of MCP).
disallowed_tools=["Bash", "Task", "WebSearch", "WebFetch"],
permission_mode="acceptEdits",
)
) as agent:
await agent.query(prompt)
async for msg in agent.receive_response():
print_activity(msg)
messages.append(msg)🤖 Using: mcp__github__get_file_contents()
🤖 Using: mcp__github__list_commits()
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: mcp__github__get_file_contents()
🤖 Using: mcp__github__get_file_contents()
🤖 Using: mcp__github__list_pull_requests()
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: mcp__github__get_commit()
🤖 Using: mcp__github__get_commit()
🤖 Using: mcp__github__get_commit()
🤖 Using: mcp__github__pull_request_read()
🤖 Using: mcp__github__pull_request_read()
🤖 Using: mcp__github__pull_request_read()
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: mcp__github__pull_request_read()
🤖 Using: mcp__github__pull_request_read()
🤖 Using: mcp__github__pull_request_read()
🤖 Using: mcp__github__pull_request_read()
🤖 Using: mcp__github__pull_request_read()
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: mcp__github__search_issues()
🤖 Using: mcp__github__search_issues()
✓ Tool completed
✓ Tool completed
🤖 Thinking...display(Markdown(f"\nResult:\n{messages[-1].result}"))<IPython.core.display.Markdown object>reset_activity_context()
visualize_conversation(messages)<IPython.core.display.HTML object>reset_activity_context()
display_agent_response(messages)<IPython.core.display.HTML object>可观测性 Agent 作为模块
observability_agent/agent.py 模块将可观测性模式封装为可复用的 send_query 函数。它在内部导入并使用来自 utils.agent_visualizer 的共享可视化工具： - reset_activity_context()：在每次查询开始时自动调用 - print_activity()：在执行期间提供实时反馈 - display_agent_response()：渲染最终结果（由 display_result 参数控制）
这意味着你可以用最少的代码使用该模块：
# Reload the module to pick up any changes (useful during development)
from observability_agent.agent import send_query
# The module handles activity display, context reset, and result visualization internally
result = await send_query(
"Check the CI status for the last 2 runs in anthropics/claude-agent-sdk-python. Just do 3 tool calls, be efficient."
)🤖 Using: mcp__github__list_commits()
✓ Tool completed
🤖 Using: mcp__github__get_commit()
🤖 Using: mcp__github__get_commit()
✓ Tool completed
✓ Tool completed
🤖 Thinking...<IPython.core.display.HTML object>多轮对话可以无缝工作——只需传入 continue_conversation=True：
# Example 2: Multi-turn conversation for deeper monitoring
result1 = await send_query("What's the current CI status for facebook/react?")🤖 Using: mcp__github__list_pull_requests()
🤖 Using: mcp__github__list_commits()
✓ Tool completed
✓ Tool completed
🤖 Thinking...
🤖 Using: mcp__github__pull_request_read()
🤖 Using: mcp__github__pull_request_read()
🤖 Using: mcp__github__pull_request_read()
🤖 Using: mcp__github__get_commit()
✓ Tool completed
✓ Tool completed
✓ Tool completed
✓ Tool completed
🤖 Thinking...<IPython.core.display.HTML object># Continue the conversation to dig deeper
result2 = await send_query(
"Are there any flaky tests in the recent failures? You can only make one tool call.",
continue_conversation=True,
)🤖 Using: mcp__github__search_issues()
✓ Tool completed
🤖 Thinking...<IPython.core.display.HTML object>总结
我们演示了 Claude Code SDK 如何通过 Model Context Protocol (MCP) 实现与外部系统的无缝集成。从通过 Git MCP 服务器进行本地 Git 操作开始，我们逐步扩展到完整的 GitHub 平台集成，可以访问超过 100 个 GitHub 专用工具。这将我们的 Agent 从本地助手转变为一个强大的可观测性系统，能够监控工作流、分析 CI/CD 故障，并为生产系统提供可操作的洞察。
通过将 MCP 服务器连接到我们的 Agent，我们创建了一个自主的可观测性系统，可以监控 GitHub Actions 工作流，区分真实故障和安全限制，并提供对测试失败的详细分析。该系统展示了 Agent 如何积极参与你的 DevOps 工作流，从被动监控转向智能事件响应。
到目前为止，我们的 Claude Code SDK 教程系列之旅就结束了。我们从简单的研究 Agent 进阶到复杂的多 Agent 编排，最后到通过 MCP 进行外部系统集成。这些模式共同为构建能够处理现实世界复杂性同时保持治理、合规性和可观测性的生产就绪 Agent 系统提供了基础。
你在所有 Notebook 中学到的内容
来自 Notebook 00（研究 Agent） - 使用 query() 和 ClaudeSDKClient 的 SDK 基础知识 - 使用 WebSearch 和 Read 的基本工具用法 - 简单的 Agent 循环和对话管理
来自 Notebook 01（参谋长） - 高级功能：记忆、输出样式、规划模式 - 通过子 Agent 实现多 Agent 协调 - 通过 hooks 和自定义命令实现治理 - 企业级 Agent 架构
来自 Notebook 02（可观测性 Agent） - 通过 MCP 服务器实现外部系统集成 - 实时监控和事件响应 - 生产工作流自动化 - 可扩展的 Agent 部署模式
所有三个 Agent 的完整实现可在各自的目录中找到（research_agent/、chief_of_staff_agent/、observability_agent/），可作为集成到你的生产系统中的灵感来源。
中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。