Lifespan Protocol

The Message AMGI lifespan sub-specification outlines how applications are started up and shut down within AMGI.

It follows the same semantics and event flow as the ASGI lifespan specification. This allows servers and applications to coordinate resource allocation, background tasks, and graceful shutdown.

A simple implementation would be:

async def app(scope, receive, send):
    if scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                ...  # Do some startup here!
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                ...  # Do some shutdown here!
                await send({"type": "lifespan.shutdown.complete"})
                return
    else:
        pass  # Handle other types

Lifespan

A lifespan session has a single lifespan scope. Your application will be called exactly once for the duration of the process lifetime.

The lifespan scope information passed in scope contains:

  • scope["type"] (Literal['lifespan'])

  • scope["amgi"]["spec_version"] (str) – Version of the AMGI message spec this server understands-

  • scope["amgi"]["version"] (Literal['1.0']) – Version of the AMGI spec

  • scope["state"] (NotRequired[dict[str, Any]]) – An empty namespace where the application can persist state to be used when handling subsequent requests. Optional; if missing the server does not support this feature.

Lifespan startup - receive() event

Sent to the application to indicate that the server is starting up.

Applications should perform any required initialization at this point (for example: opening connections, warming caches, or starting background tasks).

Keys:

  • message["type"] (Literal['lifespan.startup'])

Lifespan startup complete - send() event

Sent by the application to signify that startup has completed successfully.

Once this event is sent, the server may begin delivering messages to the application.

Keys:

  • message["type"] (Literal['lifespan.startup.complete'])

Lifespan startup failed - send() event

Sent by the application to signify that startup has failed.

If this event is sent, the server must not start message processing and should terminate the application.

Keys:

  • message["type"] (Literal['lifespan.startup.failed'])

  • message["message"] (str)

Lifespan shutdown - receive() event

Sent to the application to indicate that the server is shutting down.

Applications should begin graceful shutdown at this point (for example: stopping background tasks, closing connections, and flushing buffers).

Keys:

  • message["type"] (Literal['lifespan.shutdown'])

Lifespan shutdown complete - send() event

Sent by the application to signify that shutdown has completed successfully.

Once this event is sent, the server may safely terminate the application.

Keys:

  • message["type"] (Literal['lifespan.shutdown.complete'])

Lifespan shutdown failed - send() event

Sent by the application to signify that shutdown has failed.

The server should still terminate the application, but may log or surface the failure as appropriate.

Keys:

  • message["type"] (Literal['lifespan.shutdown.failed'])

  • message["message"] (str)