kafklient

Async Kafka utilities built on confluent-kafka (librdkafka).

This library wraps the synchronous Consumer/Producer with dedicated thread executors so Kafka operations do not block the event loop, and provides typed streams and RPC utilities via Parser[T].

What's inside

• KafkaListener: Subscribe to topics and stream parsed objects as TypeStream[T].
• KafkaRPC: Send requests and await responses matched by correlation id.
• KafkaRPCServer: Consume request topics and produce responses to reply topics specified in headers.
• MCP over Kafka (optional): Run MCP (JSON-RPC) over Kafka topics + stdio bridges via CLI (kafklient mcp-client / kafklient mcp-server).

This library supports consumer-group subscribe mode only. (Manual assign is intentionally not supported.)

Requirements

• Python >= 3.12
• A reachable Kafka broker
• confluent-kafka >= 2.15.1

Install

pip install kafklient

Optional extras:

# MCP server/bridge support
pip install "kafklient[mcp]"

# Dev tools (pyright/ruff)
pip install "kafklient[all]"

Local Kafka for development (optional)

This repository includes a single-node Kafka docker-compose.yml for local testing.

docker compose up -d

Default bootstrap server: 127.0.0.1:9092.

Core concepts

Consumer group (group.id)

• Same group.id: competing consumers / load balancing (each record is delivered to one member in the group).
• Different group.id: each instance receives the full stream (broadcast-style consumption).
• Important: do not share a group.id between different roles (e.g. RPC clients vs RPC servers).

Start-from-latest behavior (seek_to_end_on_assign)

With the default seek_to_end_on_assign=True, the consumer seeks to the end when partitions are assigned. This focuses processing on messages produced after the client becomes ready and reduces accidental reprocessing of old data.

To read from older offsets:

• seek_to_end_on_assign=False
• optionally set consumer_config["auto.offset.reset"] = "earliest"

Parser[T]

Parser[T] declares which topics to parse and what type to parse into.

• Recommended: provide a factory for JSON/custom binary payloads.
• factory can be sync or async; the input is a Message.

Examples

KafkaListener: consume as a typed stream

This example parses JSON into a Hello dataclass and consumes a TypeStream[Hello].

import asyncio
import json
from dataclasses import dataclass

from kafklient import ConsumerConfig, KafkaListener, Message, Parser, ProducerConfig


@dataclass(frozen=True, slots=True)
class Hello:
    message: str
    count: int


def parse_hello(rec: Message) -> Hello:
    raw = rec.value() or b"{}"
    data = json.loads(raw.decode("utf-8"))
    return Hello(
        message=str(data.get("message", "")),
        count=int(data.get("count", 0)),
    )


async def main() -> None:
    topic = "hello-events"

    consumer_config: ConsumerConfig = {
        "bootstrap.servers": "127.0.0.1:9092",
        "group.id": "hello-listener",
        "auto.offset.reset": "latest",
    }
    producer_config: ProducerConfig = {"bootstrap.servers": "127.0.0.1:9092"}

    async with KafkaListener(
        parsers=[Parser[Hello](topics=[topic], factory=parse_hello)],
        consumer_config=consumer_config,
        producer_config=producer_config,
        auto_create_topics=True,
    ) as listener:
        stream = await listener.subscribe(Hello)

        # Demo only: produce and consume in the same process
        await listener.produce(
            topic,
            json.dumps({"message": "hi", "count": 1}).encode("utf-8"),
            flush=True,
        )

        async def receive_one() -> Hello:
            async for item in stream:
                return item
            raise RuntimeError("stream stopped before receiving a message")

        msg = await asyncio.wait_for(receive_one(), timeout=5.0)
        print(msg)


if __name__ == "__main__":
    asyncio.run(main())

KafkaRPC + KafkaRPCServer: request/response

RPC follows these rules:

• Request topic: consumed by KafkaRPCServer (share the same group.id across servers to load-balance).
• Reply topic: consumed by KafkaRPC (clients typically should use a unique group.id).
• Reply routing: passed via one or more x-reply-topic headers on the request message.
• Correlation matching: by default uses the message key (or the x-corr-id header).

import asyncio
from dataclasses import dataclass

from kafklient import ConsumerConfig, KafkaRPC, KafkaRPCServer, Message, Parser, ProducerConfig


@dataclass(frozen=True, slots=True)
class EchoRequest:
    data: bytes


def parse_echo_request(msg: Message) -> EchoRequest:
    return EchoRequest(data=msg.value() or b"")


def parse_bytes(msg: Message) -> bytes:
    return msg.value() or b""


async def run_server(*, ready: asyncio.Event, stop: asyncio.Event) -> None:
    request_topic = "rpc-requests"

    server_consumer_config: ConsumerConfig = {
        "bootstrap.servers": "127.0.0.1:9092",
        "group.id": "rpc-server",
        "auto.offset.reset": "latest",
    }
    server_producer_config: ProducerConfig = {"bootstrap.servers": "127.0.0.1:9092"}

    server = KafkaRPCServer(
        parsers=[Parser[EchoRequest](topics=[request_topic], factory=parse_echo_request)],
        consumer_config=server_consumer_config,
        producer_config=server_producer_config,
        auto_create_topics=True,
    )

    @server.handler(EchoRequest)
    async def echo(req: EchoRequest, message: Message) -> bytes:  # pyright: ignore[reportUnusedFunction]
        return req.data

    await server.start()
    ready.set()
    try:
        await stop.wait()
    finally:
        await server.stop()


