如何用 Pathway LLM xpack 搭建自己的 MCP Server 并暴露实时统计工具【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway本文解决的问题是你有一个实时更新的 Pathway 表希望把它封装成 MCPModel Context ProtocolServer让 MCP 客户端如 Claude Desktop 或自写的 fastmcp 客户端能调用工具拿到这张表当前的行数、最小值、最大值、平均值等实时统计。完成的标志是MCP 客户端能通过list_tools看到注册的工具通过call_tool调用后返回统计字符串并且多次调用时数值会随表更新而变化。适用前提均来自项目文档平台为 MacOS 或 LinuxPathway 目前不支持 Windows文档建议用 WSL、Docker 或 VM见 安装文档Python 环境Claude Desktop 教程给出的最低要求是 Python 3.9MCP Server 功能属于 LLM xpack需要一个 Pathway 的 license key文档说明可免费申请MCP Server 文档要求安装pathway[xpack-llm]。1. 安装与 license key安装带 LLM xpack 的 Pathwaypip install pathway[xpack-llm]MCP Server 需要 license key。安装文档说明免费 license key 需要到官方页面注册获取文档中的get-license入口。文档中另外给出了一种演示用法可以直接写入代码Claude Desktop 教程的示例代码即如此pw.set_license_key(demo-license-key-with-telemetry)这是文档原样给出的演示 key带遥测说明正式使用请替换为你自己申请的 key。企业 license 还支持环境变量PATHWAY_LICENSE_KEY或file:///path/to/license.lic的写法详见 安装文档。2. 三个核心 APIMcpServable、register_mcp、PathwayMcpMCP Server 的搭建围绕三个对象展开以下约束均来自 MCP Server 文档McpServable你要暴露的操作必须是一个继承McpServable的类的实例。被暴露的函数有硬性约束函数必须有两个参数self和一个pw.Table客户端输入表输入表遵循你指定的 Schema且只含一行客户端传入的每个参数都在对应列里函数必须返回一个带result列的单行表且该行的 ID 必须与输入表相同。register_mcp(self, server: McpServer)在实现中调用server.tool(...)注册工具需要三个参数工具在 MCP Server 中的名字、request_handler上面那个函数、schema输入 Schema 类。PathwayMcp启动 Server参数有nameMCP 客户端用它识别你的 Servertransport目前文档只支持streamable-httphost/port监听地址与端口serve要暴露的McpServable实例列表。最小可运行示例文档中的常量工具用于先跑通链路import pathway as pw from pathway.xpacks.llm.mcp_server import McpServable, McpServer, PathwayMcp # 无参数输入 class EmptyRequestSchema(pw.Schema): pass class ConstantValueTool(McpServable): def get_constant_value(self, input_from_client: pw.Table) - pw.Table: Return a constant value. return input_from_client.select(result1) def register_mcp(self, server: McpServer): server.tool( get_constant_value, request_handlerself.get_constant_value, schemaEmptyRequestSchema, ) function_to_serve ConstantValueTool() pathway_mcp_server PathwayMcp( nameStreamable MCP Server, transportstreamable-http, hostlocalhost, port8123, serve[function_to_serve], ) pw.run()保存为 Python 文件后执行python 文件名.py启动。Server 启动后端点为http://localhost:8123/mcp/Claude Desktop 教程对 8123 端口的这个地址给出了同样的说明。3. 用 MCP 客户端验证工具是否暴露成功文档给出的测试客户端基于fastmcp包外部库文档示例直接from fastmcp import Clientimport asyncio from fastmcp import Client PATHWAY_MCP_URL http://localhost:8123/mcp/ client Client(PATHWAY_MCP_URL) async def main(): async with client: tools await client.list_tools() print(tools) async with client: result await client.call_tool(nameget_constant_value, arguments{}) print(result) asyncio.run(main())验证方式list_tools()确认get_constant_value出现在返回的工具列表中call_tool(name..., arguments...)arguments是字典与工具 Schema 的字段对应。上例工具无输入所以传空字典{}。如果list_tools列不出工具或连接失败先确认 Server 进程是否正在运行、端口是否被占用——这两个是文档中 Server 定义host/port与实际端点能对应的条件。4. 暴露实时统计工具Count 与 Statistics统计工具的核心在于实时表不能直接作为返回值返回必须是与输入同 ID 的单行表需要先用聚合器reducer压成单行再 join 回输入表。文档给了两级示例数据源都来自pw.demo模块的流table pw.demo.range_stream(nb_rows50)该流有一列value每秒插入一行从 0 到 49range_stream的完整参数见 demo 模块文档。4.1 行数统计Count文档示例代码如下class CountTool(McpServable): def get_count(self, empty_row: pw.Table) - pw.Table: Return a the number of entries in the Pathway table. single_row_table table.reduce(countpw.reducers.count()) results empty_row.join_left(single_row_table, idempty_row.id).select( countpw.right.count ) results results.select( resultpw.if_else(pw.this.count.is_none(), 0, pw.this.count) ) return results def register_mcp(self, server: McpServer): server.tool( get_count, request_handlerself.get_count, schemaInputEmptyRequestSchema, ) function_to_serve CountTool()代码中两个容易出错的点文档都有明确解释用left join且idempty_row.id或idpw.left.id是为了在聚合表为空源表还没有数据时也能保留输入行此时count为None再用pw.if_else(pw.this.count.is_none(), 0, pw.this.count)把None转成0保证返回“表为空时是 0否则是当前行数”。注意文档原例里register_mcp引用的是InputEmptyRequestSchema而本文第 2 节定义的无参 Schema 类名是EmptyRequestSchema——直接运行前请统一为你实际定义的类名否则工具注册会报NameError。客户端调用async with client: result await client.call_tool(nameget_count, arguments{}) print(result)文档说明的验证方式是连续多次调用返回的行数会随流每秒插入一行而变化——这是“实时统计”是否真正生效的判断依据文档未给出固定数值不应把某次返回值当成功标准。4.2 完整统计Statistics文档给出了 count / min / max / avg / latest 五合一的完整示例用pw.udf把五个聚合值格式化成字符串import pathway as pw from pathway.xpacks.llm.mcp_server import McpServable, McpServer, PathwayMcp class ValueRequestSchema(pw.Schema): pass table pw.demo.range_stream(nb_rows50) class StatisticsTool(McpServable): def get_statistics(self, input_from_client: pw.Table) - pw.Table: Return basic statistics about the table. pw.udf def statistics_udf(count, minimum, maximum, avg, latest) - str: return fcount: {count}, min: {minimum}, max: {maximum}, avg: {avg}, latest: {latest} single_row_table table.groupby().reduce( countpw.reducers.count(pw.this.value), minpw.reducers.min(pw.this.value), maxpw.reducers.max(pw.this.value), avgpw.reducers.avg(pw.this.value), latestpw.reducers.latest(pw.this.value), ) single_cell_table single_row_table.select( single_cellstatistics_udf( pw.this.count, pw.this.min, pw.this.max, pw.this.avg, pw.this.latest, ) ) results empty_row.join_left(single_cell_table, idempty_row.id).select( single_cellpw.right.single_cell ) results results.select( resultpw.if_else( pw.this.single_cell.is_none(), count: 0, min: None, max: None, avg: None, latest: None, pw.this.single_cell ) ) return results def register_mcp(self, server: McpServer): server.tool( get_statistics, request_handlerself.get_statistics, schemaValueRequestSchema, ) function_to_serve StatisticsTool() pathway_mcp_server PathwayMcp( nameStreamable MCP Server, transportstreamable-http, hostlocalhost, port8123, serve[function_to_serve], ) pw.run( monitoring_levelpw.MonitoringLevel.NONE, terminate_on_errorFalse, )两处使用注意均为文档代码原样保留文档原例在get_statistics内部引用了empty_row而该函数参数名为input_from_client运行前把empty_row统一改为函数参数名或反过来改参数名逻辑不变文档说明无参工具的输入表是只含id列的单行表因此join_left(..., idempty_row.id)的写法就是把聚合结果“搬”到与输入同 ID 的那一行上。pw.run(monitoring_levelpw.MonitoringLevel.NONE, terminate_on_errorFalse)是文档示例使用的启动参数关闭监控输出、遇错不直接终止进程。客户端调用方式与 Count 相同async with client: result await client.call_tool(nameget_statistics, arguments{}) print(result)返回的是形如count: ..., min: ..., max: ..., avg: ..., latest: ...的字符串上面代码中 UDF 的输出格式文档提示可以把它改成 JSON 再交给客户端侧做进一步计算。同样地文档强调这些数值会随表更新而演进多次调用结果应不同。5. 一个 Server 暴露多个工具如果除了统计还要暴露其他工具文档还给了求和工具add的例子有两种等价写法每个工具一个McpServable类serve[constant_tool, add_tool]传多个实例或在一个类里写多个 handler在同一个register_mcp里多次调用server.tool(...)。带参数的工具用 Schema 约束输入文档示例class AddRequestSchema(pw.Schema): x: int y: int客户端对应传arguments{x: 4, y: 6}。验证方式不变list_tools()中应看到全部注册的工具名逐个call_tool检查返回。6. 限制与下一步transport目前只支持streamable-http文档原文 onlystreamable-httpis available for nowServer 地址、端口以你PathwayMcp中设置的host/port为准本文示例端点是http://localhost:8123/mcp/。MCP Server 依赖 license key缺 key 时无法启动安装与文档均将其列为 Important 前提。文档中的客户端示例依赖fastmcp包若你的目标客户端是 Claude Desktop配套教程给出了完整配置路径编辑claude_desktop_config.json用npx mcp-remote代理到http://localhost:8123/mcp/并提示需要 npm/npx 以及 Node.js 出现在 Detected Tools 中。如果后续要把 RAG 检索也纳入同一个 Server文档给出了现成做法DocumentStore继承自McpServable可以直接放进serve列表YAML 应用中的写法为mcp_http: !pw.xpacks.llm.mcp_server.PathwayMcp name: Streamable MCP Server transport: streamable-http host: localhost port: 8068 serve: - $document_store这一步与统计工具互不冲突可以在同一个PathwayMcp实例中共存。【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 如何用 Pathway LLM xpack 搭建自己的 MCP Server 并暴露实时统计工具？
如何用 Pathway LLM xpack 搭建自己的 MCP Server 并暴露实时统计工具【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway本文解决的问题是你有一个实时更新的 Pathway 表希望把它封装成 MCPModel Context ProtocolServer让 MCP 客户端如 Claude Desktop 或自写的 fastmcp 客户端能调用工具拿到这张表当前的行数、最小值、最大值、平均值等实时统计。完成的标志是MCP 客户端能通过list_tools看到注册的工具通过call_tool调用后返回统计字符串并且多次调用时数值会随表更新而变化。适用前提均来自项目文档平台为 MacOS 或 LinuxPathway 目前不支持 Windows文档建议用 WSL、Docker 或 VM见 安装文档Python 环境Claude Desktop 教程给出的最低要求是 Python 3.9MCP Server 功能属于 LLM xpack需要一个 Pathway 的 license key文档说明可免费申请MCP Server 文档要求安装pathway[xpack-llm]。1. 安装与 license key安装带 LLM xpack 的 Pathwaypip install pathway[xpack-llm]MCP Server 需要 license key。安装文档说明免费 license key 需要到官方页面注册获取文档中的get-license入口。文档中另外给出了一种演示用法可以直接写入代码Claude Desktop 教程的示例代码即如此pw.set_license_key(demo-license-key-with-telemetry)这是文档原样给出的演示 key带遥测说明正式使用请替换为你自己申请的 key。企业 license 还支持环境变量PATHWAY_LICENSE_KEY或file:///path/to/license.lic的写法详见 安装文档。2. 三个核心 APIMcpServable、register_mcp、PathwayMcpMCP Server 的搭建围绕三个对象展开以下约束均来自 MCP Server 文档McpServable你要暴露的操作必须是一个继承McpServable的类的实例。被暴露的函数有硬性约束函数必须有两个参数self和一个pw.Table客户端输入表输入表遵循你指定的 Schema且只含一行客户端传入的每个参数都在对应列里函数必须返回一个带result列的单行表且该行的 ID 必须与输入表相同。register_mcp(self, server: McpServer)在实现中调用server.tool(...)注册工具需要三个参数工具在 MCP Server 中的名字、request_handler上面那个函数、schema输入 Schema 类。PathwayMcp启动 Server参数有nameMCP 客户端用它识别你的 Servertransport目前文档只支持streamable-httphost/port监听地址与端口serve要暴露的McpServable实例列表。最小可运行示例文档中的常量工具用于先跑通链路import pathway as pw from pathway.xpacks.llm.mcp_server import McpServable, McpServer, PathwayMcp # 无参数输入 class EmptyRequestSchema(pw.Schema): pass class ConstantValueTool(McpServable): def get_constant_value(self, input_from_client: pw.Table) - pw.Table: Return a constant value. return input_from_client.select(result1) def register_mcp(self, server: McpServer): server.tool( get_constant_value, request_handlerself.get_constant_value, schemaEmptyRequestSchema, ) function_to_serve ConstantValueTool() pathway_mcp_server PathwayMcp( nameStreamable MCP Server, transportstreamable-http, hostlocalhost, port8123, serve[function_to_serve], ) pw.run()保存为 Python 文件后执行python 文件名.py启动。Server 启动后端点为http://localhost:8123/mcp/Claude Desktop 教程对 8123 端口的这个地址给出了同样的说明。3. 用 MCP 客户端验证工具是否暴露成功文档给出的测试客户端基于fastmcp包外部库文档示例直接from fastmcp import Clientimport asyncio from fastmcp import Client PATHWAY_MCP_URL http://localhost:8123/mcp/ client Client(PATHWAY_MCP_URL) async def main(): async with client: tools await client.list_tools() print(tools) async with client: result await client.call_tool(nameget_constant_value, arguments{}) print(result) asyncio.run(main())验证方式list_tools()确认get_constant_value出现在返回的工具列表中call_tool(name..., arguments...)arguments是字典与工具 Schema 的字段对应。上例工具无输入所以传空字典{}。如果list_tools列不出工具或连接失败先确认 Server 进程是否正在运行、端口是否被占用——这两个是文档中 Server 定义host/port与实际端点能对应的条件。4. 暴露实时统计工具Count 与 Statistics统计工具的核心在于实时表不能直接作为返回值返回必须是与输入同 ID 的单行表需要先用聚合器reducer压成单行再 join 回输入表。文档给了两级示例数据源都来自pw.demo模块的流table pw.demo.range_stream(nb_rows50)该流有一列value每秒插入一行从 0 到 49range_stream的完整参数见 demo 模块文档。4.1 行数统计Count文档示例代码如下class CountTool(McpServable): def get_count(self, empty_row: pw.Table) - pw.Table: Return a the number of entries in the Pathway table. single_row_table table.reduce(countpw.reducers.count()) results empty_row.join_left(single_row_table, idempty_row.id).select( countpw.right.count ) results results.select( resultpw.if_else(pw.this.count.is_none(), 0, pw.this.count) ) return results def register_mcp(self, server: McpServer): server.tool( get_count, request_handlerself.get_count, schemaInputEmptyRequestSchema, ) function_to_serve CountTool()代码中两个容易出错的点文档都有明确解释用left join且idempty_row.id或idpw.left.id是为了在聚合表为空源表还没有数据时也能保留输入行此时count为None再用pw.if_else(pw.this.count.is_none(), 0, pw.this.count)把None转成0保证返回“表为空时是 0否则是当前行数”。注意文档原例里register_mcp引用的是InputEmptyRequestSchema而本文第 2 节定义的无参 Schema 类名是EmptyRequestSchema——直接运行前请统一为你实际定义的类名否则工具注册会报NameError。客户端调用async with client: result await client.call_tool(nameget_count, arguments{}) print(result)文档说明的验证方式是连续多次调用返回的行数会随流每秒插入一行而变化——这是“实时统计”是否真正生效的判断依据文档未给出固定数值不应把某次返回值当成功标准。4.2 完整统计Statistics文档给出了 count / min / max / avg / latest 五合一的完整示例用pw.udf把五个聚合值格式化成字符串import pathway as pw from pathway.xpacks.llm.mcp_server import McpServable, McpServer, PathwayMcp class ValueRequestSchema(pw.Schema): pass table pw.demo.range_stream(nb_rows50) class StatisticsTool(McpServable): def get_statistics(self, input_from_client: pw.Table) - pw.Table: Return basic statistics about the table. pw.udf def statistics_udf(count, minimum, maximum, avg, latest) - str: return fcount: {count}, min: {minimum}, max: {maximum}, avg: {avg}, latest: {latest} single_row_table table.groupby().reduce( countpw.reducers.count(pw.this.value), minpw.reducers.min(pw.this.value), maxpw.reducers.max(pw.this.value), avgpw.reducers.avg(pw.this.value), latestpw.reducers.latest(pw.this.value), ) single_cell_table single_row_table.select( single_cellstatistics_udf( pw.this.count, pw.this.min, pw.this.max, pw.this.avg, pw.this.latest, ) ) results empty_row.join_left(single_cell_table, idempty_row.id).select( single_cellpw.right.single_cell ) results results.select( resultpw.if_else( pw.this.single_cell.is_none(), count: 0, min: None, max: None, avg: None, latest: None, pw.this.single_cell ) ) return results def register_mcp(self, server: McpServer): server.tool( get_statistics, request_handlerself.get_statistics, schemaValueRequestSchema, ) function_to_serve StatisticsTool() pathway_mcp_server PathwayMcp( nameStreamable MCP Server, transportstreamable-http, hostlocalhost, port8123, serve[function_to_serve], ) pw.run( monitoring_levelpw.MonitoringLevel.NONE, terminate_on_errorFalse, )两处使用注意均为文档代码原样保留文档原例在get_statistics内部引用了empty_row而该函数参数名为input_from_client运行前把empty_row统一改为函数参数名或反过来改参数名逻辑不变文档说明无参工具的输入表是只含id列的单行表因此join_left(..., idempty_row.id)的写法就是把聚合结果“搬”到与输入同 ID 的那一行上。pw.run(monitoring_levelpw.MonitoringLevel.NONE, terminate_on_errorFalse)是文档示例使用的启动参数关闭监控输出、遇错不直接终止进程。客户端调用方式与 Count 相同async with client: result await client.call_tool(nameget_statistics, arguments{}) print(result)返回的是形如count: ..., min: ..., max: ..., avg: ..., latest: ...的字符串上面代码中 UDF 的输出格式文档提示可以把它改成 JSON 再交给客户端侧做进一步计算。同样地文档强调这些数值会随表更新而演进多次调用结果应不同。5. 一个 Server 暴露多个工具如果除了统计还要暴露其他工具文档还给了求和工具add的例子有两种等价写法每个工具一个McpServable类serve[constant_tool, add_tool]传多个实例或在一个类里写多个 handler在同一个register_mcp里多次调用server.tool(...)。带参数的工具用 Schema 约束输入文档示例class AddRequestSchema(pw.Schema): x: int y: int客户端对应传arguments{x: 4, y: 6}。验证方式不变list_tools()中应看到全部注册的工具名逐个call_tool检查返回。6. 限制与下一步transport目前只支持streamable-http文档原文 onlystreamable-httpis available for nowServer 地址、端口以你PathwayMcp中设置的host/port为准本文示例端点是http://localhost:8123/mcp/。MCP Server 依赖 license key缺 key 时无法启动安装与文档均将其列为 Important 前提。文档中的客户端示例依赖fastmcp包若你的目标客户端是 Claude Desktop配套教程给出了完整配置路径编辑claude_desktop_config.json用npx mcp-remote代理到http://localhost:8123/mcp/并提示需要 npm/npx 以及 Node.js 出现在 Detected Tools 中。如果后续要把 RAG 检索也纳入同一个 Server文档给出了现成做法DocumentStore继承自McpServable可以直接放进serve列表YAML 应用中的写法为mcp_http: !pw.xpacks.llm.mcp_server.PathwayMcp name: Streamable MCP Server transport: streamable-http host: localhost port: 8068 serve: - $document_store这一步与统计工具互不冲突可以在同一个PathwayMcp实例中共存。【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考