首页
/ Pathway Live Data Framework MCP Server：把实时流处理引擎接入 LLM Agent 的完整实践

Pathway Live Data Framework MCP Server：把实时流处理引擎接入 LLM Agent 的完整实践

2026-09-04 21:00:46作者：殷蕙予
pathway
Python 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 中，共三个关键类：

类 职责
McpServable 抽象基类，任何要注册到 MCP Server 的对象都必须实现 register_mcp(server) 方法
McpServer 实现 MCP 协议的服务器本体，继承自 PathwayServer，底层用 FastMCP 承载工具注册与传输层
PathwayMcp 简化配置的 dataclass：构造时自动创建 McpServer 并把 serve 列表里的每个 servable 注册进去

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() 除三个核心参数外还支持一批可选参数，可用于精细化控制工具行为：

参数 默认 说明
name 必填 工具名
request_handler 必填 处理函数，签名必须是 (self, table) -> table
schema 必填 客户端输入 schema，用于生成工具 input schema
delete_completed_queries False 是否删除已完成的查询
cache_strategy None 可选缓存策略
title 缺省用 name 工具展示标题
description 缺省用处理函数 docstring 工具描述
output_schema 未设置 可选输出 schema
annotations None MCP 元注解（如 readOnlyHint、idempotentHint 等）
meta None 工具元数据
autocommit_duration_ms 50 两次 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(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 工作流。

参考文件

pathway
Python ETL framework for stream processing, real-time analytics, LLM pipelines, and RAG.
项目地址：https://gitcode.com/GitHub_Trending/pa/pathway
登录后查看全文

相关内容推荐

1 Pathway MCP Server 对接 Claude Desktop 实战：让 Claude 实时访问流式统计与文档库2 Pathway 实时 RAG 应用实战：用 DocumentStore 与 OpenAI 构建端到端的实时检索增强生成流水线3 Pathway LLM/RAG 实战示例详解：用 LLM xpack 构建实时 RAG、Adaptive RAG 与文档索引流水线4 Pathway Live Data Framework 入门全景：从 Schema、连接器到 Rust 引擎的实时数据管道5 Pathway 安装指南：pip 一键安装、可选依赖组、Docker 与 License Key 配置6 Pathway 实战：将 Python 网页爬虫改造为实时数据管道7 Pathway 入门指南：用 Python API + Rust 引擎构建实时流式 ETL 与 RAG 管道8 Pathway Live Data Framework 实例监控实战：OpenTelemetry Collector、Grafana Loki 接入与引擎遥测实现剖析9 Pathway 连接器体系详解：流式与静态双模式数据接入、更新语义与持久化恢复10 Pathway 开源框架许可指南：Community / Scale 双许可模型、资源限额与 License Key 激活机制详解
热门项目推荐
相关项目推荐

热门内容推荐

1 编程实践项目探索指南：从零构建技术能力体系2 技术解构式学习：从0到1构建你的编程知识体系3 构建自己的技术世界：build-your-own-x项目的实践探索指南4 解锁编程技能的实践之旅：从零构建你的技术世界5 技术实践探索：从零开始构建核心系统的实践指南6 亲手锻造技术引擎：从0到1构建核心系统的实践指南

最新内容推荐

Puppeteer CDPSession.send() 方法详解:类型安全的 CDP 命令调用、参数与超时控制机制gemini-cli 设置项文案规范：名词前置、正向布尔与动词精简的设计原则Crawl4AI Docker 部署的 /ask 文档上下文系统：c4ai-doc-context.md 如何为 AI 问答端点提供检索基础learn-claude-code s02 深度解析：Agent 从单一 Bash 扩展到 5 个工具的 TOOL_HANDLERS 分派机制全源码剖析claude-mem 错误处理反模式清理计划：132 项债务清单、检测器原理与零静默失败验收标准Starship 跨 Shell 提示符完全指南：安装、Shell 初始化接入与 init 底层机制解析Scrapling 爬虫代理管理与封锁处理：ProxyRotator 轮转与 Blocked 重试机制实战ruflo 多智能体代码审查蜂群：agent-code-review-swarm 技能的完整配置与实现解析graphify 增量更新与实体去重设计：从全量重建到内容哈希驱动的增量知识图谱Electron NavigationHistory API 详解：主进程侧导航栈的查询、跳转、编辑与恢复

项目优选

收起
kernelkernel
deepin linux kernel
C
33
18
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件，TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件，使 PyTorch 框架能够直接调用昇腾 NPU，为开发者提供昇腾 AI 处理器的超强算力。
Python
854
1.34 K
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库，实现网络在NPU上加速计算。
C++
1.12 K
2.72 K
kernelkernel
openEuler内核是openEuler操作系统的核心，既是系统性能与稳定性的基石，也是连接处理器、设备与服务的桥梁。
C
528
588
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库，实现网络在NPU上加速计算。
C++
906
1.83 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent，它能够将大语言模型的强大能力，通过你日常使用的各类通讯应用，直接延伸至你的指尖。
Python
3.53 K
1.01 K
cann-learning-hubcann-learning-hub
CANN 学习中心仓，支持在线互动运行、边学边练，提供教程、示例与优化方案，一站式助力昇腾开发者快速上手。
Jupyter Notebook
989
508
ops-mathops-math
本项目是CANN提供的数学类基础计算算子库，实现网络在NPU上加速计算。
C++
1.34 K
1.45 K
docsdocs
暂无描述
Markdown
892
5.79 K
openGauss-serveropenGauss-server
openGauss kernel ~ openGauss is an open source relational database management system
C++
213
313