async def main() -> None:
    request_topic = "rpc-requests"
    reply_topic = "rpc-replies"

    server_ready = asyncio.Event()
    server_stop = asyncio.Event()
    server_task = asyncio.create_task(run_server(ready=server_ready, stop=server_stop))

    rpc_consumer_config: ConsumerConfig = {
        "bootstrap.servers": "127.0.0.1:9092",
        "group.id": "rpc-client-1",
        "auto.offset.reset": "latest",
    }
    rpc_producer_config: ProducerConfig = {"bootstrap.servers": "127.0.0.1:9092"}

    rpc = KafkaRPC(
        parsers=[Parser[bytes](topics=[reply_topic], factory=parse_bytes)],
        consumer_config=rpc_consumer_config,
        producer_config=rpc_producer_config,
        auto_create_topics=True,
    )

    try:
        await server_ready.wait()
        await rpc.start()

        res = await rpc.request(
            req_topic=request_topic,
            req_value=b"ping",
            req_headers_reply_to=[reply_topic],
            res_timeout=5.0,
            res_expect_type=bytes,
        )
        print(res)
    finally:
        await rpc.stop()
        server_stop.set()
        await server_task


if __name__ == "__main__":
    asyncio.run(main())

Automatic topic creation

All clients (KafkaListener/KafkaRPC/KafkaRPCServer) can create topics in one of two ways:

• auto_create_topics=True: create topics referenced by parsers on start (before subscribing).
• Manual call: await client.create_topics("a", "b", ...)

Options:

• topic_num_partitions (default: 1)
• topic_replication_factor (default: 1)

Internally this uses AdminClient and requires broker permissions.

MCP over Kafka (optional)

You can run MCP (JSON-RPC) over Kafka topics and connect any stdio-based MCP client via a bridge.

Install

pip install "kafklient[mcp]"

Server example (FastMCP + run_server)

import logging

from fastmcp import FastMCP

from kafklient.mcp.server import run_server

logging.basicConfig(level=logging.INFO)

mcp = FastMCP("My Kafka MCP Server")


@mcp.tool()
def echo(message: str) -> str:
    return f"Echo: {message}"


if __name__ == "__main__":
    run_server(
        mcp,
        bootstrap_servers="127.0.0.1:9092",
        consumer_topic="mcp-requests",
        producer_topic="mcp-responses",
        consumer_group_id="mcp-server",
        auto_create_topics=True,
        show_banner=False,
        log_level="info",
    )

CLI: server (kafklient mcp-server)

The CLI can load a FastMCP instance from a module or a Python file and run it over Kafka.

Supported --mcp formats:

• Module: mypkg.myserver:mcp (or mypkg.myserver:mcp.some_attr)
• File: ./myserver.py:mcp (or ./myserver.py:mcp.some_attr)
• If : is omitted, it defaults to :mcp (e.g. mypkg.myserver == mypkg.myserver:mcp)

# Module-based
kafklient mcp-server --mcp mypkg.myserver:mcp --bootstrap-servers 127.0.0.1:9092

# File-based
kafklient mcp-server --mcp ./myserver.py:mcp --bootstrap-servers 127.0.0.1:9092

CLI: stdio <-> Kafka bridge (kafklient mcp-client)

uv run kafklient mcp-client --bootstrap-servers 127.0.0.1:9092

Useful flags:

• --consumer-topic: topic to read responses/notifications from (default: mcp-responses)
• --producer-topic: topic to write requests to (default: mcp-requests)
• Session isolation is always enabled (responses are filtered using x-session-id).

For local MCP Inspector experiments against this checkout, run the bridge through uv --directory:

uv --directory C:/Projects/kafklient run kafklient mcp-client -b 127.0.0.1:9092

Development

Install the locked runtime, MCP dependencies, and development tools:

uv sync --locked --all-extras

Lint, format, and strict type-check (Python 3.12 baseline):

uv run --all-extras ruff check .
uv run --all-extras ruff format --check .
uv run --all-extras pyright

Run tests that do not require Kafka:

uv run --all-extras python -m unittest tests.test_broadcaster tests.test_parser tests.test_mcp_server_types tests.test_mcp_client_session_helpers.TestMcpClientSessionHelpers.test_inprocess_client_session_list_tools

If Kafka is running (e.g. after docker compose up -d), run the tests:

uv run --all-extras python -m tests

The test runner returns a nonzero exit code on failure. Set KAFKLIENT_TEST_BOOTSTRAP to use a different test broker.

Dependency compatibility

Python dependencies and uv.lock were refreshed on 2026-09-11. MCP support uses the latest compatible FastMCP 3.x (3.4.7) and MCP SDK 1.x (1.30.0). FastMCP 4 / MCP SDK 2 require a separate transport migration: the SDK removes mcp.server.FastMCP and changes JSON-RPC models and session APIs used by this library. The <4 and <2 bounds preserve both currently supported server implementations. See the FastMCP release notes.

License

MIT