Function on_request

tests/shared/test_jsonrpc_dispatcher.py:165–168

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

    handler_started = anyio.Event()

    async def on_request(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> dict[str, Any]:
        handler_started.set()
        await ctx.cancel_requested.wait()
        return {"completed": "after-cancel"}

    async def on_notify(ctx: DCtx, method: str, params: Mapping[str, Any] | None) -> None:
        pass  # the cancelled notification is teed here; nothing to observe