Skip to content

session

ServerSession: server-to-client requests and notifications.

A per-request proxy built by the kernel for each inbound request. Exposes the request-scoped outbound channel and the connection's standalone channel. Handlers reach it as ctx.session and use the typed helpers (elicit_form, send_log_message, ...) to call back to the client.

ServerSession

Per-request proxy for server-to-client requests and notifications.

Built once per inbound request by the kernel's _make_context. Holds two Outbound channels: the request-scoped one (the per-request DispatchContext, which on streamable HTTP routes onto the originating POST's response stream) and the connection's standalone channel (connection.outbound). related_request_id on the public methods is the selector — present means request-scoped, absent means standalone — and never crosses the Outbound Protocol.

Source code in src/mcp/server/session.py
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
class ServerSession:
    """Per-request proxy for server-to-client requests and notifications.

    Built once per inbound request by the kernel's `_make_context`. Holds two
    `Outbound` channels: the request-scoped one (the per-request
    `DispatchContext`, which on streamable HTTP routes onto the originating
    POST's response stream) and the connection's standalone channel
    (`connection.outbound`). `related_request_id` on the public methods is the
    selector — present means request-scoped, absent means standalone — and
    never crosses the `Outbound` Protocol.
    """

    def __init__(self, request_outbound: DispatchContext[Any], connection: Connection) -> None:
        self._request_outbound = request_outbound
        self._connection = connection

    @property
    def client_params(self) -> types.InitializeRequestParams | None:
        """The client's `initialize` request params; `None` when no client info was supplied."""
        return self._connection.client_params

    @property
    def protocol_version(self) -> str:
        """The protocol version this connection speaks.

        Populated at `Connection` construction and overwritten once the
        handshake commits on the loop path; never `None`.
        """
        return self._connection.protocol_version

    async def send_request(
        self,
        request: types.ServerRequest,
        result_type: type[ResultT],
        request_read_timeout_seconds: float | None = None,
        metadata: ServerMessageMetadata | None = None,
        progress_callback: ProgressFnT | None = None,
    ) -> ResultT:
        """Send a typed server-to-client request and validate the result.

        Raises:
            MCPError: The peer responded with an error.
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests (raised by the held `Outbound`).
            pydantic.ValidationError: The peer's result does not match `result_type`.
        """
        related = metadata.related_request_id if metadata is not None else None
        channel = self._request_outbound if related is not None else self._connection.outbound
        data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
        opts: CallOptions = {}
        if request_read_timeout_seconds is not None:
            opts["timeout"] = request_read_timeout_seconds
        if progress_callback is not None:
            opts["on_progress"] = progress_callback
        result = await channel.send_raw_request(data["method"], data.get("params"), opts or None)
        try:
            _methods.validate_client_result(request.method, self.protocol_version, result)
        except KeyError:
            pass
        return result_type.model_validate(result, by_name=False)

    async def send_notification(
        self,
        notification: types.ServerNotification,
        related_request_id: types.RequestId | None = None,
    ) -> None:
        """Send a typed server-to-client notification."""
        channel = self._request_outbound if related_request_id is not None else self._connection.outbound
        data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
        await channel.notify(data["method"], data.get("params"))

    def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
        """Check if the client supports a specific capability."""
        return self._connection.check_capability(capability)

    @deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def send_log_message(
        self,
        level: types.LoggingLevel,
        data: Any,
        logger: str | None = None,
        related_request_id: types.RequestId | None = None,
    ) -> None:
        """Send a log message notification."""
        await self.send_notification(
            types.LoggingMessageNotification(
                params=types.LoggingMessageNotificationParams(
                    level=level,
                    data=data,
                    logger=logger,
                ),
            ),
            related_request_id,
        )

    async def send_resource_updated(self, uri: str | AnyUrl) -> None:
        """Send a resource updated notification."""
        await self.send_notification(
            types.ResourceUpdatedNotification(
                params=types.ResourceUpdatedNotificationParams(uri=str(uri)),
            )
        )

    @overload
    @deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def create_message(
        self,
        messages: list[types.SamplingMessage],
        *,
        max_tokens: int,
        system_prompt: str | None = None,
        include_context: types.IncludeContext | None = None,
        temperature: float | None = None,
        stop_sequences: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        model_preferences: types.ModelPreferences | None = None,
        tools: None = None,
        tool_choice: types.ToolChoice | None = None,
        related_request_id: types.RequestId | None = None,
    ) -> types.CreateMessageResult:
        """Overload: Without tools, returns single content."""
        ...

    @overload
    @deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def create_message(
        self,
        messages: list[types.SamplingMessage],
        *,
        max_tokens: int,
        system_prompt: str | None = None,
        include_context: types.IncludeContext | None = None,
        temperature: float | None = None,
        stop_sequences: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        model_preferences: types.ModelPreferences | None = None,
        tools: list[types.Tool],
        tool_choice: types.ToolChoice | None = None,
        related_request_id: types.RequestId | None = None,
    ) -> types.CreateMessageResultWithTools:
        """Overload: With tools, returns array-capable content."""
        ...

    @deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def create_message(
        self,
        messages: list[types.SamplingMessage],
        *,
        max_tokens: int,
        system_prompt: str | None = None,
        include_context: types.IncludeContext | None = None,
        temperature: float | None = None,
        stop_sequences: list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        model_preferences: types.ModelPreferences | None = None,
        tools: list[types.Tool] | None = None,
        tool_choice: types.ToolChoice | None = None,
        related_request_id: types.RequestId | None = None,
    ) -> types.CreateMessageResult | types.CreateMessageResultWithTools:
        """Send a sampling/create_message request.

        Args:
            messages: The conversation messages to send.
            max_tokens: Maximum number of tokens to generate.
            system_prompt: Optional system prompt.
            include_context: Optional context inclusion setting.
                Should only be set to "thisServer" or "allServers"
                if the client has sampling.context capability.
            temperature: Optional sampling temperature.
            stop_sequences: Optional stop sequences.
            metadata: Optional metadata to pass through to the LLM provider.
            model_preferences: Optional model selection preferences.
            tools: Optional list of tools the LLM can use during sampling.
                Requires client to have sampling.tools capability.
            tool_choice: Optional control over tool usage behavior.
                Requires client to have sampling.tools capability.
            related_request_id: Optional ID of a related request.

        Returns:
            The sampling result from the client.

        Raises:
            MCPError: If tools are provided but client doesn't support them.
            ValueError: If tool_use or tool_result message structure is invalid.
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests.
        """
        client_caps = self.client_params.capabilities if self.client_params else None
        validate_sampling_tools(client_caps, tools, tool_choice)
        validate_tool_use_result_messages(messages)

        request = types.CreateMessageRequest(
            params=types.CreateMessageRequestParams(
                messages=messages,
                system_prompt=system_prompt,
                include_context=include_context,
                temperature=temperature,
                max_tokens=max_tokens,
                stop_sequences=stop_sequences,
                metadata=metadata,
                model_preferences=model_preferences,
                tools=tools,
                tool_choice=tool_choice,
            ),
        )
        metadata_obj = ServerMessageMetadata(related_request_id=related_request_id)

        if tools is not None:
            return await self.send_request(
                request=request,
                result_type=types.CreateMessageResultWithTools,
                metadata=metadata_obj,
            )
        return await self.send_request(
            request=request,
            result_type=types.CreateMessageResult,
            metadata=metadata_obj,
        )

    @deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
    async def list_roots(self) -> types.ListRootsResult:
        """Send a roots/list request.

        Raises:
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests.
        """
        return await self.send_request(
            types.ListRootsRequest(),
            types.ListRootsResult,
        )

    async def elicit(
        self,
        message: str,
        requested_schema: types.ElicitRequestedSchema,
        related_request_id: types.RequestId | None = None,
    ) -> types.ElicitResult:
        """Send a form mode elicitation/create request.

        Args:
            message: The message to present to the user.
            requested_schema: Schema defining the expected response structure.
            related_request_id: Optional ID of the request that triggered this elicitation.

        Returns:
            The client's response.

        Note:
            This method is deprecated in favor of elicit_form(). It remains for
            backward compatibility but new code should use elicit_form().
        """
        return await self.elicit_form(message, requested_schema, related_request_id)

    async def elicit_form(
        self,
        message: str,
        requested_schema: types.ElicitRequestedSchema,
        related_request_id: types.RequestId | None = None,
    ) -> types.ElicitResult:
        """Send a form mode elicitation/create request.

        Args:
            message: The message to present to the user.
            requested_schema: Schema defining the expected response structure.
            related_request_id: Optional ID of the request that triggered this elicitation.

        Returns:
            The client's response with form data.

        Raises:
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests.
        """
        return await self.send_request(
            types.ElicitRequest(
                params=types.ElicitRequestFormParams(
                    message=message,
                    requested_schema=requested_schema,
                ),
            ),
            types.ElicitResult,
            metadata=ServerMessageMetadata(related_request_id=related_request_id),
        )

    async def elicit_url(
        self,
        message: str,
        url: str,
        elicitation_id: str,
        related_request_id: types.RequestId | None = None,
    ) -> types.ElicitResult:
        """Send a URL mode elicitation/create request.

        This directs the user to an external URL for out-of-band interactions
        like OAuth flows, credential collection, or payment processing.

        Args:
            message: Human-readable explanation of why the interaction is needed.
            url: The URL the user should navigate to.
            elicitation_id: Unique identifier for tracking this elicitation.
            related_request_id: Optional ID of the request that triggered this elicitation.

        Returns:
            The client's response indicating acceptance, decline, or cancellation.

        Raises:
            NoBackChannelError: The connection has no back-channel for
                server-initiated requests.
        """
        return await self.send_request(
            types.ElicitRequest(
                params=types.ElicitRequestURLParams(
                    message=message,
                    url=url,
                    elicitation_id=elicitation_id,
                ),
            ),
            types.ElicitResult,
            metadata=ServerMessageMetadata(related_request_id=related_request_id),
        )

    async def send_ping(self) -> types.EmptyResult:
        """Send a ping request."""
        return await self.send_request(
            types.PingRequest(),
            types.EmptyResult,
        )

    async def report_progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
        """Report progress for the inbound request this session is scoped to.

        A no-op when the caller did not request progress. Dispatcher-agnostic:
        on JSON-RPC the held `DispatchContext` emits ``notifications/progress``
        against the caller's token; on the in-process direct dispatcher it
        invokes the caller's callback directly.
        """
        await self._request_outbound.progress(progress, total, message)

    async def send_progress_notification(
        self,
        progress_token: str | int,
        progress: float,
        total: float | None = None,
        message: str | None = None,
        related_request_id: str | None = None,
    ) -> None:
        """Send a progress notification."""
        await self.send_notification(
            types.ProgressNotification(
                params=types.ProgressNotificationParams(
                    progress_token=progress_token,
                    progress=progress,
                    total=total,
                    message=message,
                ),
            ),
            related_request_id,
        )

    async def send_resource_list_changed(self) -> None:
        """Send a resource list changed notification."""
        await self.send_notification(types.ResourceListChangedNotification())

    async def send_tool_list_changed(self) -> None:
        """Send a tool list changed notification."""
        await self.send_notification(types.ToolListChangedNotification())

    async def send_prompt_list_changed(self) -> None:
        """Send a prompt list changed notification."""
        await self.send_notification(types.PromptListChangedNotification())

    async def send_elicit_complete(
        self,
        elicitation_id: str,
        related_request_id: types.RequestId | None = None,
    ) -> None:
        """Send an elicitation completion notification.

        This should be sent when a URL mode elicitation has been completed
        out-of-band to inform the client that it may retry any requests
        that were waiting for this elicitation.

        Args:
            elicitation_id: The unique identifier of the completed elicitation
            related_request_id: Optional ID of the request that triggered this notification
        """
        await self.send_notification(
            types.ElicitCompleteNotification(
                params=types.ElicitCompleteNotificationParams(elicitation_id=elicitation_id)
            ),
            related_request_id,
        )

