Skip to content

Core API Reference

Configuration Classes

flowkit.core.config.CoordinatorConfig dataclass 

CoordinatorConfig(kafka_bootstrap: str = 'kafka:9092', worker_types: list[str] = (lambda: ['indexer', 'enricher', 'grouper', 'analyzer'])(), topic_cmd_fmt: str = 'cmd.{type}.v1', topic_status_fmt: str = 'status.{type}.v1', topic_worker_announce: str = 'workers.announce.v1', topic_query: str = 'query.tasks.v1', topic_reply: str = 'reply.tasks.v1', topic_signals: str = 'signals.v1', heartbeat_soft_sec: int = 300, heartbeat_hard_sec: int = 3600, lease_ttl_sec: int = 45, discovery_window_sec: int = 8, cancel_grace_sec: int = 30, scheduler_tick_sec: float = 1.0, finalizer_tick_sec: float = 5.0, hb_monitor_tick_sec: float = 10.0, outbox_dispatch_tick_sec: float = 0.25, outbox_max_retry: int = 12, outbox_backoff_min_ms: int = 250, outbox_backoff_max_ms: int = 60000, hb_soft_ms: int = 0, hb_hard_ms: int = 0, lease_ttl_ms: int = 0, discovery_window_ms: int = 0, cancel_grace_ms: int = 0, scheduler_tick_ms: int = 0, finalizer_tick_ms: int = 0, hb_monitor_tick_ms: int = 0, outbox_dispatch_tick_ms: int = 0)

flowkit.core.config.WorkerConfig dataclass 

WorkerConfig(kafka_bootstrap: str = 'kafka:9092', topic_cmd_fmt: str = 'cmd.{type}.v1', topic_status_fmt: str = 'status.{type}.v1', topic_worker_announce: str = 'workers.announce.v1', topic_query: str = 'query.tasks.v1', topic_reply: str = 'reply.tasks.v1', topic_signals: str = 'signals.v1', roles: list[str] = (lambda: ['echo'])(), worker_id: str | None = None, worker_version: str = '2.0.0', lease_ttl_sec: int = 60, hb_interval_sec: int = 20, announce_interval_sec: int = 60, dedup_cache_size: int = 10000, dedup_ttl_ms: int = 3600000, pull_poll_ms_default: int = 300, pull_empty_backoff_ms_max: int = 4000, db_cancel_poll_ms: int = 500, strict_input_adapters: bool = False, lease_ttl_ms: int = 60000, hb_interval_ms: int = 20000, announce_interval_ms: int = 60000)

Time and Clock

flowkit.core.time

Classes

SystemClock

Default production/test clock.

ManualClock 

ManualClock(start_ms: int = 0)

Bases: SystemClock

Simple controllable clock for tests. - Wall time is advanced manually (affects persistence fields). - Monotonic time mirrors wall unless overridden.

Source code in src/flowkit/core/time.py 
41
42
43
def __init__(self, start_ms: int = 0) -> None:
    self._wall = start_ms
    self._mono = start_ms

Utilities

flowkit.core.utils