kontiki 1.5.0

Part of the Kontiki suite — a compact open-source stack for startups and small teams that need ops without the heavy stack.

Full suite overview → https://kontiki-org.github.io/

Ops demo → kontiki-monitor Quickstart.

Overview

Kontiki is a Python microservices framework built on AMQP (aio-pika) and asyncio.

• Write only your business logic: Kontiki manages connections to RabbitMQ, message routing (RPC, events, sessions, broadcast), service lifecycle and configuration merge for you.
• Design service interactions as messages: combine RPC, events, broadcast and per‑instance sessions to describe how services collaborate.
• Configuration‑driven: merged YAML config and a unified runner (cli.run) to start services the same way in development and production.
• Optional service registry: heartbeats, degraded state (degraded_on), and basic observability (service status, events, exceptions).
• Testability-first API: integration helpers (kontiki.testing) are designed to make end-to-end scenarios easy to express with Behave (mock services, captured events/RPC/HTTP, synchronous test runner).

For a detailed overview of all features, see docs/features.md. For patterns that are easy to miss (registry health, flow_id, delivery modes, testing, …), see docs/advanced-features.md.

Quickstart

Install Kontiki (via pip or Poetry):

pip install kontiki

Define a simple service:

from kontiki.delegate import ServiceDelegate
from kontiki.messaging import Messenger, on_event, rpc


class MyDelegate(ServiceDelegate):
    async def setup(self):
        # init from self.container.config
        pass

    async def start(self):
        # optional: start background tasks / open connections
        pass

    async def stop(self):
        # optional: stop background tasks / close connections
        pass

    def do_something(self, x):
        # business logic
        return x * 2

    async def handle_thing(self, payload):
        # business logic
        return {"processed": payload}


class MyService:
    name = "compute-api"  # optional: if omitted, the class name "MyService" is used
    delegate = MyDelegate()
    messenger = Messenger()  # delegate: publish events, RPC to other services

    @rpc
    async def compute(self, x):
        return self.delegate.do_something(x)

    @on_event("thing_happened")
    async def on_thing(self, payload):
        result = await self.delegate.handle_thing(payload)
        await self.messenger.publish("thing_processed", result)

def run():
    cli.run(MyService, "Example Kontiki service.", version="0.1.0")

Expose it as a CLI command in pyproject.toml:

[tool.poetry.scripts]
my_service = "myapp.main:run"

Run your service:

my_service --config config.yaml

Kontiki relies on RabbitMQ. For local development and to run the examples, you can start a RabbitMQ instance via Docker with:

make run-amqp

Documentation

• Features: docs/features.md
• Advanced features (patterns & gotchas): docs/advanced-features.md
• Configuration reference: docs/configuration.md
• Example configuration: docs/kontiki-config.example.yaml
• Contributing guidelines: CONTRIBUTING.md
• License: LICENSE

Examples

Examples can be run via the Makefile (see targets such as run-rpc-service, run-rpc-example, run-simple-events-service, etc.).

You can also have a look at the following repository kontiki-scheduler. It's an example scheduler service built with Kontiki.

Monitoring (KontikiTUI)

When services register with the Kontiki registry, you can use KontikiTUI: a small terminal UI to browse registered services, inspect registry events and exceptions, and tail log files from the terminal.

Misc

Kontiki did not come out of a naming workshop but from the album Kontiki by the band Cotton Mather.
If you enjoy vintage 4-track indie pop as much as microservices, you should check it out.