Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 29 additions & 1 deletion src/mcp/client/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,12 @@ def __init__(
async def initialize(self) -> types.InitializeResult:
sampling = types.SamplingCapability() if self._sampling_callback is not _default_sampling_callback else None
elicitation = (
types.ElicitationCapability() if self._elicitation_callback is not _default_elicitation_callback else None
types.ElicitationCapability(
form=types.FormElicitationCapability(),
url=types.UrlElicitationCapability(),
)
if self._elicitation_callback is not _default_elicitation_callback
else None
)
roots = (
# TODO: Should this be based on whether we
Expand Down Expand Up @@ -501,6 +506,29 @@ async def list_tools(

return result

async def track_elicitation(
self,
elicitation_id: str,
progress_token: types.ProgressToken | None = None,
) -> types.ElicitTrackResult:
"""Send an elicitation/track request to monitor URL mode elicitation progress.

Args:
elicitation_id: The unique identifier of the elicitation to track
progress_token: Optional token for receiving progress notifications

Returns:
ElicitTrackResult indicating the status of the elicitation
"""
params = types.ElicitTrackRequestParams(elicitationId=elicitation_id)
if progress_token is not None:
params.meta = types.RequestParams.Meta(progressToken=progress_token)

return await self.send_request(
types.ClientRequest(types.ElicitTrackRequest(params=params)),
types.ElicitTrackResult,
)

async def send_roots_list_changed(self) -> None:
"""Send a roots/list_changed notification."""
await self.send_notification(types.ClientNotification(types.RootsListChangedNotification()))
Expand Down
63 changes: 61 additions & 2 deletions src/mcp/server/elicitation.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,15 @@ class CancelledElicitation(BaseModel):
ElicitationResult = AcceptedElicitation[ElicitSchemaModelT] | DeclinedElicitation | CancelledElicitation


class AcceptedUrlElicitation(BaseModel):
"""Result when user accepts a URL mode elicitation."""

action: Literal["accept"] = "accept"


UrlElicitationResult = AcceptedUrlElicitation | DeclinedElicitation | CancelledElicitation


# Primitive types allowed in elicitation schemas
_ELICITATION_PRIMITIVE_TYPES = (str, int, float, bool)

Expand Down Expand Up @@ -79,20 +88,22 @@ async def elicit_with_validation(
schema: type[ElicitSchemaModelT],
related_request_id: RequestId | None = None,
) -> ElicitationResult[ElicitSchemaModelT]:
"""Elicit information from the client/user with schema validation.
"""Elicit information from the client/user with schema validation (form mode).

This method can be used to interactively ask for additional information from the
client within a tool's execution. The client might display the message to the
user and collect a response according to the provided schema. Or in case a
client is an agent, it might decide how to handle the elicitation -- either by asking
the user or automatically generating a response.

For sensitive data like credentials or OAuth flows, use elicit_url() instead.
"""
# Validate that schema only contains primitive types and fail loudly if not
_validate_elicitation_schema(schema)

json_schema = schema.model_json_schema()

result = await session.elicit(
result = await session.elicit_form(
message=message,
requestedSchema=json_schema,
related_request_id=related_request_id,
Expand All @@ -109,3 +120,51 @@ async def elicit_with_validation(
else:
# This should never happen, but handle it just in case
raise ValueError(f"Unexpected elicitation action: {result.action}")


async def elicit_url(
session: ServerSession,
message: str,
url: str,
elicitation_id: str,
related_request_id: RequestId | None = None,
) -> UrlElicitationResult:
"""Elicit information from the user via out-of-band URL navigation (URL mode).

This method directs the user to an external URL where sensitive interactions can
occur without passing data through the MCP client. Use this for:
- Collecting sensitive credentials (API keys, passwords)
- OAuth authorization flows with third-party services
- Payment and subscription flows
- Any interaction where data should not pass through the LLM context

The response indicates whether the user consented to navigate to the URL.
The actual interaction happens out-of-band, and you can track progress using
session.track_elicitation().

Args:
session: The server session
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:
UrlElicitationResult indicating accept, decline, or cancel
"""
result = await session.elicit_url(
message=message,
url=url,
elicitation_id=elicitation_id,
related_request_id=related_request_id,
)

if result.action == "accept":
return AcceptedUrlElicitation()
elif result.action == "decline":
return DeclinedElicitation()
elif result.action == "cancel":
return CancelledElicitation()
else:
# This should never happen, but handle it just in case
raise ValueError(f"Unexpected elicitation action: {result.action}")
62 changes: 61 additions & 1 deletion src/mcp/server/session.py
Original file line number Diff line number Diff line change
Expand Up @@ -260,19 +260,43 @@ async def elicit(
requestedSchema: types.ElicitRequestedSchema,
related_request_id: types.RequestId | None = None,
) -> types.ElicitResult:
"""Send an elicitation/create request.
"""Send a form mode elicitation/create request.

Args:
message: The message to present to the user
requestedSchema: 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, requestedSchema, related_request_id)

async def elicit_form(
self,
message: str,
requestedSchema: 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
requestedSchema: 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
"""
return await self.send_request(
types.ServerRequest(
types.ElicitRequest(
params=types.ElicitRequestParams(
mode="form",
message=message,
requestedSchema=requestedSchema,
),
Expand All @@ -282,6 +306,42 @@ async def elicit(
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
"""
return await self.send_request(
types.ServerRequest(
types.ElicitRequest(
params=types.ElicitRequestParams(
mode="url",
message=message,
url=url,
elicitationId=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(
Expand Down
Loading