企业官网建设流程全解析

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/pathway

Model Context Protocol（MCP）是标准化 LLM 应用与外部数据源、工具之间交互的开放协议，而 Pathway Live Data Framework 通过自带的 MCP Server，将其“实时表”（live table）处理能力开放给任意 MCP 客户端——让 AI 应用可以直接调用实时统计、检索实时文档索引，而不是读取一次性的静态数据快照。读完本文，你将能够：用McpServable+PathwayMcp在十行代码内暴露自定义 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 xpack：

pip 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 中，共三个关键类：

PathwayMcp的参数（源码默认值与官方文档一致）：

• name：服务器名称，MCP 客户端用它识别服务器，默认"pathway-mcp-server"；
• transport：传输方式，默认"streamable-http"；源码中"stdio"也存在但被标记为“不稳定且实验性”，选择它会发出警告且不允许设置 host/port；
• host/port：服务器绑定地址；streamable-http模式下二者必填，缺失会抛ValueError；
• serve：要暴露的McpServable实例列表。

工具的“单行契约”

工具函数必须满足以下约束（官方文档明确要求，底层由引擎的请求/响应管道强制）：

1. 方法有两个参数：self和一张pw.Table（如input_from_client）。该表的 schema 即你在注册时传入的schema，且客户端的一次调用对应表中的一行；客户端传入的每个参数放在同名列中。
2. 返回值必须是一张带result列、单行、且 ID 与输入行相同的表，用于把计算结果回传给客户端。
3. 暴露方式为McpServable.register_mcp(server)中调用server.tool(...)，传入三个核心参数：工具在 MCP Server 中的名称、request_handler（处理方法）、schema（客户端输入 schema）。

从源码结构看，这份契约是这样落地的：McpServer.tool()（mcp_server.py）内部创建_McpServerSubject，再用pw.io.python.read(subject=..., schema=schema, format="json", autocommit_duration_ms=50)把 HTTP 请求流“物化”成一张pw.Table交给request_handler，处理后的表再经 response writer 序列化回写。请求体在交给引擎前会做json.dumps，且_McpServerSubject._verify_payload会校验 schema 中“无默认值”的列是否都有提供——这就是为什么请求参数必须与pw.Schema的列一一对应。

server.tool()除三个核心参数外还支持一批可选参数，可用于精细化控制工具行为：

另外，源码中的_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(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="Streamable MCP Server", transport="streamable-http", host="localhost", port=8123, serve=[function_to_serve], ) pw.run()

要点拆解：

• EmptyRequestSchema没有列，表示该工具不接收任何参数；
• get_constant_value基于输入表select(result=1)，天然保留了输入行的 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(name="get_constant_value", arguments={}) print(result) asyncio.run(main())

list_tools列出服务器上所有工具；call_tool(name=..., arguments={...})调用指定工具，arguments是与pw.Schema各列对应的字典。仓库的集成测试 test_mcp_server.py 采用了同样的验证方式：用multiprocessing子进程拉起McpServer，fastmcp.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(result=pw.this.x + pw.this.y) return results def register_mcp(self, server: McpServer): server.tool( "add", request_handler=self.add, schema=AddRequestSchema, ) function_to_serve = AddTool()

客户端调用时传入{"x": 4, "y": 6}：

async with client: result = await client.call_tool(name="add", arguments={"x": 4, "y": 6}) print(result)

注意select(result=pw.this.x + pw.this.y)直接对输入行做列运算，结果表仍为单行且 ID 不变，无需任何额外处理。

示例三：同一个 Server 暴露多个工具

两种方式效果完全等价。

方式 A：多个 servable 实例放入serve列表

