SRE代理MCP服务器

SRE代理MCP服务器

目录

• 问题
• SRE代理做什么
• 演示
• 架构与系统设计
• 项目结构
• 工人
• MCP接口
• 数据集场景
• 安装说明
• 环境变量
• 运行服务器
• 从MCP客户端连接
• 示例查询
• 测试
• 故障排除
• 已实施强制性功能
• 已实现自定义功能
• 贡献
• 许可证

问题

生产事件响应是软件工程中压力最大、时间最敏感的活动之一。当服务在凌晨3点停止时，会呼叫一名随叫随到的现场可靠性工程师（SRE），他必须在极端的时间压力下回答一系列问题：

• 什么东西真的坏了？ 是单个端点、整个服务，还是跨多个系统的级联故障？
• 发生了什么变化？ 最近是否有与事件开始相关的部署、配置更改或基础设施事件？
• 证据在哪里？ 日志、度量和跟踪分散在不同的可观察性平台上。工程师必须逐一查询每个时间点，在脑海中关联时间戳，并拼凑出一个时间线。
• runbook上写着什么？ 大多数团队都为已知的故障模式维护运行手册，但找到正确的运行手册并遵循其步骤，同时进行调查，在认知上要求很高。
• 解决办法是什么？ 一旦确定了根本原因，工程师必须决定补救措施：回滚部署、扩展基础设施、切换功能标志或应用修补程序。每种选择都有风险。

这个过程很慢，容易出错，而且精神疲惫。研究表明，整个行业生产事故的平均解决时间（MTTR）平均为1-4小时，其中很大一部分时间花在调查阶段，而不是实际修复上。在停机期间，每一分钟都要花钱：收入损失、SLA违规、客户流失和工程生产力流失。

核心挑战在于 事件分类基本上是一个多步骤的推理任务 这需要从多个来源收集证据，形成假设，根据数据对其进行测试，并得出根本原因。如今，这种推理完全发生在工程师的头脑中，没有结构化的框架来指导调查或防止导致错误结论的认知捷径。

为什么现有工具不足

当前的事件响应工具解决了难题的各个部分，但没有解决调查工作流程本身：

• 观测平台 （Datadog、Grafana、New Relic）擅长存储和可视化数据，但他们不会推理数据的含义或指导调查。
• 警报系统 （PagerDuty、OpsGenie）通知合适的人，但在初始警报上下文之外不提供调查支持。
• Runbook工具 （Confluence、Notion）存储机构知识，但没有自动的方法将症状与正确的runbook相匹配，也没有自动的方式验证其步骤是否得到遵循。
• 聊天机器人 （自定义Slack机器人）可以查询单个数据源，但缺乏计划结构化调查或跨源综合调查结果的能力。

缺少的是 智能编排层 它可以对症状进行自然语言描述，计划结构化调查，通过以正确的顺序查询正确的工具来执行调查，并将结果综合成具有补救建议的可操作事件报告。

SRE代理做什么

SRE代理是一个特定于域的MCP服务器，充当 人工智能驱动的第一响应者 生产事故。它通过自动化SRE目前手动执行的调查工作流程，弥合了原始可观察性数据和可操作事件响应之间的差距。

当随叫随到的工程师报告症状时（例如。 “自14:05 UTC以来，Prod结账时间飙升了500秒”)，服务器：

1. 计划调查 --LLM驱动的编排器分析症状，生成排名假设（例如“部署不良”、“依赖失败”、“资源耗尽”），并生成一个结构化的分流计划，其中包含分阶段的任务和依赖顺序。

2. 并行执行调查 --专业人员同时查询日志、指标、部署记录和运行手册。同阶段任务并行运行以提高速度；跨阶段依赖关系是自动执行的。

3. 将调查结果综合到事件报告中 --工人输出被汇总和分析，以产生根本原因评估、证据总结、严重性分类和建议的下一步行动。

