MCP 实时流式数据

这篇适合谁、读完能做到什么

适合想把 MCP 接进 Kafka 或 Pulsar 这类消息系统的开发者，也适合需要在流式数据管道里调用工具的团队。读完你能看懂两套自定义传输的结构，知道示例在生产环境里会在哪里出问题，并列出上线前要补的容错和监控。

实时流处理是什么，MCP 在里面做什么

实时流处理指数据一边产生、一边被持续传输和处理，系统因此能立刻对新信息做出反应^[1]。它和批处理的区别在于：数据在流动中、按到达顺序逐条处理，而不是先攒成静态数据集再统一计算^[1]。

这类系统通常要同时面对五个问题：上下文丢失、扩展性、集成复杂度、延迟管理和数据一致性^[1]。示例中的 MCP 方案把整体拆成五个部分：上下文处理器、流处理器、协议适配器、上下文存储和流连接器。其中流连接器负责对接 Kafka、Pulsar、Kinesis 等平台^[1]。

落到代码上，要先分清一件事。这里的 MCP 并没有在 Kafka 或 Pulsar 之上另外发明一套协议，而是把 MCP 的 JSON-RPC 消息放进消息队列里传输，也就是实现一个自定义传输^[1]。理解了这一点，后面的代码就容易读。

先确认版本和范围

示例自称与 2025 年 6 月前后的 MCP 规范保持一致^[1]。规范之后又有过更新，不同版本的协议假设可能不同。照抄之前，先确认你使用的是哪个版本。

另外，物联网、金融交易和 AI 分析等场景的好处，例如低延迟和数据一致性，目前都没有给出测量数据或案例^[1]。把它们当作设计目标比较合适，不宜当作已经验证的结论。

准备：依赖和本地服务

从导入语句看，Kafka 示例需要 confluent_kafka，Pulsar 示例需要 Pulsar 的 Python 客户端，两者都还要用到 mcp 相关模块^[1]。示例没有给出安装命令，也没有写明 mcp 包的版本，需要自己确认。

本地联调时，Kafka 示例连接 localhost:9092，Pulsar 示例连接 pulsar://localhost:6650^[1]。先把对应的消息服务跑起来，再运行示例，否则只会看到连接错误。两个示例都假设只有一个本地节点，没有配置认证。生产环境的 Kafka 集群或 Pulsar 集群要用不同的地址和凭据连接，这部分示例没有涉及。

上手步骤

确定两个方向的主题。Kafka 示例从 mcp-responses 读、往 mcp-requests 写；Pulsar 示例则从 mcp-requests 读、往 mcp-responses 写^[1]。客户端和服务端的读写方向必须互为镜像，否则双方会互相等待。
先看 Kafka 客户端。它创建 Kafka 的生产者和消费者，消费者的 group ID 是 mcp-client-group，auto.offset.reset 设为 earliest^[1]。
调用 connect 时，传输会订阅输入主题，并启动后台任务 _consume_messages^[1]。接收消息的核心就在这个任务里：

        while self.running:
            try:
                msg = self.consumer.poll(1.0)
                if msg is None:
                    await asyncio.sleep(0.1)
                    continue
                
                if msg.error():
                    if msg.error().code() == KafkaError._PARTITION_EOF:
                        continue
                    print(f"Consumer error: {msg.error()}")
                    continue
                
                # Parse the message value as JSON-RPC
                try:
                    message_str = msg.value().decode('utf-8')
                    message_data = json.loads(message_str)
                    mcp_message = JsonRpcMessage.from_dict(message_data)
                    await self.message_queue.put(mcp_message)
                except Exception as e:
                    print(f"Error parsing message: {e}")
            except Exception as e:
                print(f"Error in consumer loop: {e}")
                await asyncio.sleep(1)

运行示例时，客户端会调用 execute_tool 执行 process_data，并附带 source 和 timestamp 两个元数据字段^[1]。
再看 Pulsar 服务端。它创建 Pulsar 客户端，在响应主题上建立生产者，并在请求主题上订阅名为 mcp-server-subscription 的 Shared 订阅^[1]。服务端注册了名为 process_streaming_data 的工具，该工具要求 streaming 能力^[1]。