client_params property

client_params: InitializeRequestParams | None

The client's initialize request params; None when no client info was supplied.

protocol_version property

protocol_version: str

The protocol version this connection speaks.

Populated at Connection construction and overwritten once the handshake commits on the loop path; never None.

send_request async

send_request(
    request: ServerRequest,
    result_type: type[ResultT],
    request_read_timeout_seconds: float | None = None,
    metadata: ServerMessageMetadata | None = None,
    progress_callback: ProgressFnT | None = None,
) -> ResultT

Send a typed server-to-client request and validate the result.

Raises:

Type Description
MCPError

The peer responded with an error.

NoBackChannelError

The connection has no back-channel for server-initiated requests (raised by the held Outbound).

ValidationError

The peer's result does not match result_type.

Source code in src/mcp/server/session.py
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
async def send_request(
    self,
    request: types.ServerRequest,
    result_type: type[ResultT],
    request_read_timeout_seconds: float | None = None,
    metadata: ServerMessageMetadata | None = None,
    progress_callback: ProgressFnT | None = None,
) -> ResultT:
    """Send a typed server-to-client request and validate the result.

    Raises:
        MCPError: The peer responded with an error.
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests (raised by the held `Outbound`).
        pydantic.ValidationError: The peer's result does not match `result_type`.
    """
    related = metadata.related_request_id if metadata is not None else None
    channel = self._request_outbound if related is not None else self._connection.outbound
    data = request.model_dump(by_alias=True, mode="json", exclude_none=True)
    opts: CallOptions = {}
    if request_read_timeout_seconds is not None:
        opts["timeout"] = request_read_timeout_seconds
    if progress_callback is not None:
        opts["on_progress"] = progress_callback
    result = await channel.send_raw_request(data["method"], data.get("params"), opts or None)
    try:
        _methods.validate_client_result(request.method, self.protocol_version, result)
    except KeyError:
        pass
    return result_type.model_validate(result, by_name=False)