4. 在人类批准的情况下提出补救措施 --系统生成补丁工件（例如回滚差异、runbook更新），但从不自动应用它们。人工审批门确保工程师在任何更改生效之前进行审查并明确批准。

结果：一个通常会花30-60分钟查询仪表板、阅读日志和交叉引用部署历史的工程师可以在不到一分钟的时间内获得一份结构化的事件报告，其中包含根本原因分析和补救选项。

用例

• 随叫随到的分流加速 --通过在生产事故期间自动化证据收集和关联来降低MTTR。
• 事故尸检准备 --生成结构化的时间线和证据摘要，直接输入尸检文件。
• Runbook验证 --自动将事件症状与相关运行手册相匹配，并验证补救步骤是否适用。
• 初级SRE培训 --结构化的分流计划是一种教育工具，向经验不足的工程师展示如何系统地调查停机。
• 人工智能辅助配对调试 --使用MCP客户端交互式地逐步探索事件，代理在每个阶段提供上下文和建议。

演示

demo_sre_agent.mp4

架构与系统设计

该系统遵循Clean Architecture，具有明确的组合根，确保业务逻辑与基础设施问题和MCP协议边界解耦。

分层

端口和适配器

依赖注入接口在 mcp_server/src/app/ports.py:

• LanguageModelPort — generate_json() 和 generate_text() LLM电话
• SreToolGatewayPort --日志、指标、部署、运行手册和补丁操作
• ApprovalStorePort --创建、批准和应用补丁建议

混凝土适配器 mcp_server/src/infrastructure/gateways.py:

• GeminiLanguageModelGateway --具有结构化输出的Google Gemini
• SreGuardianToolGateway --由当地人支持 mcp_tools/sre_guardian
• PatchApprovalStoreGateway --内存中批准存储

工作流阶段

1. 开始 --接受用户查询并初始化进度跟踪。
2. 编排器 --构建一个 IncidentTriagePlan 包含假设、任务、依赖关系和阶段编号。使用Gemini结构化输出或确定性回退。
3. 工人执行 --按阶段对任务进行分组。通过并行运行相同阶段的任务 asyncio.gather().强制执行 depends_on 在下游任务之前。
4. 合成 --将工人的输出汇总到一份结构化的事故报告中，其中包含根本原因评估、证据和下一步行动。创建补丁建议。
5. 审批门 --补丁提案仍有待人类明确批准。如果获得批准，则应用补丁；否则，它将保持未应用状态。
6. 完成 --返回最终的结构化有效载荷，包括计划、每个任务的结果、综合输出和批准状态。

项目结构

.
├── mcp_server/
│   └── src/
│       ├── server.py                  # FastMCP entry point
│       ├── routers/                   # MCP tool/resource/prompt registration
│       │   ├── tools.py
│       │   ├── resources.py
│       │   └── prompts.py
│       ├── app/                       # Use-case logic
│       │   ├── ports.py               # Dependency injection interfaces
│       │   ├── orchestrator.py        # Query -> IncidentTriagePlan
│       │   ├── phase_runner.py        # Phased parallel task execution
│       │   ├── task_runner.py         # Single task dispatch
│       │   ├── synthesizer.py         # Findings -> report + patch
│       │   └── worker_registry.py     # Tool name -> worker mapping
│       ├── infrastructure/
│       │   └── gateways.py            # Gemini, SRE tool, approval adapters
│       ├── entities/                  # Pydantic domain models
│       │   ├── incident_triage_plan.py
│       │   ├── tasks.py
│       │   ├── workers.py
│       │   ├── hypothesis.py
│       │   ├── severity.py
│       │   └── ...
│       ├── nodes/                     # Concrete worker implementations
│       │   ├── base.py                # BaseWorker abstract class
│       │   ├── logs_query_worker.py
│       │   ├── metrics_query_worker.py
│       │   ├── deploy_list_worker.py
│       │   ├── incident_context_worker.py
│       │   ├── runbooks_search_worker.py
│       │   ├── runbook_get_worker.py
│       │   └── patch_generate_worker.py
│       ├── tools/                     # Tool implementation layer
│       ├── workflows/
│       │   └── main_workflow.py       # Composition root
│       ├── mcp_tools/
│       │   └── sre_guardian.py        # Local fallback tool backends
│       ├── prompts/                   # LLM prompt templates
│       ├── resources/                 # MCP resource implementations
│       ├── config/
│       │   └── settings.py            # Environment-based config (Pydantic)
│       ├── models/
│       │   └── get_model.py           # Gemini client initialization
│       └── utils/
│           ├── approval_store.py      # In-memory patch proposals
│           ├── opik_utils.py          # Observability integration
│           └── ...
├── mcp_client/
│   └── src/
│       ├── client.py                  # Interactive MCP client
│       ├── settings.py                # Client configuration
│       └── utils/                     # Agent loop, LLM, parsing, commands
├── data/scenarios/                    # Sample incident data
│   ├── bad_deploy/
│   ├── dependency_latency/
│   └── saturation_db_pool/
├── docs/
├── pyproject.toml
├── .python-version
├── .env.sample
└── uv.lock

