Function echo_handlers

tests/shared/test_dispatcher.py:42–57

(recorder: Recorder)

def echo_handlers(recorder: Recorder) -> tuple[OnRequest, OnNotify]:
    async def on_request(
        ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
    ) -> dict[str, Any]:
        # Strip `_meta` so JSON-RPC and direct dispatch record identically:
        # the JSON-RPC outbound path always attaches `_meta` (otel injection).
        recorded = {k: v for k, v in (params or {}).items() if k != "_meta"} if params is not None else None
        recorder.requests.append((method, recorded))
        recorder.contexts.append(ctx)
        return {"echoed": method, "params": recorded or {}}

    async def on_notify(ctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None) -> None:
        recorder.notifications.append((method, params))
        recorder.notified.set()

    return on_request, on_notify


@asynccontextmanager