news_and_weather.py

@app.get("/news-and-weather")
@contextlib.asynccontextmanager
async def news_and_weather() -> AsyncGenerator[MemoryObjectRecieveStream[bytes]]:
    async with anyio.create_task_group() as tg:
        tx, rx = anyio.create_memory_object_stream[bytes]()
        with tx, rx:
             tg.start_soon(ws_stream, "ws://example.com/news", tx.clone())
             tg.start_soon(ws_stream, "ws://example.com/weather", tx.clone())
             tx.close()
             yield rx

Just for completeness, in case anyone else comes here checking for this discussion that we finished on chat, here's the version that should work fine.

The trick is that the task group is created in a dependency with yield, that is converted by FastAPI to an async context manager, and the exit code is run after the response is done, so nothing ever "yields out" of the task group:

import random
from collections.abc import AsyncIterator
from typing import Annotated

import anyio
from anyio.abc import TaskGroup
from anyio.streams.memory import MemoryObjectSendStream
from fastapi import Depends, FastAPI

app = FastAPI()


async def fake_stream(url: str, tx: MemoryObjectSendStream[bytes]) -> None:
    i = 0
    while True:
        i += 1
        await tx.send(f"{url} message {i}".encode())
        await anyio.sleep(1 + 1 * random.random())


async def task_group() -> AsyncIterator[TaskGroup]:
    async with anyio.create_task_group() as tg:
        yield tg
        tg.cancel_scope.cancel()


@app.get("/news-and-weather")
async def news_and_weather(
    tg: Annotated[TaskGroup, Depends(task_group, scope="request")],
) -> AsyncIterator[bytes]:
    tx, rx = anyio.create_memory_object_stream[bytes]()
    async with tx, rx:
        tg.start_soon(fake_stream, "ws://example.com/news", tx.clone())
        tg.start_soon(fake_stream, "ws://example.com/weather", tx.clone())
        tx.close()
        async for message in rx:
            yield message