工人

所有工人继承自 BaseWorker 并实施 async run(task, context_id) -> WorkerResult.

MCP接口

工具

注册于 mcp_server/src/routers/tools.py:

• 上下文管理： incident_start_context, incident_get_context
• 可观察性： logs_query, metrics_query
• 部署： deploys_list
• 运行手册： runbooks_search, runbooks_get
• 修补： patch_generate
• 批准： request_patch_approval, set_patch_approval, apply_approved_patch, get_patch_approval_status
• 工作流程： process_user_query_workflow （主编排器入口点）

资源

• system://status --系统健康状况（CPU、正常运行时间）
• system://memory --内存使用统计数据

提示

• sre_triage_execution_prompt --事件分流的完整SRE说明。该提示将调查工具联系在一起，形成一个有指导的代理工作流程，并包括需要用户反馈和确认的步骤（例如，审查分流计划，在应用前批准补丁建议）。

数据集场景

样本事件数据集 data/scenarios/ 对于本地回退和演示：

每个场景包括 logs.jsonl, metrics.json, deploys.json，以及 runbooks.md。这些被消费 mcp_server/src/mcp_tools/sre_guardian.py 当启用本地回退时。

安装说明

先决条件

• Python 3.14+
• 紫外线 （快速Python包管理器）

安装

1. 克隆存储库：

git clone https://github.com/your-username/sre-agent.git
cd sre-agent

2. 安装依赖项：

uv sync

3. 根据示例创建环境文件：

cp .env.sample .env

4. 填写 .env 使用您的真实钥匙（请参见 环境变量 在......下面

环境变量

示例文件提供在 .env.sample。复制到 .env 并替换占位符值。

运行服务器

MCP服务器（stdio）

uv --directory mcp_server run -m src.server --transport stdio

MCP服务器（可流式传输HTTP）

uv --directory mcp_server run -m src.server --transport streamable-http --port 8001

交互式MCP客户端

内存服务器模式（MCP服务器在同一进程中运行）：

uv run python -m mcp_client.src.client

Stdio服务器模式（连接到外部MCP服务器）：

uv run python -m mcp_client.src.client --transport stdio

从MCP客户端连接

光标/通用MCP客户端

{
  "mcp_servers": {
    "sre-first-responder": {
      "command": "uv",
      "args": [
        "--directory",
        "/absolute/path/to/sre-agent/mcp_server",
        "run",
        "-m",
        "src.server",
        "--transport",
        "stdio"
      ],
      "env": {
        "GOOGLE_API_KEY": "your-google-api-key",
        "OPIK_API_KEY": "your-opik-api-key",
        "OPIK_PROJECT_NAME": "sre_guardian"
      }
    }
  }
}

克劳德桌面版

工作示例包含在 claude_desktop_config.json.