send_notification async

send_notification(
    notification: ServerNotification,
    related_request_id: RequestId | None = None,
) -> None

Send a typed server-to-client notification.

Source code in src/mcp/server/session.py
88
89
90
91
92
93
94
95
96
async def send_notification(
    self,
    notification: types.ServerNotification,
    related_request_id: types.RequestId | None = None,
) -> None:
    """Send a typed server-to-client notification."""
    channel = self._request_outbound if related_request_id is not None else self._connection.outbound
    data = notification.model_dump(by_alias=True, mode="json", exclude_none=True)
    await channel.notify(data["method"], data.get("params"))

check_client_capability

check_client_capability(
    capability: ClientCapabilities,
) -> bool

Check if the client supports a specific capability.

Source code in src/mcp/server/session.py
 98
 99
100
def check_client_capability(self, capability: types.ClientCapabilities) -> bool:
    """Check if the client supports a specific capability."""
    return self._connection.check_capability(capability)

send_log_message async

send_log_message(
    level: LoggingLevel,
    data: Any,
    logger: str | None = None,
    related_request_id: RequestId | None = None,
) -> None

Send a log message notification.

Source code in src/mcp/server/session.py
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@deprecated("The logging capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def send_log_message(
    self,
    level: types.LoggingLevel,
    data: Any,
    logger: str | None = None,
    related_request_id: types.RequestId | None = None,
) -> None:
    """Send a log message notification."""
    await self.send_notification(
        types.LoggingMessageNotification(
            params=types.LoggingMessageNotificationParams(
                level=level,
                data=data,
                logger=logger,
            ),
        ),
        related_request_id,
    )

