Metadata-Version: 2.3
Name: kafklient
Version: 0.7.0
Summary: Async Kafka helpers built on confluent_kafka.
Requires-Dist: confluent-kafka>=2.15.1
Requires-Dist: pydantic>=2.13.5
Requires-Dist: typer>=0.27.2
Requires-Dist: kafklient[dev] ; extra == 'all'
Requires-Dist: kafklient[mcp] ; extra == 'all'
Requires-Dist: pyright>=1.1.414 ; extra == 'dev'
Requires-Dist: ruff>=0.16.7 ; extra == 'dev'
Requires-Dist: fastmcp>=4.0.3,<5 ; extra == 'mcp'
Requires-Dist: mcp>=2.2.0,<3 ; extra == 'mcp'
Requires-Dist: anyio>=4.15.1,<5 ; extra == 'mcp'
Requires-Dist: rich>=15.0.0 ; extra == 'mcp'
Requires-Python: >=3.12
Provides-Extra: all
Provides-Extra: dev
Provides-Extra: mcp
Description-Content-Type: text/markdown

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

```bash
pip install kafklient
```

Optional extras:

```bash
# 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.

```bash
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]`.

```python
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).

```python
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

```bash
pip install "kafklient[mcp]"
```

### Server example (`FastMCP` + `run_server`)

```python
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`)

```bash
# 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`)

```bash
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`:

```bash
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:

```bash
uv sync --locked --all-extras
```

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

```bash
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:

```bash
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:

```bash
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

| kafklient | FastMCP | MCP Python SDK |
| --- | --- | --- |
| 0.6.3 | `>=3.4.7,<4` | `>=1.30.0,<2` |
| 0.7.0 | `>=4.0.3,<5` | `>=2.2.0,<3` |

Python 3.12 remains supported. To keep an existing FastMCP 3 application on the compatible release:

```bash
uv add "kafklient[mcp]>=0.6.3,<0.7"
```

### Migrating to 0.7.0

- Import servers from `fastmcp`: SDK 2 no longer provides `mcp.server.FastMCP`.
- SDK result attributes use snake_case, such as `result.is_error` and `result.structured_content`.
  JSON sent over Kafka still uses the protocol's wire aliases.
- Sampling/elicitation/root callbacks receive `mcp.client.session.ClientRequestContext`.
  Notification handlers receive the concrete notification directly, without a `.root` wrapper.
- `kafka_client_session` and `inprocess_client_session` accept both seconds (`float`) and `timedelta`
  for `read_timeout_seconds`. Raw SDK 2 `ClientSession` accepts seconds.

Choose one protocol handshake per connection:

```python
from kafklient.mcp import kafka_client_session


async def call_echo() -> None:
    async with kafka_client_session("127.0.0.1:9092", read_timeout_seconds=10.0) as session:
        await session.discover()  # Modern MCP protocol (2026-07-28)
        result = await session.call_tool("echo", {"message": "hello"})
        print(result.is_error, result.content)
```

Use `await session.initialize()` for the legacy handshake instead. Both modes work over Kafka and the
stdio bridge; the server selects the mode from the client's first request. Legacy-mode sampling,
elicitation, and resource-subscription behavior is covered by integration tests. Those legacy APIs are
deprecated by SDK 2 and are not available on the modern protocol. Modern interactive tools use
`InputRequiredResult` with `input_responses` and `request_state` on the next call.

See the [FastMCP 4 upgrade guide](https://gofastmcp.com/getting-started/upgrading/from-fastmcp-3)
for application-level changes.

## License

MIT

