Per-request context wrapping a `DispatchContext`. `ServerRunner` constructs one per inbound request and passes it to the user's handler.
| 25 | |
| 26 | |
| 27 | class BaseContext(Generic[TransportT]): |
| 28 | """Per-request context wrapping a `DispatchContext`. |
| 29 | |
| 30 | `ServerRunner` constructs one per inbound request and passes it to the |
| 31 | user's handler. |
| 32 | """ |
| 33 | |
| 34 | def __init__(self, dctx: DispatchContext[TransportT], meta: RequestParamsMeta | None = None) -> None: |
| 35 | self._dctx = dctx |
| 36 | self._meta = meta |
| 37 | |
| 38 | @property |
| 39 | def transport(self) -> TransportT: |
| 40 | """Transport-specific metadata for this inbound request.""" |
| 41 | return self._dctx.transport |
| 42 | |
| 43 | @property |
| 44 | def cancel_requested(self) -> anyio.Event: |
| 45 | """Set when the peer sends `notifications/cancelled` for this request.""" |
| 46 | return self._dctx.cancel_requested |
| 47 | |
| 48 | @property |
| 49 | def can_send_request(self) -> bool: |
| 50 | """Whether the back-channel can currently deliver server-initiated requests. |
| 51 | |
| 52 | `False` when the transport has no back-channel, or when the underlying |
| 53 | dispatch context has been closed because the inbound request finished. |
| 54 | """ |
| 55 | return self._dctx.can_send_request |
| 56 | |
| 57 | @property |
| 58 | def meta(self) -> RequestParamsMeta | None: |
| 59 | """The inbound request's `_meta` field, if present.""" |
| 60 | return self._meta |
| 61 | |
| 62 | async def send_raw_request( |
| 63 | self, |
| 64 | method: str, |
| 65 | params: Mapping[str, Any] | None, |
| 66 | opts: CallOptions | None = None, |
| 67 | ) -> dict[str, Any]: |
| 68 | """Send a request to the peer on the back-channel. |
| 69 | |
| 70 | Raises: |
| 71 | MCPError: The peer responded with an error. |
| 72 | NoBackChannelError: `can_send_request` is `False`. |
| 73 | """ |
| 74 | return await self._dctx.send_raw_request(method, params, opts) |
| 75 | |
| 76 | async def notify(self, method: str, params: Mapping[str, Any] | None, opts: CallOptions | None = None) -> None: |
| 77 | """Send a notification to the peer on the back-channel.""" |
| 78 | await self._dctx.notify(method, params, opts) |
| 79 | |
| 80 | async def report_progress(self, progress: float, total: float | None = None, message: str | None = None) -> None: |
| 81 | """Report progress for this request, if the peer supplied a progress token. |
| 82 | |
| 83 | A no-op when no token was supplied. |
| 84 | """ |
no outgoing calls