Answer a server-initiated request via the registered callbacks.
(
self, dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None
)
| 784 | await self.send_notification(types.RootsListChangedNotification()) |
| 785 | |
| 786 | async def _on_request( |
| 787 | self, dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None |
| 788 | ) -> dict[str, Any]: |
| 789 | """Answer a server-initiated request via the registered callbacks.""" |
| 790 | # Literal, not LATEST_PROTOCOL_VERSION: the fallback covers the initialize |
| 791 | # handshake (which only exists at <=2025) and stateless until the header |
| 792 | # is plumbed; its meaning is fixed regardless of LATEST bumps. |
| 793 | version = self._negotiated_version or "2025-11-25" |
| 794 | try: |
| 795 | request = cast(types.ServerRequest, _methods.parse_server_request(method, version, params)) |
| 796 | except KeyError: |
| 797 | raise MCPError(code=METHOD_NOT_FOUND, message="Method not found", data=method) from None |
| 798 | |
| 799 | response: types.ClientResult | types.ErrorData |
| 800 | if isinstance(request, types.PingRequest): |
| 801 | # Answered without a context: ping has no callback that would need one. |
| 802 | response = types.EmptyResult() |
| 803 | else: |
| 804 | assert dctx.request_id is not None # the callback-driving dispatchers always assign ids |
| 805 | ctx = ClientRequestContext( |
| 806 | session=self, request_id=dctx.request_id, meta=request.params.meta if request.params else None |
| 807 | ) |
| 808 | match request: |
| 809 | case types.CreateMessageRequest(params=sampling_params): |
| 810 | response = await self._sampling_callback(ctx, sampling_params) |
| 811 | case types.ElicitRequest(params=elicit_params): |
| 812 | response = await self._elicitation_callback(ctx, elicit_params) |
| 813 | case types.ListRootsRequest(): # pragma: no branch |
| 814 | response = await self._list_roots_callback(ctx) |
| 815 | client_response = ClientResponse.validate_python(response) |
| 816 | if isinstance(client_response, types.ErrorData): |
| 817 | raise MCPError.from_error_data(client_response) |
| 818 | dumped = client_response.model_dump(by_alias=True, mode="json", exclude_none=True) |
| 819 | try: |
| 820 | _methods.validate_client_result(method, version, dumped) |
| 821 | except ValidationError: |
| 822 | logger.exception("client callback for %r returned an invalid result", method) |
| 823 | raise MCPError(code=INTERNAL_ERROR, message="Client callback returned an invalid result") from None |
| 824 | return dumped |
| 825 | |
| 826 | async def _on_notify( |
| 827 | self, dctx: DispatchContext[TransportContext], method: str, params: Mapping[str, Any] | None |
nothing calls this directly
no test coverage detected