send_resource_updated async

send_resource_updated(uri: str | AnyUrl) -> None

Send a resource updated notification.

Source code in src/mcp/server/session.py
122
123
124
125
126
127
128
async def send_resource_updated(self, uri: str | AnyUrl) -> None:
    """Send a resource updated notification."""
    await self.send_notification(
        types.ResourceUpdatedNotification(
            params=types.ResourceUpdatedNotificationParams(uri=str(uri)),
        )
    )

create_message async

create_message(
    messages: list[SamplingMessage],
    *,
    max_tokens: int,
    system_prompt: str | None = None,
    include_context: IncludeContext | None = None,
    temperature: float | None = None,
    stop_sequences: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    model_preferences: ModelPreferences | None = None,
    tools: None = None,
    tool_choice: ToolChoice | None = None,
    related_request_id: RequestId | None = None
) -> CreateMessageResult
create_message(
    messages: list[SamplingMessage],
    *,
    max_tokens: int,
    system_prompt: str | None = None,
    include_context: IncludeContext | None = None,
    temperature: float | None = None,
    stop_sequences: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    model_preferences: ModelPreferences | None = None,
    tools: list[Tool],
    tool_choice: ToolChoice | None = None,
    related_request_id: RequestId | None = None
) -> CreateMessageResultWithTools
create_message(
    messages: list[SamplingMessage],
    *,
    max_tokens: int,
    system_prompt: str | None = None,
    include_context: IncludeContext | None = None,
    temperature: float | None = None,
    stop_sequences: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    model_preferences: ModelPreferences | None = None,
    tools: list[Tool] | None = None,
    tool_choice: ToolChoice | None = None,
    related_request_id: RequestId | None = None
) -> CreateMessageResult | CreateMessageResultWithTools

