AMGI (Asynchronous Messaging Gateway Interface) is the message-driven sibling of ASGI.
Where ASGI standardises the interface between Python applications and HTTP servers, AMGI standardises the interface between Python applications and message brokers.
It enables you to build broker-agnostic, strongly-typed, message-driven applications that are portable across protocols and compute environments without rewriting business logic.
AMGI defines a clean separation between:
- Your application logic
- The framework layer (e.g. AsyncFast)
- The AMGI interface contract
- The underlying protocol implementation
Protocol implementations may include Kafka, SQS, MQTT, Redis, and others - but your application code remains unchanged.
Modern systems are message-driven - but each broker has:
- Different client libraries
- Different runtime models
- Different infrastructure constraints
- Different deployment assumptions
AMGI introduces a minimal interface boundary so that:
- Switching brokers does not require rewriting your application
- Running in Lambda is no more complex than running on EC2
- Local development mirrors production semantics
If it runs on one protocol, it should run on another.
This repository also includes AsyncFast, a typed microframework built on AMGI, inspired by FastAPI.
If FastAPI made HTTP APIs easy to build, AsyncFast aims to do the same for message-driven systems.
from typing import Annotated
from asyncfast import AsyncFast
from asyncfast import Header
from pydantic import BaseModel
app = AsyncFast()
class UserCreated(BaseModel):
id: int
email: str
@app.channel("user.created")
async def handle_user_created(
payload: UserCreated,
correlation_id: Annotated[str, Header()],
) -> None:
print(f"User created: {payload.email} ({correlation_id})")What you get automatically:
- Typed payload validation
- Typed headers
- Clean dependency injection
- AsyncAPI-aligned schema generation
- Broker portability
from amgi_aiokafka import Server
server = Server(
app, "my-topic", bootstrap_servers="localhost:9092", group_id="my-group"
)
server.run()or
asyncfast run amgi-aiokafka main:app my-topic --group-id my-group
from amgi_sqs_event_source_mapping import SqsEventSourceMappingHandler
handler = SqsEventSourceMappingHandler(app)The application code remains unchanged.
At its core, AMGI defines a minimal, low-level callable interface, similar in spirit to ASGI:
async def app(scope, receive, send):
try:
# Do some message handling here!
await send(
{
"type": "message.ack",
}
)
except Exception as e:
await send({"type": "message.nack", "message": str(e)})AMGI provides:
- A standard application callable
- A structured message scope
- A send mechanism
Frameworks like AsyncFast can be built on top of this interface.
Applications written against AMGI:
- Are independent of a specific broker
- Are independent of compute environment
- Can run in containers, VMs, or Lambda
Infrastructure changes should not rewrite business logic.
AMGI and AsyncFast are aligned with:
- AsyncAPI
- JSON Schema
- Strong Python typing (Pydantic v2)
Schemas are first-class. Validation and documentation are built into the design.
Each protocol implementation aims to be:
- Minimal
- Explicit
- Correct
- Efficient
This is engineered abstraction.
The following AMGI servers are currently available:
- amgi-aiokafka - Kafka server implementation
- amgi-kafka-event-source-mapping - AWS Lambda integration for Kafka
- amgi-paho-mqtt - MQTT server implementation
- amgi-redis - Redis server implementation
- amgi-aiobotocore - Contains a SQS server implementation
- amgi-sqs-event-source-mapping - AWS Lambda integration for SQS
-
AMGI Specification
https://amgi.readthedocs.io/en/latest/ -
AsyncFast Documentation
https://asyncfast.readthedocs.io/en/latest/
For questions or suggestions, please contact jack.burridge@mail.com.