Implement SEP-1036: URL mode elicitation for secure out-of-band interactions

src/mcp/client/session.py

@@ -138,7 +138,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
@@ -504,6 +509,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)  # pragma: no cover
+        if progress_token is not None:  # pragma: no cover
+            params.meta = types.RequestParams.Meta(progressToken=progress_token)
+
+        return await self.send_request(  # pragma: no cover
+            types.ClientRequest(types.ElicitTrackRequest(params=params)),
+            types.ElicitTrackResult,
+        )
+
     async def send_roots_list_changed(self) -> None:  # pragma: no cover
         """Send a roots/list_changed notification."""
         await self.send_notification(types.ClientNotification(types.RootsListChangedNotification()))
@@ -552,5 +580,10 @@ async def _received_notification(self, notification: types.ServerNotification) -
         match notification.root:
             case types.LoggingMessageNotification(params=params):
                 await self._logging_callback(params)
+            case types.ElicitCompleteNotification(params=params):
+                # Handle elicitation completion notification
+                # Clients MAY use this to retry requests or update UI
+                # The notification contains the elicitationId of the completed elicitation
+                pass
             case _:
                 pass

src/mcp/server/elicitation.py

@@ -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)
 
@@ -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,
@@ -109,3 +120,51 @@ async def elicit_with_validation(
     else:  # pragma: no cover
         # 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:  # pragma: no cover
+        # This should never happen, but handle it just in case
+        raise ValueError(f"Unexpected elicitation action: {result.action}")

src/mcp/server/session.py

@@ -335,19 +335,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,
                     ),
@@ -357,6 +381,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:  # pragma: no cover
         """Send a ping request."""
         return await self.send_request(
@@ -399,6 +459,30 @@ async def send_prompt_list_changed(self) -> None:  # pragma: no cover
         """Send a prompt list changed notification."""
         await self.send_notification(types.ServerNotification(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
+        """
+        await self.send_notification(
+            types.ServerNotification(
+                types.ElicitCompleteNotification(
+                    params=types.ElicitCompleteNotificationParams(elicitationId=elicitation_id)
+                )
+            ),
+            related_request_id,
+        )
+
     async def _handle_incoming(self, req: ServerRequestResponder) -> None:
         await self._incoming_message_stream_writer.send(req)