将 Kubernetes 服务注册为 MCP Tools
前些天看到阿里云的一篇文章《Nacos 发布 MCP Registry，实现存量应用接口“0改动”升级到 MCP 协议》，深受启发，用服务注册发现结合网关设施，将存量服务转换为 MCP 工具的玩法，能非常有效地将存量服务装进 MCP 的新瓶子。那么按照我之前发表的《MCP 是一座桥》一文的思路，是不是可以更进一步，去除对特定厂商的依赖，用更通用的方式实现 MCP 的快速上车呢？下面讲讲我的尝试。
太长不看
总体流程如下图所示：
总的说来，就是利用 Kubernetes Service 的标签和注解进行标识，MCP Server 读取 Service 信息，将其作为 Tools 暴露给 MCP Client。
注册发现的实现
MCP 规范中，提供了 Tools 列表和调用的接口，在 Python SDK 里，通常对 Tools 的实现都是使用如下代码：
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("My App")
@mcp.tool()
def calculate_bmi(weight_kg: float, height_m: float) -> float:
"""Calculate BMI given weight in kg and height in meters"""
return weight_kg / (height_m**2)
在我们这个场景中，Tools 是动态刷新的，因此不能用这种硬编码的形式来声明，Python SDK 提供了低阶的 Server，便于我们进行更细致的能力实现。源码（lowlevel/server.py）中提供了创建实例、实现能力直到运行的简单介绍，例如：
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
# Implementation
@server.get_prompt()
async def handle_get_prompt(
name: str, arguments: dict[str, str] | None
) -> types.GetPromptResult:
# Implementation
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
# Implementation
@server.call_tool()
async def handle_call_tool(
name: str, arguments: dict | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
# Implementation
字面意思上看，只要实现 list_tool 和 call_tool，就能动态注册工具并执行了。例如官方 README 中介绍的代码：
from mcp.server.lowlevel import NotificationOptions, Server
from mcp.server.models import InitializationOptions
# Create a server instance
server = Server("example-server")
@server.list_prompts()
async def handle_list_prompts() -> list[types.Prompt]:
...
@server.get_prompt()
async def handle_get_prompt(
name: str, arguments: dict[str, str] | None
...
async def run():
async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
InitializationOptions(
server_name="example",
server_version="0.1.0",
capabilities=server.get_capabilities(
notification_options=NotificationOptions(),
experimental_capabilities={},
),
),
)
if __name__ == "__main__":
import asyncio
asyncio.run(run())
可以看到，低阶 MCP Server 实现中，需要自行处理原语的列表、读取和执行操作。如此看来在我们的场景中，只需要实现 list_tools 即可，例如使用如下的伪代码，查找被标记的 Service，并从 Annotation 中读取 MCP 工具定义，从而生成 MCP Tools：
try:
kubeconfig_path = os.path.expanduser("/Users/dummy/.kube/config")
namespace = "default"
fetcher = kubernetes_tools.KubernetesToolsFetcher(kubeconfig_path)
try:
k8s_tools = fetcher.get_all_mcp_tools(namespace)
for tool in k8s_tools:
missing_fields = [field for field in ("name", "description", "inputSchema") if field not in tool]
if missing_fields:
logger.error(f"Kubernetes 工具定义缺失字段: {missing_fields}，内容: {tool}")
continue
result.append(types.Tool(
name=tool["name"],
description=tool["description"],
inputSchema=tool["inputSchema"]
))
logger.info(f"从 Kubernetes 加载了 {len(k8s_tools)} 个工具")
finally:
fetcher.close()
except Exception as e:
logger.error(f"从 Kubernetes 加载工具时出错: {e}")
根据如上代码编写的 MCP Server，可以在 Client 配置中加入如下内容启动：
{
"mcpServers": {
"Demo": {
"command": "uv",
"args": [
"run",
"--with",
"mcp[cli]",
"--with",
"pyyaml",
"--with",
"kubernetes",
"[somewhere]/main.py"
]
}
},
"globalShortcut": ""
}
创建几个 Service，加上必要的标签和注解，例如：
apiVersion: v1
kind: Service
metadata:
name: mcp-demo-service-1
namespace: default
labels:
mcp: "true"
annotations:
mcptool: |
[
{
"name": "echo",
"description": "Echo back the input text",
"inputSchema": {
"type": "object",
"required": ["message"],
"properties": {
"message": {
"type": "string",
"description": "The message to echo"
}
}
}
}
]
spec:
selector:
app: mcp-demo-1
ports:
- protocol: TCP
port: 80
targetPort: 8080
创建 Kubernetes 服务后，启动 MCP 客户端，例如 Claude.app，可以看到 Kubernetes 服务已经出现，如下图所示：
还不能调用
前文贴的代码里，已经展示了 call_tool 的能力，下一步应该就是调用了，这方面会稍显复杂，要涉及到通信、认证、加解密、封装等问题，各种 API 网关、或者 ZTM 这样的应用层产品，应该能辅助解决这类问题。