Skip to content

Getting Started

This guide takes you from an empty environment to a running stage: install Flechtwerk, learn the one contract every stage is built on, and run a complete stage against a Kafka broker with a single call. Everything is injected by the caller — the framework reads nothing from the environment.

Flechtwerk has exactly two stage shapes, both built on the same contract:

  • an Extractor brings an external source into Kafka — polling it on a timer, or receiving pushed messages with the MQTT Extractor — exactly-once from its resume cursor to Kafka;
  • a Transformer consumes Kafka topics and publishes derived records, with exactly-once delivery.

Read this page once for the basics — the contract and how any stage is run. Each stage guide then covers only what is specific to its shape, and Best Practices shows how the two shapes work together.

Prerequisites

  • Python 3.12+.
  • A running Kafka broker reachable at the bootstrap_servers you pass below — any broker works, and a local single-node cluster is fine for development.
  • The topics your stage needs, created up front. Which topics depend on the shape — each stage guide lists them. Flechtwerk reads their partition counts at startup, creates the compacted changelog topic for any stateful stage, and (if your broker has topic auto-creation enabled) writes output topics on first use.

Installation

pip install flechtwerk            # or: uv add flechtwerk
pip install "flechtwerk[mqtt]"    # with the MQTT→Kafka bridge (paho-mqtt)

Flechtwerk requires Python 3.12+. Its runtime dependencies are aiokafka[zstd], prometheus-client, reactor-di, and rocksdict.

The mqtt Extra Is Optional

Install flechtwerk[mqtt] only if you plan to build a push-driven source with the MQTT→Kafka bridge. It pulls in paho-mqtt, which stays confined to flechtwerk.mqtt — a plain import flechtwerk never loads it.

The Two-Yield Contract

The whole contract is two yield statements inside an async generator:

  • yield Message(...) to emit an output record.
  • yield State(...) to persist state for the current key.
    • yield <falsy State> to tombstone the key.

That's it. There are no agents, no tables, no DSL, no global app to register against, and no fluent builders. An async generator is already the right shape — pull-based, backpressure-friendly, naturally composable — and every Python developer already knows how to read one; a single decorator names its topics and turns it into a runnable stage.

Both stage shapes apply this contract — an Extractor once per poll cycle, a Transformer once per input record. The runner persists a State only when it differs from the current one, so a stage that never yields State is stateless and never opens a RocksDB file.

Running a Stage

Running a stage is one call, whatever its shape. Below is a complete, runnable program: a trivial identity transformer that forwards every input record to an output topic unchanged, wired to a broker. All configuration is injected — nothing is read from the environment.

import asyncio
from collections.abc import AsyncIterator

from flechtwerk import Flechtwerk, IncomingMessage, Message, State, transformer

@transformer(input_topics=["my-input"])
async def stage(msg: IncomingMessage, state: State) -> AsyncIterator[Message | State]:
    yield Message(key=msg.key, topic="my-output", value=msg.value)   # forward unchanged

async def main() -> None:
    await Flechtwerk.of(
        application_id="my-stage",
        bootstrap_servers="localhost:9092",
        client_id="my-stage-0",          # process identity: unique per instance, stable across restarts
        stage=stage,
    ).run()

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

Produce a record to my-input and it comes straight back out on my-output. That is the whole program. To do real work, swap in a stage from one of the guides below — the Flechtwerk.of(...).run() call is identical for every shape; only the stage you pass changes. The required knobs:

  • application_id — the stage's application identity: a transformer's Kafka consumer group, and the prefix of its changelog topic and transactional IDs.
  • bootstrap_servers — your broker.
  • client_id — the process identity (see the warning below).
  • stage — the Extractor or Transformer you've built.

Optional knobs: compression_type (defaults to "zstd" — JSON compresses ~13×; pass None to disable), max_poll_records (caps the records per consumer fetch — Kafka's max.poll.records, same default of 500; the cap is what makes a backlog drain as many ordinary batches instead of one giant one), metrics_port / metrics_labels (Prometheus; disabled while metrics_port is 0), mqtt (broker settings, used only by an MQTT Extractor), and poll_interval (a timedelta — an extractor's poll cadence, required for extractors, ignored by transformers). Like mqtt, the last two are shape-specific and may be passed unconditionally. Scaling needs no knob at all: extractor replicas shard the config set among themselves. See the API reference for the full signature.

Event Loop

Run it on uvloop for best throughput. The framework works on stock asyncio too (and therefore on Windows) — the event loop is the application's choice.

client_id Is the Process Identity

Give each instance a client_id that is unique per instance but stable across restarts (in Kubernetes, the pod name works well). It anchors the transactional producer's fencing and the MQTT session identity.

Next Steps

  • Extractors — bring an external source into Kafka on a timer.
  • MQTT Extractors — a push-driven extractor fed over MQTT.
  • Transformers — consume topics and publish derived records, with exactly-once delivery.
  • Best Practices — pair an extractor with a transformer into a raw-then-refined pipeline you can reprocess without re-ingesting.
  • Observability — Prometheus metrics for throughput, transactions, config arrival, and MQTT health.
  • Typed attributes & records — the Attribute library that keeps the JSON boundary honest.
  • Config topics — a shared, eventually-consistent lookup table for every instance (Kafka Streams' GlobalKTable).