{"id":46817798,"title":"Show HN: AsyncFast – a typed, message-driven framework inspired by FastAPI","points":1,"user":"jackwburridge","time":1769726275,"time_ago":"2 hours ago","type":"link","content":"

Hi HN,

I've been working on AsyncFast, a Python framework for building message-driven APIs with FastAPI-style ergonomics.

The goal is to let you write handlers once and run them unchanged across different brokers and runtimes (Kafka, SQS, MQTT, Redis, containers, Lambda, etc.).

Under the hood it's built on a small spec (AMGI) that plays a similar role to ASGI, but for message-based systems instead of HTTP.

Minimal example

    from asyncfast import AsyncFast\n    from pydantic import BaseModel\n\n    app = AsyncFast()\n\n\n    class Payload(BaseModel):\n        id: str\n        name: str\n\n\n    @app.channel("channel")\n    async def handle_channel(payload: Payload) -> None:\n        print(payload)\n\n

\nRunning it locally (Kafka example)

    pip install asyncfast amgi-aiokafka\n    asyncfast run app:app kafka --bootstrap-servers localhost:9092\n\n

\nThe same application code can be run against different brokers by swapping the runtime adapter.

Key ideas:

- broker-agnostic application code

- explicit, typed address parameters / payloads / headers

- portability between long-running consumers and serverless environments

Docs: https://asyncfast.readthedocs.io

Source: https://github.com/asyncfast/amgi

AMGI: https://amgi.readthedocs.io

This is still early and evolving - I'd be especially interested in feedback from people who've built Kafka or event-driven systems and had to deal with portability, or tight broker coupling issues.","url":"https://asyncfast.readthedocs.io/en/latest/","domain":"asyncfast.readthedocs.io","comments":[],"comments_count":0}