Pathway MCP Server 对接 Claude Desktop 实战：让 Claude 实时访问流式统计与文档库

本篇指南基于 Pathway 官方教程，讲解如何将 Pathway Live Data Framework 的 MCP Server 接入 Claude Desktop：从创建一个暴露实时计算工具的 pathway_mcp_server.py，到编写 claude_desktop_config.json 配置完成连接，再到让 Claude 调用工具获取实时统计与文档库检索结果。读完本文，你可以独立搭起一套“AI 客户端 + Pathway 流式引擎”的实时数据通道，并理解其底层 McpServable/PathwayMcp 的实现机制。

一、背景：MCP 与 Pathway MCP Server 的角色

Model Context Protocol（MCP） 是一种标准化协议，让 AI 应用以统一方式访问外部数据源与工具。Pathway 在 LLM xpack 中内置了一个自己的 MCP Server（源码位于 python/pathway/xpacks/llm/mcp_server.py），它充当 AI 应用（如 Claude Desktop）与 Pathway 流式引擎之间的中介，为 AI 应用提供三类实时数据处理能力：

• 获取实时统计（real-time statistics）；
• 访问与查询文档库（Document Store）；
• 在 live data 上执行计算。

其核心设计原则是：每个工具都通过显式的 tool 注册暴露，确保你始终掌控 Claude 能访问和修改什么。协议层面的原理与更多服务端示例，可参考仓库中的姊妹篇文档 Pathway Live Data Framework MCP Server。

二、前置条件

开始之前，请确认环境中具备以下条件：

依赖	说明
Claude Desktop	已安装并更新到最新版本
Python 3.9+	运行 MCP Server 脚本
Pathway（xpack-llm 扩展）	执行 pip install pathway[xpack-llm] 安装；该扩展在 pyproject.toml 的 xpack-llm optional-dependencies 中定义，包含 openai、litellm、fastmcp 相关依赖等
npm / npx	Claude Desktop 端通过 npx mcp-remote 连接远端 MCP Server 时需要

重要提示：MCP Server 需要一个 Pathway Live Data Framework 许可证（license key）。本地开发可以使用遥测演示 key（下文示例即使用 demo-license-key-with-telemetry），正式使用可申请免费 key。源码层面这一约束是硬性校验——McpServer 构造函数 首先调用 _check_entitlements("xpack-llm-mcp")，没有合法授权将无法启动服务。

三、搭建 MCP Server

3.1 创建一个最简 MCP Server

创建 Python 文件 pathway_mcp_server.py，内容如下（这是官方教程给出的完整可运行示例，暴露一个返回常量 1 的 get_constant_value 工具）：

import pathway as pw
from pathway.xpacks.llm.mcp_server import McpServable, McpServer, PathwayMcp

pw.set_license_key("demo-license-key-with-telemetry")

class EmptyRequestSchema(pw.Schema):
    pass

class ConstantValueTool(McpServable):
    def get_constant_value(self, input_from_client: pw.Table) -> pw.Table:
        return input_from_client.select(result=1)

    def register_mcp(self, server: McpServer):
        server.tool(
            "get_constant_value",
            request_handler=self.get_constant_value,
            schema=EmptyRequestSchema,
        )

function_to_serve = ConstantValueTool()
pathway_mcp_server = PathwayMcp(
    name="Pathway Live Data Framework MCP Server",
    transport="streamable-http",
    host="localhost",
    port=8123,
    serve=[function_to_serve],
)
pw.run()

各要素的作用：

• EmptyRequestSchema：工具入参的 pw.Schema。空 Schema 表示该工具不接受任何客户端参数，Claude 调用时传空对象即可。
• ConstantValueTool(McpServable)：所有可暴露对象都要继承抽象基类 McpServable，并实现 register_mcp(server) 方法，在其中通过 server.tool(name, request_handler=..., schema=...) 完成注册。
• get_constant_value 请求处理函数：签名约定为 (self, input_from_client: pw.Table) -> pw.Table。入参表遵循 EmptyRequestSchema、只含一行，每个客户端参数对应一列；返回值必须是含 result 列的单行表，且行的 id 与入参表一致。
• PathwayMcp：配置对象，参数含义为：
  • name：服务器名称，MCP 客户端用它识别你的 Server；
  • transport：传输方式。文档写作时以 streamable-http 为主；从 McpConfig 源码 看，当前代码也接受 "stdio"，但会触发实验性告警且禁止同时设置 host/port；
  • host / port：streamable-http 传输下二者必填，否则 __post_init__ 校验会直接抛 ValueError；
  • serve：要注册的 McpServable 实例列表。

3.2 启动并验证

运行脚本：

python pathway_mcp_server.py

启动后，MCP 服务端点位于 http://localhost:8123/mcp/。

从源码结构看，PathwayMcp 的初始化会经 McpServer.get(config) 按 name 获取（或创建）单例，然后依次调用每个 servable 的 register_mcp（见 PathwayMcp.post_init）；McpServer 继承自 PathwayServer，内部封装了一个 FastMCP 实例，并在独立守护线程中通过 self._fastmcp.run(**runner_kwargs) 提供 streamable-http 服务（见 McpServer._run）。

3.3 源码纵深：一个 tool 是如何变成 MCP 接口的

McpServer.tool() 是整个机制的核心，其调用链值得拆开看：

