# 将 Kubernetes 服务注册为 MCP Tools

前些天看到阿里云的一篇文章《Nacos 发布 MCP Registry，实现存量应用接口“0改动”升级到 MCP 协议》，深受启发，用服务注册发现结合网关设施，将存量服务转换为 MCP 工具的玩法，能非常有效地将存量服务装进 MCP 的新瓶子。那么按照我之前发表的《MCP 是一座桥》一文的思路，是不是可以更进一步，去除对特定厂商的依赖，用更通用的方式实现 MCP 的快速上车呢？下面讲讲我的尝试。

## 太长不看

总体流程如下图所示：

```mermaid
sequenceDiagram
    participant User as MCP Client
    participant MainPy as MCP Server
    participant K8sAPI as Kubernetes API Server

    User->>MainPy: 调用 list_tools()
    MainPy->>MainPy: 创建 KubernetesToolsFetcher 实例
    MainPy->>MainPy: 调用 get_all_mcp_tools(namespace)
    MainPy->>K8sAPI: 查找 MCP 标签服务
    K8sAPI-->>MainPy: 返回 Service 列表
    loop 遍历每个 Service
        MainPy->>MainPy: 提取 mcptool annotation
    end
    MainPy->>MainPy: 将所有合格 tool 转为 types.Tool
    MainPy-->>User: 返回 MCP Tool 列表
```

总的说来，就是利用 Kubernetes Service 的标签和注解进行标识，MCP Server 读取 Service 信息，将其作为 Tools 暴露给 MCP Client。

## 注册发现的实现

MCP 规范中，提供了 Tools 列表和调用的接口，在 Python SDK 里，通常对 Tools 的实现都是使用如下代码：

```python
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`）中提供了创建实例、实现能力直到运行的简单介绍，例如：

```python
@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 中介绍的代码：

```python
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：

```python
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 配置中加入如下内容启动：

```json
{
  "mcpServers": {
    "Demo": {
      "command": "uv",
      "args": [
        "run",
        "--with",
        "mcp[cli]",
        "--with",
        "pyyaml",
        "--with",
        "kubernetes",
        "[somewhere]/main.py"
      ]
    }
  },
  "globalShortcut": ""
}
```

创建几个 Service，加上必要的标签和注解，例如：

```yaml
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 服务已经出现，如下图所示：

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1745896605037/85f5cac4-51b1-4489-96ad-046aef04c693.png align="center")

## 还不能调用

前文贴的代码里，已经展示了 `call_tool` 的能力，下一步应该就是调用了，这方面会稍显复杂，要涉及到通信、认证、加解密、封装等问题，各种 API 网关、或者 ZTM 这样的应用层产品，应该能辅助解决这类问题。