DevOps领域的MCP服务器：什么是以及本周即可搭建的两种类型
DevOps用MCP服务器：让Claude能够实时访问Kubernetes、AWS及运行手册。这三种服务器类型，以及只读型Kubernetes MCP服务器和常见错误。
来源: What Is an MCP Server and Why Every DevOps Team Should Have One in 2026
MCP，即模型上下文协议，是Anthropic推出的一项开放标准，它让人工智能模型能够以结构化的方式连接到外部系统：各类工具、API、数据库和服务。没有它，Claude就如同被限制在封闭空间中，所有信息都必须手动输入；而有了它，Claude可以直接向外部的系统发起请求。
MCP中的“服务器”其实只是一个小型程序。它负责向Claude提供工具和数据，处理数据之间的交互，并在数据所在的位置运行，无论是笔记本电脑、计算集群还是服务器均可。这并非复杂的基础设施，一个可用的MCP服务器甚至可以是只有50行的Python脚本。
一个有助于理解的类比
Slack集成是一个很好的对比例子。在引入这些集成之前，需要手动发布“已开始部署到生产环境”的消息；引入之后，CI/CD流水线会自动发布该消息，而Slack则成为掌握实时情况的中心平台。MCP也为Claude实现了类似功能：之前需要人工向它输入基础设施的相关信息，之后则可以通过直接连接自行查询。
为何DevOps团队受益最多
大多数MCP相关开发都是针对应用程序开发者的。而DevOps工程师、SRE工程师以及平台工程师能获得更大收益，因为他们的工作就是管理那些已经具备API的系统，比如Kubernetes、AWS、Datadog或Grafana、PagerDuty、Terraform Cloud、GitHub、Confluence和Notion。这些系统都可以通过MCP与Claude相连。一旦实现连接，就无需在各种工具之间来回切换来收集信息，工作方式转变为直接提出问题即可。
团队如何使用MCP服务器
- 集群相关问题。询问哪些生产环境部署已运行30天以上且未进行过更新时，Claude会连接到Kubernetes MCP服务器，该服务器查询集群后返回答案。无需使用kubectl，也无需复制粘贴。
- 成本相关问题。询问本周与上周相比哪些AWS服务的使用量激增时，Claude会连接到AWS服务器及Cost Explorer，随后以通俗易懂的英文形式给出分析结果。
- 事件背景信息。询问过去24小时内PagerDuty记录的事件以及是否有未处理的事件时，系统会从PagerDuty服务器获取当前状态并进行汇总。
- 运行手册查询。询问关于数据库连接池耗尽的解决方案时，系统会通过Confluence服务器在知识库中搜索并返回对应的文档。
三种类型的服务器
通常，团队最终会构建三种类型的服务器。
只读监控服务器
这类服务器将Claude与监控及基础设施API相连，使其能够描述当前的状态。典型例子包括Kubernetes集群状态服务器、AWS资产清单服务器、Datadog指标服务器以及CloudWatch日志服务器。由于Claude仅能读取数据，因此风险极低。这是最佳的起点，在授予任何写入权限之前可先建立信任。
知识库服务器
这些链接将 Claude 与用于搜索和检索的内部文档相连：Confluence 或 Notion 的操作手册服务器、GitHub Wiki 服务器、本地 Markdown 文件夹，或是架构决策记录的存储库。由于是对文档进行搜索，其风险实际上为零。它们之所以重要，是因为能让 Claude 产生仿佛真正了解公司的感觉。
操作服务器
这类服务器能让 Claude 执行创建工单、触发流水线或重启服务等操作。示例包括创建 Jira 工单、管理 PagerDuty 事件、创建 GitHub pull request 以及发送 Slack 通知。由于 Claude 现在正在修改某些内容，其风险属于中等到高水平。所有写操作都应设置审批流程，因此没有确认步骤的话，绝不能重启容器。
构建 Kubernetes MCP 服务器
以下是一个用 Python 编写的可运行的只读 Kubernetes 服务器。请先安装 SDK：
pip install mcp kubernetes --break-system-packages
然后将服务器文件保存为 k8s_mcp_server.py：
import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
from kubernetes import client, config as k8s_config
# Load kubeconfig
try:
k8s_config.load_incluster_config()
except:
k8s_config.load_kube_config()
v1 = client.CoreV1Api()
apps_v1 = client.AppsV1Api()
# Create MCP server
app = Server("k8s-devops-server")
@app.list_tools()
async def list_tools() -> list[types.Tool]:
"""Tell Claude what tools are available."""
return [
types.Tool(
name="get_pods",
description="Get all pods in a namespace with their status, restarts, and age.",
inputSchema={
"type": "object",
"properties": {
"namespace": {
"type": "string",
"description": "Kubernetes namespace (e.g. production, staging)"
}
},
"required": ["namespace"]
}
),
types.Tool(
name="get_pod_logs",
description="Get recent logs from a pod or deployment.",
inputSchema={
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Pod or deployment name"
},
"namespace": {
"type": "string",
"description": "Kubernetes namespace"
},
"lines": {
"type": "integer",
"description": "Number of log lines to return (default 50)"
}
},
"required": ["name", "namespace"]
}
),
types.Tool(
name="get_deployments",
description="List all deployments with replica counts and status.",
inputSchema={
"type": "object",
"properties": {
"namespace": {
"type": "string",
"description": "Kubernetes namespace"
}
},
"required": ["namespace"]
}
),
types.Tool(
name="get_events",
description="Get recent Kubernetes events. Useful for diagnosing issues.",
inputSchema={
"type": "object",
"properties": {
"namespace": {
"type": "string",
"description": "Kubernetes namespace"
}
},
"required": ["namespace"]
}
),
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
"""Handle tool calls from Claude."""
if name == "get_pods":
namespace = arguments["namespace"]
pods = v1.list_namespaced_pod(namespace=namespace)
lines = [f"Pods in {namespace}:\n"]
for pod in pods.items:
phase = pod.status.phase
restarts = sum(
cs.restart_count
for cs in (pod.status.container_statuses or [])
)
name_str = pod.metadata.name
lines.append(f" {name_str}: {phase} | restarts: {restarts}")
return [types.TextContent(type="text", text="\n".join(lines))]
elif name == "get_pod_logs":
pod_name = arguments["name"]
namespace = arguments["namespace"]
lines = arguments.get("lines", 50)
import subprocess
result = subprocess.run(
f"kubectl logs {pod_name} -n {namespace} --tail={lines}",
shell=True, capture_output=True, text=True
)
output = result.stdout or result.stderr or "No logs found"
return [types.TextContent(type="text", text=output)]
elif name == "get_deployments":
namespace = arguments["namespace"]
deployments = apps_v1.list_namespaced_deployment(namespace=namespace)
lines = [f"Deployments in {namespace}:\n"]
for d in deployments.items:
desired = d.spec.replicas or 0
available = d.status.available_replicas or 0
status = "✅" if desired == available else "⚠️"
lines.append(
f" {status} {d.metadata.name}: {available}/{desired} replicas"
)
return [types.TextContent(type="text", text="\n".join(lines))]
elif name == "get_events":
namespace = arguments["namespace"]
events = v1.list_namespaced_event(namespace=namespace)
# Sort by last timestamp, show recent ones
sorted_events = sorted(
events.items,
key=lambda e: e.last_timestamp or e.event_time,
reverse=True
)[:20]
lines = [f"Recent events in {namespace}:\n"]
for e in sorted_events:
lines.append(
f" [{e.type}] {e.reason}: {e.message[:100]}"
)
return [types.TextContent(type="text", text="\n".join(lines))]
return [types.TextContent(type="text", text=f"Unknown tool: {name}")]
async def main():
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
要将其连接到 Claude Desktop，需编辑配置文件，该文件位于以下路径之一：
Mac: ~/Library/Application Support/Claude/claude_desktop_config.json
Windows: %APPDATA%\Claude\claude_desktop_config.json
然后添加服务器条目：
{
"mcpServers": {
"kubernetes": {
"command": "python",
"args": ["/path/to/k8s_mcp_server.py"]
}
}
}
重启 Claude Desktop 后即可使用。当询问当前生产环境中运行了多少个 Pod 时，Claude 会实际进行查询。
构建运行手册服务器
这个功能可能更有价值。它能让 Claude 直接访问团队的运行手册，这样在发生故障时，Claude 可以直接找到对应的手册，而无需有人手动在 Confluence 中搜索。
import asyncio
import os
import glob
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
app = Server("runbook-server")
RUNBOOKS_PATH = os.path.expanduser("~/runbooks") # point to your runbooks folder
def load_all_runbooks() -> dict[str, str]:
"""Load all markdown runbooks from the folder."""
runbooks = {}
for filepath in glob.glob(f"{RUNBOOKS_PATH}/**/*.md", recursive=True):
with open(filepath, 'r', errors='replace') as f:
content = f.read()
name = os.path.relpath(filepath, RUNBOOKS_PATH)
runbooks[name] = content
return runbooks
@app.list_tools()
async def list_tools() -> list[types.Tool]:
return [
types.Tool(
name="search_runbooks",
description=(
"Search your team's runbooks for a specific topic, service, or error. "
"Use this when someone asks about a procedure, a known issue, "
"or how to fix something."
),
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "What to search for (e.g. 'database connection pool', 'nginx 502', 'SSL cert renewal')"
}
},
"required": ["query"]
}
),
types.Tool(
name="list_runbooks",
description="List all available runbooks by category.",
inputSchema={
"type": "object",
"properties": {}
}
),
types.Tool(
name="get_runbook",
description="Get the full content of a specific runbook by name.",
inputSchema={
"type": "object",
"properties": {
"name": {
"type": "string",
"description": "Runbook filename (from list_runbooks)"
}
},
"required": ["name"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
runbooks = load_all_runbooks()
if name == "list_runbooks":
names = sorted(runbooks.keys())
output = "Available runbooks:\n" + "\n".join(f" - {n}" for n in names)
return [types.TextContent(type="text", text=output)]
elif name == "search_runbooks":
query = arguments["query"].lower()
matches = []
for rb_name, content in runbooks.items():
if query in content.lower() or query in rb_name.lower():
# Return first 500 chars as preview
preview = content[:500].strip()
matches.append(f"📄 {rb_name}\n{preview}\n...")
if not matches:
return [types.TextContent(
type="text",
text=f"No runbooks found for '{query}'"
)]
return [types.TextContent(
type="text",
text=f"Found {len(matches)} runbook(s):\n\n" + "\n\n---\n\n".join(matches)
)]
elif name == "get_runbook":
rb_name = arguments["name"]
content = runbooks.get(rb_name)
if not content:
return [types.TextContent(type="text", text=f"Runbook not found: {rb_name}")]
return [types.TextContent(type="text", text=content)]
return [types.TextContent(type="text", text=f"Unknown tool: {name}")]
async def main():
async with stdio_server() as (read_stream, write_stream):
await app.run(read_stream, write_stream, app.create_initialization_options())
if __name__ == "__main__":
asyncio.run(main())
将其与其他服务器一起添加到 Claude Desktop 的配置中：
{
"mcpServers": {
"kubernetes": {
"command": "python",
"args": ["/path/to/k8s_mcp_server.py"]
},
"runbooks": {
"command": "python",
"args": ["/path/to/runbook_mcp_server.py"]
}
}
}
之后，当有人询问“API容器内存使用量过高的处理方案是什么？”时，系统返回的会是团队自己制定的处理方案，而非通用的搜索结果。
已存在的服务器
并非所有东西都需要从零开始构建。GitHub上的modelcontextprotocol组织提供了官方的参考服务器集合，涵盖GitHub（仓库、拉取请求、问题与提交记录）、Postgres、Slack、AWS知识库搜索、本地文件系统访问以及Brave网页搜索功能。社区还开发了用于Datadog（指标、仪表板与监控功能）、PagerDuty（事件与警报功能）以及Jira（工单与迭代计划）的服务器。首先查看这个集合是值得的，因为可能已经有人构建好了所需的功能。
应避免的错误
- 过早授予写入权限。应先从只读模式开始，待建立信任后再逐个添加写入工具。
- 构建单一大型服务器。小型、专注的服务器更易于维护、调试，也便于分配给特定人员。
k8s-server仅存放Kubernetes相关工具，aws-server仅存放AWS相关工具，runbook-server仅存放文档。 - 忽略日志记录。需记录每一次工具调用，包括Claude查询了什么、何时查询以及得到了什么结果，尤其是在添加写入工具之前。
- 跳过写入操作的审批流程。任何会创建工单、重启服务或合并拉取请求的操作都应经过确认步骤，且Claude需先说明其即将执行的操作内容。
合理配置下的高效早晨
一次典型的会话会从询问生产环境的状态开始。Claude会检查Kubernetes服务器，并汇报部署任务和容器的运行情况，标记出反复重启的容器，随后针对这些问题进行进一步处理。
You: What's the state of production right now?
Claude: [checks kubernetes MCP server]
3 deployments all healthy, 47/47 pods running.
One pod in payments namespace has 8 restarts - api-worker-6d4f9b.
Everything else looks stable.
You: Why is that pod restarting?
Claude: [checks pod logs via MCP]
Logs show OOMKilled. Memory limit is 256Mi but it's hitting
248Mi regularly. Looks like it needs a bump to at least 512Mi.
Want me to check if we have a runbook for this?
You: Yes
Claude: [searches runbook MCP server]
Found it - "memory-limit-tuning.md"
The runbook says: increase limit in Helm values,
run helm upgrade, monitor for 10 minutes.
Last updated by Rahul on March 12th.
You: What did our AWS bill look like this week?
Claude: [checks AWS MCP server]
$4,823 this week vs $4,210 last week - up 14.5%.
Spike is from EC2 in us-east-1. Looks like
3 r6g.2xlarge instances were added on Tuesday.
整个操作过程大约需要90秒，无需切换标签页、复制粘贴或搜索。
从本周开始
同时构建五台服务器是个错误。请选择一台。对于大多数DevOps团队而言，最佳的首选是运行手册服务器，因为它为只读模式，能立即发挥作用，并且可以与现有的任何文档配合使用。设置步骤如下：
- 将团队的运行手册收集到同一个文件夹中，例如
~/runbooks/。 - 使用上面的运行手册服务器代码，并将其指向该文件夹。
- 将其添加到Claude Desktop的配置中并重启。
这样就能得到一个了解团队操作流程的助手。一旦证明其可靠性，再添加Kubernetes服务器，接着是AWS，之后再添加其他服务器。
更宏观的视角
MCP还处于发展初期，相关生态系统也在快速成长，但已有几点十分明确：拥有MCP服务器的团队所获得的AI体验与没有该服务器的团队存在显著差异。要从单纯的聊天窗口转变为类似操作团队的工具，关键在于拥有能够提供所需系统的服务器。而目前正在构建这类系统的团队，正在为AI在基础设施工作中的应用树立典范。这并非研究项目——各团队如今已在实际环境中使用它们，上述Kubernetes服务器大约20分钟即可启动运行。
相关阅读
- 一个能让 Claude 搜索团队运行手册的个人 MCP 服务器 —— 将 Markdown 格式的运行手册和事后分析记录索引到 ChromaDB 中，通过 MCP 服务器实现语义搜索，让 Claude 能在几秒钟内找到相关处理流程。
- 实用笔记：kagent——你应该了解的 Kubernetes 原生 AI 智能体框架 —— 详细讲解实用笔记：kagent——你应该了解的 Kubernetes 原生 AI 智能体框架，包括相关契约、检查机制以及供团队直接使用的代码模板。