验证链路时，先确认请求主题里出现了消息，再确认响应主题里出现了对应的回复。两段都有，才算打通。只看到其中一段，问题多半出在另一端的订阅或主题方向上。

两个接收循环的共同问题

两段接收循环写法相似：后台任务不断拉取消息，解析成 JSON-RPC 后放进 asyncio 队列^[1]。Kafka 版本调用 poll(1.0)，没有消息时休眠 0.1 秒再继续。它会跳过分区结束（partition EOF）的错误，其他错误只打印^[1]。

问题在于，poll 和 Pulsar 的 receive 都是同步阻塞调用，却放在 asyncio 协程里执行。阻塞期间，事件循环里的其他协程也会被卡住。这是示例的结构问题，改造时应当把阻塞调用放进线程池执行，或者换用对应的异步客户端。

Pulsar 版本的处理更细一些。它用 receive(timeout_millis=500) 拉取消息，入队之后才确认（acknowledge）；处理失败时调用 negative_acknowledge，让消息可以再次投递^[1]：

        while self.running:
            try:
                # Non-blocking receive with timeout
                msg = self.consumer.receive(timeout_millis=500)
                
                # Process the message
                try:
                    message_str = msg.data().decode('utf-8')
                    message_data = json.loads(message_str)
                    mcp_message = JsonRpcMessage.from_dict(message_data)
                    await self.message_queue.put(mcp_message)
                    
                    # Acknowledge the message
                    self.consumer.acknowledge(msg)
                except Exception as e:
                    print(f"Error processing message: {e}")
                    # Negative acknowledge if there was an error
                    self.consumer.negative_acknowledge(msg)
            except Exception as e:
                # Handle timeout or other exceptions
                await asyncio.sleep(0.1)

超时异常和其他异常都被外层的 except 捕获，然后统一休眠 0.1 秒。也就是说，正常的超时和真正的错误走的是同一条路径，日志里分不清它们。排查问题时，最好把超时单独处理。

写入和关闭

Kafka 的 write 把消息序列化成 JSON，发送到输出主题，并调用 poll(0) 触发投递回调，回调里打印成功或失败的结果^[1]。close 会取消后台任务、关闭消费者，并 flush 生产者，把缓冲区里的消息发出去^[1]。

Pulsar 的 write 直接调用 producer.send 发送 JSON 字节^[1]。它没有像 Kafka 那样挂投递回调，发送失败只能在外层的 except 里打印。

读取和写入是两条独立的路径，调试时要分开看。读取侧出问题，通常表现为服务端收不到请求；写入侧出问题，通常表现为客户端收不到响应。两边都要看日志，不能只盯着一端。

上线前要补的几块

示例的部署建议列出了几类工作：用死信队列（dead-letter queue）处理失败消息，让处理器具备幂等性，配置合适的缓冲区并使用批处理，实现背压，监控流处理指标和上下文传播，以及对敏感数据加密、做鉴权和访问控制^[1]。

其中最容易被忽略的是幂等和死信。Pulsar 示例在处理失败时会做 negative acknowledge，消息会被重投；如果处理逻辑不是幂等的，同一条消息就可能产生两次副作用。Kafka 示例没有显式提交偏移量，新消费组第一次启动时又会从最早的消息读起，历史数据可能被整体处理一遍。上线前要把这两点想清楚。

常见问题

客户端一直收不到响应：检查两端的主题方向是否互反，再确认消息服务已经启动。
Kafka 消费者读到的是旧消息：消费组一旦提交过位点，就从上次的位置继续；只有新的消费组、且没有已提交位点时，earliest 才会生效。换一个 group ID 会让它从头读。
消息被重复处理：多半是处理失败后被重投，或者消费位点没有及时提交。示例没有显式提交偏移量，需要自己管理。
进程退出时消息丢失：Kafka 示例的 close 会 flush 生产者，但如果进程被强制终止，就没有这一步的保障。
运行时报导入错误：示例依赖的 mcp 模块路径和版本，要对照你使用的 SDK 确认。先在干净的虚拟环境里安装依赖再试，能排除本机其他包的干扰。