Send a sampling/create_message request.

Parameters:

Name Type Description Default
messages list[SamplingMessage]

The conversation messages to send.

required
max_tokens int

Maximum number of tokens to generate.

required
system_prompt str | None

Optional system prompt.

None
include_context IncludeContext | None

Optional context inclusion setting. Should only be set to "thisServer" or "allServers" if the client has sampling.context capability.

None
temperature float | None

Optional sampling temperature.

None
stop_sequences list[str] | None

Optional stop sequences.

None
metadata dict[str, Any] | None

Optional metadata to pass through to the LLM provider.

None
model_preferences ModelPreferences | None

Optional model selection preferences.

None
tools list[Tool] | None

Optional list of tools the LLM can use during sampling. Requires client to have sampling.tools capability.

None
tool_choice ToolChoice | None

Optional control over tool usage behavior. Requires client to have sampling.tools capability.

None
related_request_id RequestId | None

Optional ID of a related request.

None

Returns:

Type Description
CreateMessageResult | CreateMessageResultWithTools

The sampling result from the client.

Raises:

Type Description
MCPError

If tools are provided but client doesn't support them.

ValueError

If tool_use or tool_result message structure is invalid.

NoBackChannelError

The connection has no back-channel for server-initiated requests.

Source code in src/mcp/server/session.py
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
@deprecated("The sampling capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def create_message(
    self,
    messages: list[types.SamplingMessage],
    *,
    max_tokens: int,
    system_prompt: str | None = None,
    include_context: types.IncludeContext | None = None,
    temperature: float | None = None,
    stop_sequences: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    model_preferences: types.ModelPreferences | None = None,
    tools: list[types.Tool] | None = None,
    tool_choice: types.ToolChoice | None = None,
    related_request_id: types.RequestId | None = None,
) -> types.CreateMessageResult | types.CreateMessageResultWithTools:
    """Send a sampling/create_message request.

    Args:
        messages: The conversation messages to send.
        max_tokens: Maximum number of tokens to generate.
        system_prompt: Optional system prompt.
        include_context: Optional context inclusion setting.
            Should only be set to "thisServer" or "allServers"
            if the client has sampling.context capability.
        temperature: Optional sampling temperature.
        stop_sequences: Optional stop sequences.
        metadata: Optional metadata to pass through to the LLM provider.
        model_preferences: Optional model selection preferences.
        tools: Optional list of tools the LLM can use during sampling.
            Requires client to have sampling.tools capability.
        tool_choice: Optional control over tool usage behavior.
            Requires client to have sampling.tools capability.
        related_request_id: Optional ID of a related request.

    Returns:
        The sampling result from the client.

    Raises:
        MCPError: If tools are provided but client doesn't support them.
        ValueError: If tool_use or tool_result message structure is invalid.
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests.
    """
    client_caps = self.client_params.capabilities if self.client_params else None
    validate_sampling_tools(client_caps, tools, tool_choice)
    validate_tool_use_result_messages(messages)

    request = types.CreateMessageRequest(
        params=types.CreateMessageRequestParams(
            messages=messages,
            system_prompt=system_prompt,
            include_context=include_context,
            temperature=temperature,
            max_tokens=max_tokens,
            stop_sequences=stop_sequences,
            metadata=metadata,
            model_preferences=model_preferences,
            tools=tools,
            tool_choice=tool_choice,
        ),
    )
    metadata_obj = ServerMessageMetadata(related_request_id=related_request_id)

    if tools is not None:
        return await self.send_request(
            request=request,
            result_type=types.CreateMessageResultWithTools,
            metadata=metadata_obj,
        )
    return await self.send_request(
        request=request,
        result_type=types.CreateMessageResult,
        metadata=metadata_obj,
    )

list_roots async

list_roots() -> ListRootsResult

Send a roots/list request.

Raises:

Type Description
NoBackChannelError

The connection has no back-channel for server-initiated requests.

