MCP for real-time streaming: Kafka and Pulsar transports

Who this is for and what you can do after reading

This is for developers who want to connect MCP to message systems such as Kafka or Pulsar, and for teams that need tools inside a streaming data pipeline. After reading, you can follow the structure of two custom transports, see where the samples would fail in production, and list the fault-tolerance and monitoring work to finish before launch.

What real-time streaming is and where MCP fits

Real-time streaming means data is continuously transferred and processed as it is generated, so systems can react to new information immediately^[1]. Unlike batch processing, it works on data in motion, one item at a time as it arrives, rather than collecting a static dataset and computing over it later^[1].

Such systems usually face five problems: context loss, scalability, integration complexity, latency management and data consistency^[1]. The MCP design in the samples splits the work into five parts: context handlers, stream processors, protocol adapters, a context store and streaming connectors. The streaming connectors handle links to platforms such as Kafka, Pulsar and Kinesis^[1].

In code, one point comes first. The MCP layer here does not invent a new protocol on top of Kafka or Pulsar. It carries MCP's JSON-RPC messages through a message queue, which means implementing a custom transport^[1]. Once that is clear, the rest of the code is easier to follow.

Confirm the version and the scope

The samples say they match the MCP specification as of June 2025^[1]. The specification has changed since then, and protocol assumptions can differ between versions. Before copying anything, confirm which version you are targeting.

The benefits listed for IoT, financial trading and AI analytics, such as low latency and data consistency, come without benchmark figures or case data^[1]. Treat them as design goals rather than verified results.

Preparation: dependencies and local services

From the import statements, the Kafka sample needs confluent_kafka, the Pulsar sample needs the Pulsar Python client, and both rely on mcp modules^[1]. The samples give no install commands and do not pin a version of the mcp package, so verify both yourself.

For local testing, the Kafka sample connects to localhost:9092, and the Pulsar sample connects to pulsar://localhost:6650^[1]. Start the matching message service before you run the samples, or you will only see connection errors.

Steps

1. Set the topic directions. The Kafka sample reads from mcp-responses and writes to mcp-requests. The Pulsar sample reads from mcp-requests and writes to mcp-responses^[1]. Client and server directions must mirror each other, or each side waits for the other forever.
2. Start with the Kafka client. It creates a Kafka producer and consumer, with the consumer group ID mcp-client-group and auto.offset.reset set to earliest^[1].
3. When connect runs, the transport subscribes to the input topic and starts a background task named _consume_messages^[1]. This task is the core of receiving messages:

        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)

1. When the example runs, the client calls execute_tool with process_data and attaches source and timestamp metadata fields^[1].
2. Next, the Pulsar server. It creates a Pulsar client, a producer on the response topic and a Shared subscription named mcp-server-subscription on the request topic^[1]. It registers a tool named process_streaming_data, which requires the streaming capability^[1].

The shared problem in both receive loops

The two receive loops follow the same pattern. A background task keeps pulling messages, parses them as JSON-RPC, and places them on an asyncio queue^[1]. The Kafka version calls poll(1.0), sleeps 0.1 seconds when no message arrives, skips partition-end (EOF) errors, and only prints other errors^[1].

The problem is that poll and Pulsar's receive are synchronous blocking calls, yet they run inside asyncio coroutines. While they block, other coroutines on the event loop stall too. This is a structural flaw in the samples. When you adapt them, run the blocking calls in a thread pool or switch to an asynchronous client.

The Pulsar version handles errors in more detail. It pulls with receive(timeout_millis=500), acknowledges each message only after queuing it, and calls negative_acknowledge when processing fails, so the message can be delivered again^[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)

Timeouts and other exceptions both reach the outer except, which sleeps 0.1 seconds. A normal timeout and a real error therefore take the same path, and the logs cannot tell them apart. Handle timeouts separately when you troubleshoot.

Writing and closing

The Kafka write method serializes the message as JSON, produces it to the output topic and calls poll(0) to trigger delivery callbacks, which print success or failure^[1]. close cancels the background task, closes the consumer and flushes the producer so buffered messages are sent^[1].

The Pulsar write method calls producer.send with the JSON bytes^[1]. It attaches no delivery callback as the Kafka version does, so a failed send only appears in the outer except log line.

What to add before launch

The samples' deployment guidance covers dead-letter queues for failed messages, idempotent processors, appropriate buffer sizes with batching, backpressure, monitoring of stream metrics and context propagation, and encryption with authentication and access control^[1].

The two easiest items to skip are idempotency and dead-letter handling. The Pulsar sample negatively acknowledges failed messages, so they are redelivered. If the handler is not idempotent, one message can cause two side effects. The Kafka sample does not commit offsets explicitly, and a new consumer group reads from the earliest available message on first start, so historical data may be processed in full. Settle both points before launch.

Common problems

• The client never gets a response: check that the topic directions mirror each other, then confirm the message service is running.
• Messages are processed twice: usually a failed message was redelivered, or the consumer position was not committed in time. The Kafka sample does not commit explicitly, so you must manage offsets yourself.
• Messages are lost when the process exits: the Kafka sample's close flushes the producer, but a forcibly killed process gets no such chance.
• Import errors at runtime: check the mcp module paths and versions against the SDK you use.

Next steps

The official Apache Kafka documentation is the next reading, especially the parts on consumer groups, offset commits and dead-letter topics^[2].

Adapted from lesson material in Microsoft's open-source course *MCP for Beginners*, covering real-time streaming data in its advanced topics^[1], released under the MIT License.