示例查询

Prod checkout is spiking 500s since 14:05 UTC. Start triage for checkout-api.
Start triage for checkout-api and check if a recent deploy caused regressions.
Investigate checkout latency and timeout errors in production.

测试

编译检查：

python -m compileall mcp_server/src mcp_client/src

工作流程烟雾测试：

python - <<'PY'
import asyncio
from mcp_server.src.workflows.main_workflow import process_user_query

async def main():
    result = await process_user_query(
        user_query='Prod checkout is spiking 500s since 14:05 UTC. Start triage for checkout-api.',
        render_output=False,
    )
    print(result['status'], result['task_count'], result['result_count'])

asyncio.run(main())
PY

故障排除

ModuleNotFoundError: No module named 'mcp_server' 使用基于模块的调用：

uv --directory mcp_server run -m src.server --transport stdio

Gemini配额错误(429 RESOURCE_EXHAUSTED) 这是双子座的限速。等待配额重置或通过设置启用脱机回退 ORCHESTRATOR_ENABLE_OFFLINE_FALLBACK=true 在你的 .env.

Opik登录到 Default Project 集 OPIK_PROJECT_NAME 在你的 .env 文件，并确保在创建跨度的过程中配置了Opik。

已实施强制性功能

1. Python中使用FastMCP的MCP服务器

   • 实施于 mcp_server/src/server.py。服务器是用FastMCP创建的，支持两者 stdio 和 streamable-http 运输。所有工具、资源和提示都通过路由器层在中注册 mcp_server/src/routers/.

2. 至少一个具有有意义功能的MCP工具

   • 在中实施并注册了多个MCP工具 mcp_server/src/routers/tools.py，包括 logs_query, metrics_query, deploys_list, runbooks_search, runbooks_get, patch_generate, incident_start_context, incident_get_context，审批工作流工具(request_patch_approval, set_patch_approval, apply_approved_patch, get_patch_approval_status)，以及主要 process_user_query_workflow 编排器入口点。

3. 至少一个带有用户反馈工作流的MCP提示

   • 这 sre_triage_execution_prompt 实施于 mcp_server/src/prompts/sre_instructions_prompt.py 并注册于 mcp_server/src/routers/prompts.py。此提示驱动了完整的代理分流工作流程，并包括明确的用户反馈步骤：人工审批门要求工程师审查生成的补丁提案，并在应用任何更改之前明确批准或拒绝它（通过 request_patch_approval -> set_patch_approval -> apply_approved_patch).

4. 发布在公共GitHub存储库上

   • 整个项目托管在GitHub上。

5. 已使用初始化 uv 用于依赖关系管理

   • 该项目使用 uv 随着 pyproject.toml, .python-version （3.14.0），以及 uv.lock依赖项是通过以下方式安装的 uv sync.

6. 有组织的源结构

   • 代码按以下方式组织 mcp_server/src/ 和 mcp_client/src/ 清晰的分隔： server.py （入口点）， routers/ （MCP注册）， tools/ （工具实现）， app/ （业务逻辑）， entities/ （域模型）， infrastructure/ （适配器）， config/settings.py （基于Pydantic的配置）， prompts/ （提示模板）， resources/ （资源实现），以及 utils/ （公用事业）。

7. 全面的README.md

   • 此自述文件包括：项目描述、用例、架构图、设置说明、环境变量文档、MCP客户端配置JSON、强制和自定义功能部分以及故障排除指南。

8. 没有提交API密钥或敏感凭据

   • A. .env.sample 提供具有占位符值的文件。这 .env 文件被标记为无效。README解释了需要哪些密钥以及如何获取它们。

已实现自定义功能

1. 具有自定义编排/规划的MCP客户端 （客户与集成）

   • mcp_client/src/client.py 实现了一个具有自定义代理循环的完全交互式MCP客户端(handle_agent_loop_utils.py)、工具调用编排、思维模式切换、传输模式选择（内存或stdio中）和对话历史管理。这演示了MCP服务器在Cursor或Claude Desktop之外的编程使用。