Source code in src/mcp/server/session.py
246
247
248
249
250
251
252
253
254
255
256
257
@deprecated("The roots capability is deprecated as of 2026-07-28 (SEP-2577).", category=MCPDeprecationWarning)
async def list_roots(self) -> types.ListRootsResult:
    """Send a roots/list request.

    Raises:
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests.
    """
    return await self.send_request(
        types.ListRootsRequest(),
        types.ListRootsResult,
    )

elicit async

elicit(
    message: str,
    requested_schema: ElicitRequestedSchema,
    related_request_id: RequestId | None = None,
) -> ElicitResult

Send a form mode elicitation/create request.

Parameters:

Name Type Description Default
message str

The message to present to the user.

required
requested_schema ElicitRequestedSchema

Schema defining the expected response structure.

required
related_request_id RequestId | None

Optional ID of the request that triggered this elicitation.

None

Returns:

Type Description
ElicitResult

The client's response.

Note

This method is deprecated in favor of elicit_form(). It remains for backward compatibility but new code should use elicit_form().

Source code in src/mcp/server/session.py
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
async def elicit(
    self,
    message: str,
    requested_schema: types.ElicitRequestedSchema,
    related_request_id: types.RequestId | None = None,
) -> types.ElicitResult:
    """Send a form mode elicitation/create request.

    Args:
        message: The message to present to the user.
        requested_schema: Schema defining the expected response structure.
        related_request_id: Optional ID of the request that triggered this elicitation.

    Returns:
        The client's response.

    Note:
        This method is deprecated in favor of elicit_form(). It remains for
        backward compatibility but new code should use elicit_form().
    """
    return await self.elicit_form(message, requested_schema, related_request_id)

elicit_form async

elicit_form(
    message: str,
    requested_schema: ElicitRequestedSchema,
    related_request_id: RequestId | None = None,
) -> ElicitResult

Send a form mode elicitation/create request.

Parameters:

Name Type Description Default
message str

The message to present to the user.

required
requested_schema ElicitRequestedSchema

Schema defining the expected response structure.

required
related_request_id RequestId | None

Optional ID of the request that triggered this elicitation.

None

Returns:

Type Description
ElicitResult

The client's response with form data.

Raises:

Type Description
NoBackChannelError

The connection has no back-channel for server-initiated requests.

Source code in src/mcp/server/session.py
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
async def elicit_form(
    self,
    message: str,
    requested_schema: types.ElicitRequestedSchema,
    related_request_id: types.RequestId | None = None,
) -> types.ElicitResult:
    """Send a form mode elicitation/create request.

    Args:
        message: The message to present to the user.
        requested_schema: Schema defining the expected response structure.
        related_request_id: Optional ID of the request that triggered this elicitation.

    Returns:
        The client's response with form data.

    Raises:
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests.
    """
    return await self.send_request(
        types.ElicitRequest(
            params=types.ElicitRequestFormParams(
                message=message,
                requested_schema=requested_schema,
            ),
        ),
        types.ElicitResult,
        metadata=ServerMessageMetadata(related_request_id=related_request_id),
    )

elicit_url async

elicit_url(
    message: str,
    url: str,
    elicitation_id: str,
    related_request_id: RequestId | None = None,
) -> ElicitResult

Send a URL mode elicitation/create request.

This directs the user to an external URL for out-of-band interactions like OAuth flows, credential collection, or payment processing.

Parameters:

Name Type Description Default
message str

Human-readable explanation of why the interaction is needed.

required
url str

The URL the user should navigate to.

required
elicitation_id str

Unique identifier for tracking this elicitation.

required
related_request_id RequestId | None

Optional ID of the request that triggered this elicitation.

None

Returns:

Type Description
ElicitResult

The client's response indicating acceptance, decline, or cancellation.

Raises:

Type Description
NoBackChannelError

The connection has no back-channel for server-initiated requests.

Source code in src/mcp/server/session.py
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
async def elicit_url(
    self,
    message: str,
    url: str,
    elicitation_id: str,
    related_request_id: types.RequestId | None = None,
) -> types.ElicitResult:
    """Send a URL mode elicitation/create request.

    This directs the user to an external URL for out-of-band interactions
    like OAuth flows, credential collection, or payment processing.

    Args:
        message: Human-readable explanation of why the interaction is needed.
        url: The URL the user should navigate to.
        elicitation_id: Unique identifier for tracking this elicitation.
        related_request_id: Optional ID of the request that triggered this elicitation.

    Returns:
        The client's response indicating acceptance, decline, or cancellation.

    Raises:
        NoBackChannelError: The connection has no back-channel for
            server-initiated requests.
    """
    return await self.send_request(
        types.ElicitRequest(
            params=types.ElicitRequestURLParams(
                message=message,
                url=url,
                elicitation_id=elicitation_id,
            ),
        ),
        types.ElicitResult,
        metadata=ServerMessageMetadata(related_request_id=related_request_id),
    )

