Function server_on_request

tests/server/test_server_context.py:36–39

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

    conn_holder: list[Connection] = []

    async def server_on_request(dctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
        ctx: Context[_Lifespan] = Context(dctx, lifespan=_Lifespan("app"), connection=conn_holder[0])
        captured.append(ctx)
        return {}

    async with running_pair(direct_pair, server_on_request=server_on_request) as (client, server, *_):
        conn_holder.append(Connection.for_loop(server, session_id="sess-1"))