Function server_on_request

tests/shared/test_context.py:29–32

(ctx: DCtx, method: str, params: Mapping[str, Any] | None)

    captured: list[BaseContext[TransportContext]] = []

    async def server_on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
        bctx = BaseContext(ctx)
        captured.append(bctx)
        return {}

    async with running_pair(direct_pair, server_on_request=server_on_request) as (client, *_):
        with anyio.fail_after(5):