Pathway Live Data Framework MCP Server把实时流处理引擎接入 LLM Agent 的完整实践【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathwayModel Context ProtocolMCP是标准化 LLM 应用与外部数据源、工具之间交互的开放协议而 Pathway Live Data Framework 通过自带的 MCP Server将其“实时表”live table处理能力开放给任意 MCP 客户端——让 AI 应用可以直接调用实时统计、检索实时文档索引而不是读取一次性的静态数据快照。读完本文你将能够用McpServablePathwayMcp在十行代码内暴露自定义 MCP 工具理解工具函数“单行输入表 → 单行结果表”的契约及其底层实现把实时表的统计值作为工具返回值并将DocumentStore的 RAG 索引以 YAML 应用的形式直接暴露给 MCP 客户端。MCP Server 的角色与 Pathway 的定位MCP Server 是 AI 应用与数据源/工具之间的中介层模型通过它访问实时数据、执行动作、获取上下文。使用 MCP Server 的核心收益包括预置集成可接入大量常见工具与平台的现成集成简化搭建过程自定义集成可以按自身工作流构建并挂载自定义工具与数据源开放协议可自由实现与使用兼容性强可移植性不同应用间切换时保留上下文。MCP Client 则负责把 AI 应用连接到 MCP Server从而访问数据库、文档库与实时统计数据。Pathway 的 MCP Server 在此基础上提供两类能力实时统计Real-Time Statistics把 Pathway 引擎的实时表聚合结果喂给 LLM使决策基于最新数据面向 RAG 的文档库Document Store提供一个实时维护的检索索引让客户端高效取回相关文档。与普通 MCP Server “请求一次、返回一次静态结果” 不同Pathway 的每个工具背后都是一条流式管道客户端请求被转换为引擎中的“查询”工具输出表随上游实时表持续更新因此多次调用同一工具会看到不断变化的结果——这正是实时流处理引擎的价值所在。安装与环境要求使用 MCP Server 需要先安装 LLM xpackpip install pathway[xpack-llm]重要MCP Server 需要 Pathway Live Data Framework 的 license key源码层面通过_check_entitlements(xpack-llm-mcp)做授权检查见 mcp_server.py 中McpServer.__init__。免费 license key 可从 Pathway 官方渠道获取。MCP 客户端示例中会用到 fastmcp 的Client需自行安装fastmcp包它是 xpack-llm 的依赖源码中以optional_imports(xpack-llm)方式导入。核心组件McpServable、McpServer 与 PathwayMcp所有 API 定义在 python/pathway/xpacks/llm/mcp_server.py 中共三个关键类类职责McpServable抽象基类任何要注册到 MCP Server 的对象都必须实现register_mcp(server)方法McpServer实现 MCP 协议的服务器本体继承自PathwayServer底层用 FastMCP 承载工具注册与传输层PathwayMcp简化配置的 dataclass构造时自动创建McpServer并把serve列表里的每个 servable 注册进去PathwayMcp的参数源码默认值与官方文档一致name服务器名称MCP 客户端用它识别服务器默认pathway-mcp-servertransport传输方式默认streamable-http源码中stdio也存在但被标记为“不稳定且实验性”选择它会发出警告且不允许设置 host/porthost/port服务器绑定地址streamable-http模式下二者必填缺失会抛ValueErrorserve要暴露的McpServable实例列表。工具的“单行契约”工具函数必须满足以下约束官方文档明确要求底层由引擎的请求/响应管道强制方法有两个参数self和一张pw.Table如input_from_client。该表的 schema 即你在注册时传入的schema且客户端的一次调用对应表中的一行客户端传入的每个参数放在同名列中。返回值必须是一张带result列、单行、且 ID 与输入行相同的表用于把计算结果回传给客户端。暴露方式为McpServable.register_mcp(server)中调用server.tool(...)传入三个核心参数工具在 MCP Server 中的名称、request_handler处理方法、schema客户端输入 schema。从源码结构看这份契约是这样落地的McpServer.tool()mcp_server.py内部创建_McpServerSubject再用pw.io.python.read(subject..., schemaschema, formatjson, autocommit_duration_ms50)把 HTTP 请求流“物化”成一张pw.Table交给request_handler处理后的表再经 response writer 序列化回写。请求体在交给引擎前会做json.dumps且_McpServerSubject._verify_payload会校验 schema 中“无默认值”的列是否都有提供——这就是为什么请求参数必须与pw.Schema的列一一对应。server.tool()除三个核心参数外还支持一批可选参数可用于精细化控制工具行为参数默认说明name必填工具名request_handler必填处理函数签名必须是(self, table) - tableschema必填客户端输入 schema用于生成工具 input schemadelete_completed_queriesFalse是否删除已完成的查询cache_strategyNone可选缓存策略title缺省用name工具展示标题description缺省用处理函数 docstring工具描述output_schema未设置可选输出 schemaannotationsNoneMCP 元注解如readOnlyHint、idempotentHint等metaNone工具元数据autocommit_duration_ms50两次 commit 之间的最大间隔毫秒控制请求进入引擎的批处理节奏另外源码中的_generate_handler_signature会从pw.Schema的每列生成 FastMCP 工具的参数签名JSON 类型会被替换为dict以避免 FastMCP 内部类型提示递归问题——这意味着你的pw.Schema不仅是引擎侧的请求校验器同时就是暴露给 LLM 的工具入参 schema一处定义、两端生效。示例一暴露一个无参工具get_constant_value先看最小可用示例——暴露一个返回常量1的工具import pathway as pw from pathway.xpacks.llm.mcp_server import McpServable, McpServer, PathwayMcp # no argument required 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()要点拆解EmptyRequestSchema没有列表示该工具不接收任何参数get_constant_value基于输入表select(result1)天然保留了输入行的 ID满足“单行 相同 ID” 的契约实例化PathwayMcp只是声明配置真正启动由pw.run()触发McpServer._run会在新线程中运行 FastMCP 传输层见 mcp_server.py。用 fastmcp 客户端验证import 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列出服务器上所有工具call_tool(name..., arguments{...})调用指定工具arguments是与pw.Schema各列对应的字典。仓库的集成测试 test_mcp_server.py 采用了同样的验证方式用multiprocessing子进程拉起McpServerfastmcp.Client轮询ping就绪后执行list_tools/call_tool可参照其写法做端到端测试。示例二带参数的加法工具让客户端传两个整数并求和。先用 schema 约束入参class AddRequestSchema(pw.Schema): x: int y: int再实现工具类class AddTool(McpServable): def add(self, x_y_values: pw.Table) - pw.Table: Return a table containing the sum of the parameters x and y. results x_y_values.select(resultpw.this.x pw.this.y) return results def register_mcp(self, server: McpServer): server.tool( add, request_handlerself.add, schemaAddRequestSchema, ) function_to_serve AddTool()客户端调用时传入{x: 4, y: 6}async with client: result await client.call_tool(nameadd, arguments{x: 4, y: 6}) print(result)注意select(resultpw.this.x pw.this.y)直接对输入行做列运算结果表仍为单行且 ID 不变无需任何额外处理。示例三同一个 Server 暴露多个工具两种方式效果完全等价。方式 A多个 servable 实例放入serve列表constant_tool ConstantValueTool() add_tool AddTool() pathway_mcp_server PathwayMcp( nameStreamable MCP Server, transportstreamable-http, hostlocalhost, port8123, serve[constant_tool, add_tool], )方式 B一个类中实现多个工具方法在register_mcp里逐个注册class BasicTools(McpServable): def get_constant_value(self, input_from_client: pw.Table) - pw.Table: Return a constant value. return input_from_client.select(result1) def add(self, x_y_values: pw.Table) - pw.Table: Return a table containing the sum of the parameters x and y. results x_y_values.select(resultpw.this.x pw.this.y) return results def register_mcp(self, server: McpServer): server.tool( get_constant_value, request_handlerself.get_constant_value, schemaEmptyRequestSchema, ) server.tool( add, request_handlerself.add, schemaAddRequestSchema, ) pathway_mcp_server PathwayMcp( nameStreamable MCP Server, transportstreamable-http, hostlocalhost, port8123, serve[BasicTools()], ) pw.run()两种方式最终list_tools都能同时看到get_constant_value与add客户端逐个调用即可。示例四统计实时表的行数实时能力的体现前几个例子的结果都是“静态”的。Pathway 的看点在于工具可以读取一张持续更新的实时表。先用pw.demo.range_stream生成一张合成流——每秒新增一行value列从 0 到 49table pw.demo.range_stream(nb_rows50)然后写一个统计行数的工具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, # 空 schema ) function_to_serve CountTool()这段代码集中体现了“单行契约”的工程细节逐行解释不能直接返回table返回表必须与输入行 ID 相同的单行表而table是持续增长的多行表。正确做法是先聚合成单行表再把聚合值“挂回”到客户端输入行上table.reduce(countpw.reducers.count())得到一张至多一行的计数表因为表可能为空计数表也可能是空的所以必须用left joinempty_row.join_left(single_row_table, idempty_row.id)保证客户端行一定存在此时count为Noneidempty_row.id正是保留输入行 ID 的关键最后用pw.if_else(... is_none(), 0, ...)把空表情况归一化为0——服务器在表非空时返回实时计数否则返回0。客户端调用async with client: result await client.call_tool(nameget_count, arguments{}) print(result)连续多次调用计数值会随range_stream每秒 1 而增长——这是 MCP Server 返回“新鲜数据”而非快照的最直观证据。完整示例实时统计工具下面是一个把 count/min/max/avg/latest 聚合打包成字符串返回的完整工具适合作为“实时指标喂给 LLM” 的模板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, )说明与注意事项工具不要求任何输入因此input_from_client是一张只有id列的单行表示例中empty_row指的就是这张输入表groupby().reduce(...)用五个 reducer 一次性完成聚合pw.udf把数字格式化成自然语言字符串返回也可以改成 JSON 结构方便客户端二次计算pw.run(monitoring_levelpw.MonitoringLevel.NONE, terminate_on_errorFalse)关闭监控面板输出、避免单点错误终止进程适合长期运行的服务场景与 Count 示例相同的套路聚合 → left join 回输入行 →if_else处理空表 →result列输出。客户端访问方式import asyncio from fastmcp import Client PATHWAY_MCP_URL http://localhost:8123/mcp/ client Client(PATHWAY_MCP_URL) async def main(): async with client: result await client.call_tool(nameget_statistics, arguments{}) print(result) asyncio.run(main())这些统计值会随底层实时表持续演化MCP 客户端拿到的始终是最新数据。进阶把 DocumentStore 暴露为 MCP 工具文档索引是 RAG 与 agent 管线的核心索引的组织方式决定了信息能否被快速检索取回。Pathway 的DocumentStorepython/pathway/xpacks/llm/document_store.py本身就继承自McpServable其register_mcp会向服务器注册三个工具retrieve_query按查询文本从混合索引中检索最相关的文档statistics_query返回索引的统计信息inputs_query返回索引当前输入文档的状态。因此可以把实时文档索引直接交给PathwayMcp让任意 MCP 客户端接入这个持续更新的检索层——新文档落入文件系统后索引自动重建客户端无需感知。YAML 应用写法在 YAML 应用中只需一个PathwayMcp节点并引用$document_store变量mcp_http: !pw.xpacks.llm.mcp_server.PathwayMcp name: Streamable MCP Server transport: streamable-http host: localhost port: 8068 serve: - $document_store完整 RAG MCP 管道示例如下数据源 → 解析/切分 → 混合检索工厂 → DocumentStore → MCP Server 一条链$sources: - !pw.io.fs.read path: data format: binary with_metadata: true $embedder: !pw.xpacks.llm.embedders.OpenAIEmbedder model: text-embedding-ada-002 cache_strategy: !pw.udfs.DefaultCache {} $splitter: !pw.xpacks.llm.splitters.TokenCountSplitter min_tokens: 250 max_tokens: 600 $parser: !pw.xpacks.llm.parsers.DoclingParser {} $knn_index: !pw.stdlib.indexing.BruteForceKnnFactory reserved_space: 1000 embedder: $embedder metric: !pw.engine.BruteForceKnnMetricKind.COS $bm25_index: !pw.stdlib.indexing.TantivyBM25Factory {} $retriever_factory: !pw.stdlib.indexing.HybridIndexFactory retriever_factories: - $knn_index - $bm25_index $document_store: !pw.xpacks.llm.document_store.DocumentStore docs: $sources parser: $parser splitter: $splitter retriever_factory: $retriever_factory # Streamable MCP server, can be proxied mcp_http: !pw.xpacks.llm.mcp_server.PathwayMcp name: Streamable MCP Server transport: streamable-http host: localhost port: 8068 serve: - $document_store组件说明$sourcespw.io.fs.read监听data目录format: binary保证data列是原始字节DocumentStore要求docs表含 bytes 类型的data列with_metadata: true额外产出用于过滤的_metadata列$embedderOpenAI 嵌入模型配DefaultCache避免重复请求嵌入接口$splitter按 token 数切分250600 token$parserDocling 解析器负责多格式文档转纯文本$knn_index$bm25_index向量 KNN余弦相似度与 Tantivy BM25 关键词索引由HybridIndexFactory组合成混合检索$document_store消费上面所有组件构建解析 → 切分 → 嵌入 → 建索引的流式管道mcp_http把$document_store注册为 MCP Serverstreamable-http传输可被反向代理便于暴露到团队内网。小结Pathway 的 MCP Server 本质上是把“实时流处理引擎”包装成 MCP 工具层McpServable定义了register_mcp契约McpServer把每次客户端调用转成引擎中的 JSON 请求表并执行流式管道PathwayMcp负责一站式装配。掌握“单行输入表 → 单行result表”契约、reduce聚合 left join 回输入行的空表处理模式以及DocumentStore的 YAML 集成后你就可以让 LLM 应用实时读取业务统计与文档索引构建数据始终“新鲜”的 agent 工作流。参考文件MCP Server 官方教程MCP Server 实现DocumentStore 实现MCP Server 集成测试【免费下载链接】pathwayPython ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.项目地址: https://gitcode.com/GitHub_Trending/pa/pathway创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考