1. 创建请求 subject：构造 _McpServerSubject（定义处），它负责把 MCP 客户端发来的 JSON 参数按 Schema 校验后写入引擎——缺列会抛出 Column ... is required but not provided in the payload（见 _verify_payload）。
2. 把 HTTP 请求流变成 Pathway 输入表：io.python.read(subject=subject, schema=schema, format="json", autocommit_duration_ms=50)——这意味着每一次工具调用在引擎内部就是一条流式输入，request_handler 返回的 pw.Table 再经响应写回器转成工具返回值。
3. Schema 到工具签名的自动转换：_generate_handler_signature 遍历 pw.Schema.columns()，把每一列生成为 FastMCP 工具的 KEYWORD_ONLY 参数（含类型标注与默认值），所以 Claude 端看到的 tool schema 完全由你的 pw.Schema 驱动。
4. 可选参数：tool() 还支持 delete_completed_queries、cache_strategy、title、description（缺省时取 request_handler 的 docstring）、output_schema、annotations、meta、autocommit_duration_ms（默认 50ms）等，可用于控制缓存、结构化输出与工具元数据。

仓库中的集成测试 integration_tests/xpack/test_mcp_server.py 验证了这些行为：list_tools 正确返回注册的工具名；result 列为普通文本时客户端收到 content[0].text；当 result 列是 pw.Json 且提供了 output_schema 时，客户端可拿到 structured_content（结构化 JSON）；Schema 列使用 pw.column_definition(default_value=...) 时缺省参数可生效。

四、配置 Claude Desktop 连接

完成客户端配置步骤如下：

1. 打开 Claude Desktop；
2. 点击顶部菜单栏的 “File” → “Settings...”；
3. 切换到 “Developer” 标签页，点击 “Edit Config”；
4. 打开并编辑 claude_desktop_config.json，写入：

{
  "mcpServers": {
    "pathway": {
      "command": "npx",
      "args": [
        "mcp-remote",
        "http://localhost:8123/mcp/",
        "--transport",
        "http-first"
      ]
    }
  }
}

这里的机制是：Claude Desktop 本身以本地 stdio 方式拉起 npx mcp-remote 进程，由它把 stdio 流量桥接到你 http://localhost:8123/mcp/ 的 streamable-http 端点，因此客户端侧无需改动服务器代码。

5. 按 Ctrl+R 重新加载 Claude Desktop 配置，Claude 即会连接到你的 MCP Server。

故障排查（官方教程给出的排障路径）：

• 若连接失败，进入 “Extensions” → “Advanced Settings”，检查 “Detected Tools” 一栏是否出现 Node.js；没有则需（重新）安装 npm 与 npx；
• 安装完成后可能需要重启电脑与 Claude Desktop，Windows 上尤其常见；
• 一切正常时，服务器会出现在 “Developer” 分区下的 “Local MCP Server” 列表中。

五、使用 Claude 调用工具

配置完成后，直接用自然语言让 Claude 使用 MCP Server 暴露的工具：

• “What is the constant value?” —— Claude 会调用 get_constant_value 工具并返回结果 1；
• “Can you provide real-time statistics about the data?” —— 若你已暴露统计工具（见姊妹篇文档中的 Statistics 完整示例），Claude 将实时取数并展示。

Claude Desktop 每次使用工具前都会请求你确认（validate），这是刻意保留的人机控制点，对依赖 LLM 等昂贵计算的工具尤为重要。工具可用时，“Pathway Live Data Framework” 会作为工具出现在聊天界面的工具列表中。

六、扩展：接入更多工具与 RAG 文档库

6.1 替换/增加工具

要暴露其他工具（加法、计数、实时统计等），只需参照 MCP Server 文档 中的 Addition / Count / Statistics 示例实现新的 McpServable，然后重启 MCP Server 并重新加载 Claude Desktop 即可——claude_desktop_config.json 无需任何改动，因为连接指向的端点不变。

实时统计的写法要点：工具返回值必须是与入参同 id 的单行表，因此对实时表要先用 groupby().reduce(...) 聚合成单行，再 join_left 回入参行，并用 pw.if_else(... .is_none(), 0, ...) 兜底空表情形——完整代码可查阅 Statistics 示例。

6.2 暴露 Document Store，快速搭起 RAG

MCP Server 最大的应用价值在于 RAG：Pathway 的 DocumentStore 本身继承自 McpServable，可直接传入 PathwayMcp 的 serve 列表，使任何 MCP 客户端都能检索一个随源数据持续更新的实时索引。官方文档给出了 YAML 应用模板，核心片段为：

mcp_http: !pw.xpacks.llm.mcp_server.PathwayMcp
  name: "Streamable MCP Server"
  transport: "streamable-http"
  host: "localhost"
  port: 8068
  serve:
    - $document_store

完整的“文件源 → 解析 → 切分 → 混合索引（KNN + BM25）→ DocumentStore → MCP Server”的 YAML 编排示例，见 Exposing Pathway Document Store 一节。

七、总结

通过本教程的流程，Claude Desktop 获得了三项实际能力：访问实时数据、对 live 表执行计算、查询 Pathway 文档库。整条链路的分工是——Pathway 负责流式数据加工并实现 MCP 服务端（streamable-http，端点 /mcp/），mcp-remote 负责 stdio 与 HTTP 之间的桥接，Claude Desktop 负责工具发现、调用与人工确认。由于工具注册完全由 pw.Schema 与 McpServer.tool() 驱动（参见 mcp_server.py），向现有服务添加新能力只是新增一个 McpServable 并重载客户端的事，这正是把实时流处理引擎嵌入 Agent 工作流的低成本方式。