2. 用于上下文信息的MCP资源 （客户与集成）

   • 在中实现了两个MCP资源 mcp_server/src/resources/: system://status 提供实时系统健康数据（CPU、正常运行时间）和 system://memory 提供内存使用统计信息。这些已在 mcp_server/src/routers/resources.py 并让代理对其运行的环境有上下文感知。

3. 多种工作流模式（顺序、并行、有条件） （工作流和代理模式）

   • 相位运行器(mcp_server/src/app/phase_runner.py)实现了三种不同的模式： 顺序的 阶段排序（阶段1在阶段2开始之前完成）， 平行 每个阶段内的执行（所有相同阶段的任务通过以下方式并发运行 asyncio.gather())，以及 条件分支 通过依赖门(depends_on 当上游依赖失败时跳过下游任务的字段）。

4. 计划和执行代理模式 （工作流和代理模式）

   • 系统实现了完整的计划和执行模式： 编排器 (mcp_server/src/app/orchestrator.py)使用Gemini结构化输出生成 IncidentTriagePlan 包括排名假设、分阶段任务和依赖关系排序 相位转轮 然后执行该计划。这 合成器 (mcp_server/src/app/synthesizer.py)撰写最终事件报告。这个三阶段管道（计划->执行->综合）在组合根中连接在一起(mcp_server/src/workflows/main_workflow.py).

5. 人在循环验证 （人机交互与用户体验）

   • 该系统将补丁建议（例如回滚差异、runbook更新）作为结构化工件生成，并在应用之前明确等待人工批准。审批工作流使用 request_patch_approval 创建待处理提案， set_patch_approval 让人类接受或拒绝，以及 apply_approved_patch 只有在明确批准后才能申请。这超越了简单的提示反馈，实现了完整的“人工智能生成->差异预览->人工审查->门控应用程序”模式。

6. 多代理编排 （多代理系统）

   • 该系统实现了管理者-工作者模式，其中 编排器 充当管理者代理，分解问题并分配任务，以及 7名专业工人代理 (LogsQueryWorker, MetricsQueryWorker, DeploysListWorker, IncidentContextWorker, RunbooksSearchWorker, RunbookGetWorker, PatchGenerateWorker)合作调查事件的不同方面。这 工人登记处 (mcp_server/src/app/worker_registry.py)将工具名称映射到工作实例，以及 合成器 将所有工作人员的输出汇总到一个统一的报告中。

7. Opik的可观察性 （可观察性和评估）

   • MCP服务器(mcp_server/src/utils/opik_utils.py)以及MCP客户端(mcp_client/src/utils/opik_handler.py)与Opik集成，用于LLM调用跟踪、工具调用跟踪和代理行为分析。所有MCP工具都装饰有 @opik.track() 用于自动创建跨度。

8. 自定义域特定数据 （自定义数据）

   • 中提供了三个事件场景数据集 data/scenarios/: bad_deploy, dependency_latency，以及 saturation_db_pool每个场景都包含结构化日志(logs.jsonl)，时间序列度量(metrics.json)，部署记录(deploys.json)，以及Runbook(runbooks.md).这些被消费 mcp_server/src/mcp_tools/sre_guardian.py 为调查提供现实的、特定领域的事件数据。

9. Pydantic模型的结构化输出 （结构化输出）

   • 编排器使用Gemini的结构化输出模式和Pydantic响应模式来确保一致、有效的响应。主要型号包括 IncidentTriagePlan, Task, Hypothesis, Severity, WorkerResult，以及 WorkerStatus （所有定义见 mcp_server/src/entities/).这保证了LLM输出符合预期的形状，并且可以被下游组件可靠地消耗。

贡献

以明确的问题陈述、复制步骤和拟议的更改打开问题或PR。

许可证

该项目根据MIT许可证获得许可。看 许可证.