@uriyyo You could use this

import contextlib
from collections.abc import AsyncIterator, Callable, Sequence
from contextlib import AbstractAsyncContextManager

from fastapi import FastAPI


@contextlib.asynccontextmanager
async def _manager(
    app: FastAPI,
    lifespans: Sequence[Callable[[FastAPI], AbstractAsyncContextManager[None]]],
) -> AsyncIterator[None]:
    exit_stack = contextlib.AsyncExitStack()
    async with exit_stack:
        for lifespan in lifespans:
            await exit_stack.enter_async_context(lifespan(app))
        yield


class Lifespans:
    def __init__(
        self,
        lifespans: Sequence[Callable[[FastAPI], AbstractAsyncContextManager[None]]],
    ) -> None:
        self.lifespans = lifespans

    def __call__(self, app: FastAPI) -> AbstractAsyncContextManager[None]:
        self.app = app
        return _manager(app, lifespans=self.lifespans)


@contextlib.asynccontextmanager
async def one(app: FastAPI) -> AsyncIterator[None]:
    print("One")
    yield
    print("End One")


@contextlib.asynccontextmanager
async def two(app: FastAPI) -> AsyncIterator[None]:
    print("Two")
    yield
    print("End Two")


app = FastAPI(lifespan=Lifespans([one, two]))

Thanks both for your help and inspiration! I wouldnt have found my solution without you...
I would like to suggest a (maybe) simpler approach though

from contextlib import asynccontextmanager, AsyncExitStack
from typing import List
from fastapi import FastAPI, APIRouter

def app_lifespan(lifespans: List):
    @asynccontextmanager
    async def _lifespan_manager(app: FastAPI):
        exit_stack = AsyncExitStack()
        async with exit_stack:
            for lifespan in lifespans:
                await exit_stack.enter_async_context(lifespan(app))
            yield
            
    return _lifespan_manager

@asynccontextmanager
async def one(app: FastAPI) -> AsyncIterator[None]:
    print("One")
    yield
    print("End One")


@asynccontextmanager
async def two(app: FastAPI) -> AsyncIterator[None]:
    print("Two")
    yield
    print("End Two")


app = FastAPI(lifespan=app_lifespan([one, two]))

I confess I did not run this example, but the code is ripped from a working version that I am using....so apologies for any oversights.
I hope this is useful/helpful to someone.

The solutions above discuss multiple lifespan functions at the top level. But for mounted sub-applications, I've found that lifespan handlers are only called on the top-level application. In my case, I would like two sub-applications mounted at /v1 and /v2, each with their own lifespan logic.

You could imagine the top-level lifespan handler might look like this, with a cascade_lifespan() returning a context manager that sends lifespan.startup to our child apps as we enter it and lifespan.shutdown when we exit:

@contextlib.asynccontextmanager
async def top_lifespan(app):
    async with cascade_lifespan(app):
        # now all of our children have already started up
        yield
        # now all of our children are going to shut down

A simple version looks like this:

@contextlib.asynccontextmanager
async def cascade_lifespan(parent):
    async with contextlib.AsyncExitStack() as stack:
        for route in parent.router.routes:
            if isinstance(route, fastapi.routing.Mount):
                await stack.enter_async_context(
                    route.app.router.lifespan_context(route.app)
                )
        yield    

But this is unfortunately too simple and breaks down:

1. It assumes every mount is a starlette.routing.Router and relies on implementation details of this class. But it can be any ASGI application callable.

   What we really want to do is have access to inject events into each callable's receive() queue.

2. Per the spec, an application may raise an exception:

   If an exception is raised when calling the application callable with a lifespan.startup message or a scope with type lifespan, the server must continue but not send any lifespan events.

   This means one child's exception must not stop the other children, and we need to avoid sending lifespan.shutdown later on.

   1. Since we are calling lifespan_context() directly, any exceptions that do get raised will be caught by the top-level starlette.routing.Router.lifespan() function that called us, which probably confuses things more.

3. The child application's lifespan handler can yield a state dictionary that should be passed to later requests. This is just dropped by this dummy implementation.

   It would be nice, maybe, if the state were 'namespaced' to the application, so that the app's request handler only receives its own state. However the spec specifically says the ASGI server can't meddle with the state dictionary, and that "applications should be cooperative by properly naming their keys such that they will not collide with other frameworks or middleware."

@contextlib.asynccontextmanager
async def memory_stream_context(buffer_size):
    send, receive = anyio.create_memory_object_stream(buffer_size)
    try:
        yield send, receive
    finally:
        await send.aclose()
        await receive.aclose()


def _collect_subapps(app):
    for r in getattr(app, 'routes', ()):
        if isinstance(r, fastapi.routing.Mount):
            yield from _collect_subapps(r.app)  # comment out if you don't want recursive enumeration
            yield r.app


def _lifespan_driver(app, tg):
    @contextlib.asynccontextmanager
    async def _cm():
        scope = {
            'type': 'lifespan',
            'asgi': {'version': '3.0', 'spec_version': '1.0'},
            'state': {},
        }

        async with (
            memory_stream_context(0) as (
                drv_send,
                app_recv,
            ),
            memory_stream_context(0) as (
                app_send,
                drv_recv,
            ),
        ):
            tg.start_soon(app, scope, app_recv.receive, app_send.send)

            # startup
            await drv_send.send({'type': 'lifespan.startup'})
            if (msg := await drv_recv.receive())[
                'type'
            ] != 'lifespan.startup.complete':
                raise RuntimeError(f'startup failed: {msg}')

            try:
                yield
            finally:
                # shutdown
                await drv_send.send({'type': 'lifespan.shutdown'})
                if (msg := await drv_recv.receive())[
                    'type'
                ] != 'lifespan.shutdown.complete':
                    raise RuntimeError(f'shutdown failed: {msg}')

    return _cm()

@contextlib.asynccontextmanager
async def cascade_lifespan(root_app):
    async with (
        anyio.create_task_group() as tg,
        contextlib.AsyncExitStack() as stack,
    ):
        for sub in _collect_subapps(root_app):  # inner → outer
            await stack.enter_async_context(_lifespan_driver(sub, tg))
        yield

Or ... undeprecate the original event handlers which are, anyway, a cleaner syntax.

The stated reason for deprecation is that global (or, at least, module level) variables are required to coordinate startup/shutdown work. This is not a compelling rationale when the service-level features typically access singleton objects (as shown in all the examples!). Especially in a language with weak member access controls like Python.

In short, it wasn't broken, why non un-fix it?