constant_tool = ConstantValueTool() add_tool = AddTool() pathway_mcp_server = PathwayMcp( name="Streamable MCP Server", transport="streamable-http", host="localhost", port=8123, 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(result=1) 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(result=pw.this.x + pw.this.y) return results def register_mcp(self, server: McpServer): server.tool( "get_constant_value", request_handler=self.get_constant_value, schema=EmptyRequestSchema, ) server.tool( "add", request_handler=self.add, schema=AddRequestSchema, ) pathway_mcp_server = PathwayMcp( name="Streamable MCP Server", transport="streamable-http", host="localhost", port=8123, serve=[BasicTools()], ) pw.run()

两种方式最终list_tools都能同时看到get_constant_value与add，客户端逐个调用即可。

示例四：统计实时表的行数（实时能力的体现）

前几个例子的结果都是“静态”的。Pathway 的看点在于：工具可以读取一张持续更新的实时表。先用pw.demo.range_stream生成一张合成流——每秒新增一行，value列从 0 到 49：

table = pw.demo.range_stream(nb_rows=50)

然后写一个统计行数的工具：

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(count=pw.reducers.count()) results = empty_row.join_left(single_row_table, id=empty_row.id).select( count=pw.right.count ) results = results.select( result=pw.if_else(pw.this.count.is_none(), 0, pw.this.count) ) return results def register_mcp(self, server: McpServer): server.tool( "get_count", request_handler=self.get_count, schema=InputEmptyRequestSchema, # 空 schema ) function_to_serve = CountTool()

这段代码集中体现了“单行契约”的工程细节，逐行解释：

1. 不能直接返回table：返回表必须与输入行 ID 相同的单行表，而table是持续增长的多行表。正确做法是先聚合成单行表，再把聚合值“挂回”到客户端输入行上；
2. table.reduce(count=pw.reducers.count())得到一张至多一行的计数表；
3. 因为表可能为空，计数表也可能是空的，所以必须用left join（empty_row.join_left(single_row_table, id=empty_row.id)）保证客户端行一定存在，此时count为None；id=empty_row.id正是保留输入行 ID 的关键；
4. 最后用pw.if_else(... is_none(), 0, ...)把空表情况归一化为0——服务器在表非空时返回实时计数，否则返回0。

客户端调用：

async with client: result = await client.call_tool(name="get_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_rows=50) 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 f"count: {count}, min: {minimum}, max: {maximum}, avg: {avg}, latest: {latest}" single_row_table = table.groupby().reduce( count=pw.reducers.count(pw.this.value), min=pw.reducers.min(pw.this.value), max=pw.reducers.max(pw.this.value), avg=pw.reducers.avg(pw.this.value), latest=pw.reducers.latest(pw.this.value), ) single_cell_table = single_row_table.select( single_cell=statistics_udf( pw.this.count, pw.this.min, pw.this.max, pw.this.avg, pw.this.latest, ) ) results = empty_row.join_left(single_cell_table, id=empty_row.id).select( single_cell=pw.right.single_cell ) results = results.select( result=pw.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_handler=self.get_statistics, schema=ValueRequestSchema, ) function_to_serve = StatisticsTool() pathway_mcp_server = PathwayMcp( name="Streamable MCP Server", transport="streamable-http", host="localhost", port=8123, serve=[function_to_serve], ) pw.run( monitoring_level=pw.MonitoringLevel.NONE, terminate_on_error=False, )

说明与注意事项：

• 工具不要求任何输入，因此input_from_client是一张只有id列的单行表；示例中empty_row指的就是这张输入表；
• groupby().reduce(...)用五个 reducer 一次性完成聚合，@pw.udf把数字格式化成自然语言字符串返回；也可以改成 JSON 结构，方便客户端二次计算；
• pw.run(monitoring_level=pw.MonitoringLevel.NONE, terminate_on_error=False)关闭监控面板输出、避免单点错误终止进程，适合长期运行的服务场景；
• 与 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(name="get_statistics", arguments={}) print(result) asyncio.run(main())

这些统计值会随底层实时表持续演化，MCP 客户端拿到的始终是最新数据。

进阶：把 DocumentStore 暴露为 MCP 工具

文档索引是 RAG 与 agent 管线的核心：索引的组织方式决定了信息能否被快速检索取回。Pathway 的DocumentStore（python/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

组件说明：

• $sources：pw.io.fs.read监听data目录，format: binary保证data列是原始字节（DocumentStore要求docs表含 bytes 类型的data列），with_metadata: true额外产出用于过滤的_metadata列；
• $embedder：OpenAI 嵌入模型，配DefaultCache避免重复请求嵌入接口；
• $splitter：按 token 数切分（250～600 token）；$parser：Docling 解析器负责多格式文档转纯文本；
• $knn_index+$bm25_index：向量 KNN（余弦相似度）与 Tantivy BM25 关键词索引，由HybridIndexFactory组合成混合检索；
• $document_store：消费上面所有组件，构建解析 → 切分 → 嵌入 → 建索引的流式管道；
• mcp_http：把$document_store注册为 MCP Server，streamable-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