send_ping async

send_ping() -> EmptyResult

Send a ping request.

Source code in src/mcp/server/session.py
349
350
351
352
353
354
async def send_ping(self) -> types.EmptyResult:
    """Send a ping request."""
    return await self.send_request(
        types.PingRequest(),
        types.EmptyResult,
    )

report_progress async

report_progress(
    progress: float,
    total: float | None = None,
    message: str | None = None,
) -> None

Report progress for the inbound request this session is scoped to.

A no-op when the caller did not request progress. Dispatcher-agnostic: on JSON-RPC the held DispatchContext emits notifications/progress against the caller's token; on the in-process direct dispatcher it invokes the caller's callback directly.

Source code in src/mcp/server/session.py
356
357
358
359
360
361
362
363
364
async def report_progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
    """Report progress for the inbound request this session is scoped to.

    A no-op when the caller did not request progress. Dispatcher-agnostic:
    on JSON-RPC the held `DispatchContext` emits ``notifications/progress``
    against the caller's token; on the in-process direct dispatcher it
    invokes the caller's callback directly.
    """
    await self._request_outbound.progress(progress, total, message)

send_progress_notification async

send_progress_notification(
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
    related_request_id: str | None = None,
) -> None

Send a progress notification.

Source code in src/mcp/server/session.py
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
async def send_progress_notification(
    self,
    progress_token: str | int,
    progress: float,
    total: float | None = None,
    message: str | None = None,
    related_request_id: str | None = None,
) -> None:
    """Send a progress notification."""
    await self.send_notification(
        types.ProgressNotification(
            params=types.ProgressNotificationParams(
                progress_token=progress_token,
                progress=progress,
                total=total,
                message=message,
            ),
        ),
        related_request_id,
    )

send_resource_list_changed async

send_resource_list_changed() -> None

Send a resource list changed notification.

Source code in src/mcp/server/session.py
387
388
389
async def send_resource_list_changed(self) -> None:
    """Send a resource list changed notification."""
    await self.send_notification(types.ResourceListChangedNotification())

send_tool_list_changed async

send_tool_list_changed() -> None

Send a tool list changed notification.

Source code in src/mcp/server/session.py
391
392
393
async def send_tool_list_changed(self) -> None:
    """Send a tool list changed notification."""
    await self.send_notification(types.ToolListChangedNotification())

send_prompt_list_changed async

send_prompt_list_changed() -> None

Send a prompt list changed notification.

Source code in src/mcp/server/session.py
395
396
397
async def send_prompt_list_changed(self) -> None:
    """Send a prompt list changed notification."""
    await self.send_notification(types.PromptListChangedNotification())

send_elicit_complete async

send_elicit_complete(
    elicitation_id: str,
    related_request_id: RequestId | None = None,
) -> None

Send an elicitation completion notification.

This should be sent when a URL mode elicitation has been completed out-of-band to inform the client that it may retry any requests that were waiting for this elicitation.

Parameters:

Name Type Description Default
elicitation_id str

The unique identifier of the completed elicitation

required
related_request_id RequestId | None

Optional ID of the request that triggered this notification

None
Source code in src/mcp/server/session.py
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
async def send_elicit_complete(
    self,
    elicitation_id: str,
    related_request_id: types.RequestId | None = None,
) -> None:
    """Send an elicitation completion notification.

    This should be sent when a URL mode elicitation has been completed
    out-of-band to inform the client that it may retry any requests
    that were waiting for this elicitation.

    Args:
        elicitation_id: The unique identifier of the completed elicitation
        related_request_id: Optional ID of the request that triggered this notification
    """
    await self.send_notification(
        types.ElicitCompleteNotification(
            params=types.ElicitCompleteNotificationParams(elicitation_id=elicitation_id)
        ),
        related_request_id,
    )