Add UrlElicitationRequiredError and ctx.elicit_url() for URL mode elicitation

felixweinberger · felixweinberger · commit 3f0ac89df3f7 · 2025-11-24T13:34:26.000Z
This adds the missing pieces for feature parity with the TypeScript SDK:

- UrlElicitationRequiredError exception class that can be raised from tool
  handlers to signal that URL elicitation(s) are required before proceeding.
  The error carries a list of ElicitRequestURLParams and serializes to
  JSON-RPC error code -32042.

- ctx.elicit_url() method on FastMCP Context for ergonomic URL mode
  elicitation, matching the existing ctx.elicit() for form mode.

- Updated elicitation.py example showing both URL mode patterns:
  - Using ctx.elicit_url() for direct elicitation
  - Raising UrlElicitationRequiredError for the "throw error" pattern

- Comprehensive tests for the new exception class.
diff --git a/examples/snippets/servers/elicitation.py b/examples/snippets/servers/elicitation.py
@@ -1,7 +1,18 @@
+"""Elicitation examples demonstrating form and URL mode elicitation.
+
+Form mode elicitation collects structured, non-sensitive data through a schema.
+URL mode elicitation directs users to external URLs for sensitive operations
+like OAuth flows, credential collection, or payment processing.
+"""
+
+import uuid
+
 from pydantic import BaseModel, Field
 
 from mcp.server.fastmcp import Context, FastMCP
 from mcp.server.session import ServerSession
+from mcp.shared.exceptions import UrlElicitationRequiredError
+from mcp.types import ElicitRequestURLParams
 
 mcp = FastMCP(name="Elicitation Example")
 
@@ -18,7 +29,10 @@ class BookingPreferences(BaseModel):
 
 @mcp.tool()
 async def book_table(date: str, time: str, party_size: int, ctx: Context[ServerSession, None]) -> str:
-    """Book a table with date availability check."""
+    """Book a table with date availability check.
+
+    This demonstrates form mode elicitation for collecting non-sensitive user input.
+    """
     # Check if date is available
     if date == "2024-12-25":
         # Date unavailable - ask user for alternative
@@ -35,3 +49,51 @@ async def book_table(date: str, time: str, party_size: int, ctx: Context[ServerS
 
     # Date available
     return f"[SUCCESS] Booked for {date} at {time}"
+
+
+@mcp.tool()
+async def secure_payment(amount: float, ctx: Context[ServerSession, None]) -> str:
+    """Process a secure payment requiring URL confirmation.
+
+    This demonstrates URL mode elicitation using ctx.elicit_url() for
+    operations that require out-of-band user interaction.
+    """
+    elicitation_id = str(uuid.uuid4())
+
+    result = await ctx.elicit_url(
+        message=f"Please confirm payment of ${amount:.2f}",
+        url=f"https://payments.example.com/confirm?amount={amount}&id={elicitation_id}",
+        elicitation_id=elicitation_id,
+    )
+
+    if result.action == "accept":
+        # In a real app, the payment confirmation would happen out-of-band
+        # and you'd verify the payment status from your backend
+        return f"Payment of ${amount:.2f} initiated - check your browser to complete"
+    elif result.action == "decline":
+        return "Payment declined by user"
+    return "Payment cancelled"
+
+
+@mcp.tool()
+async def connect_service(service_name: str, ctx: Context[ServerSession, None]) -> str:
+    """Connect to a third-party service requiring OAuth authorization.
+
+    This demonstrates the "throw error" pattern using UrlElicitationRequiredError.
+    Use this pattern when the tool cannot proceed without user authorization.
+    """
+    elicitation_id = str(uuid.uuid4())
+
+    # Raise UrlElicitationRequiredError to signal that the client must complete
+    # a URL elicitation before this request can be processed.
+    # The MCP framework will convert this to a -32042 error response.
+    raise UrlElicitationRequiredError(
+        [
+            ElicitRequestURLParams(
+                mode="url",
+                message=f"Authorization required to connect to {service_name}",
+                url=f"https://{service_name}.example.com/oauth/authorize?elicit={elicitation_id}",
+                elicitationId=elicitation_id,
+            )
+        ]
+    )
diff --git a/src/mcp/__init__.py b/src/mcp/__init__.py
@@ -3,7 +3,7 @@
 from .client.stdio import StdioServerParameters, stdio_client
 from .server.session import ServerSession
 from .server.stdio import stdio_server
-from .shared.exceptions import McpError
+from .shared.exceptions import McpError, UrlElicitationRequiredError
 from .types import (
     CallToolRequest,
     ClientCapabilities,
@@ -125,6 +125,7 @@
     "ToolsCapability",
     "ToolUseContent",
     "UnsubscribeRequest",
+    "UrlElicitationRequiredError",
     "stdio_client",
     "stdio_server",
 ]
diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py
@@ -42,8 +42,12 @@
 from mcp.server.elicitation import (
     ElicitationResult,
     ElicitSchemaModelT,
+    UrlElicitationResult,
     elicit_with_validation,
 )
+from mcp.server.elicitation import (
+    elicit_url as _elicit_url,
+)
 from mcp.server.fastmcp.exceptions import ResourceError
 from mcp.server.fastmcp.prompts import Prompt, PromptManager
 from mcp.server.fastmcp.resources import FunctionResource, Resource, ResourceManager
@@ -1200,6 +1204,41 @@ async def elicit(
             related_request_id=self.request_id,
         )
 
+    async def elicit_url(
+        self,
+        message: str,
+        url: str,
+        elicitation_id: str,
+    ) -> UrlElicitationResult:
+        """Request URL mode elicitation from the client.
+
+        This directs the user to an external URL for out-of-band interactions
+        that must not pass 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. When the elicitation completes,
+        call `self.session.send_elicit_complete(elicitation_id)` to notify the client.
+
+        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
+
+        Returns:
+            UrlElicitationResult indicating accept, decline, or cancel
+        """
+        return await _elicit_url(
+            session=self.request_context.session,
+            message=message,
+            url=url,
+            elicitation_id=elicitation_id,
+            related_request_id=self.request_id,
+        )
+
     async def log(
         self,
         level: Literal["debug", "info", "warning", "error"],
diff --git a/src/mcp/shared/exceptions.py b/src/mcp/shared/exceptions.py
@@ -1,4 +1,8 @@
-from mcp.types import ErrorData
+from __future__ import annotations
+
+from typing import Any, cast
+
+from mcp.types import URL_ELICITATION_REQUIRED, ElicitRequestURLParams, ErrorData
 
 
 class McpError(Exception):
@@ -12,3 +16,56 @@ def __init__(self, error: ErrorData):
         """Initialize McpError."""
         super().__init__(error.message)
         self.error = error
+
+
+class UrlElicitationRequiredError(McpError):
+    """
+    Specialized error for when a tool requires URL mode elicitation(s) before proceeding.
+
+    Servers can raise this error from tool handlers to indicate that the client
+    must complete one or more URL elicitations before the request can be processed.
+
+    Example:
+        raise UrlElicitationRequiredError([
+            ElicitRequestURLParams(
+                mode="url",
+                message="Authorization required for your files",
+                url="https://example.com/oauth/authorize",
+                elicitationId="auth-001"
+            )
+        ])
+    """
+
+    def __init__(
+        self,
+        elicitations: list[ElicitRequestURLParams],
+        message: str | None = None,
+    ):
+        """Initialize UrlElicitationRequiredError."""
+        if message is None:
+            message = f"URL elicitation{'s' if len(elicitations) > 1 else ''} required"
+
+        self._elicitations = elicitations
+
+        error = ErrorData(
+            code=URL_ELICITATION_REQUIRED,
+            message=message,
+            data={"elicitations": [e.model_dump(by_alias=True, exclude_none=True) for e in elicitations]},
+        )
+        super().__init__(error)
+
+    @property
+    def elicitations(self) -> list[ElicitRequestURLParams]:
+        """The list of URL elicitations required before the request can proceed."""
+        return self._elicitations
+
+    @classmethod
+    def from_error(cls, error: ErrorData) -> UrlElicitationRequiredError:
+        """Reconstruct from an ErrorData received over the wire."""
+        if error.code != URL_ELICITATION_REQUIRED:
+            raise ValueError(f"Expected error code {URL_ELICITATION_REQUIRED}, got {error.code}")
+
+        data = cast(dict[str, Any], error.data or {})
+        raw_elicitations = cast(list[dict[str, Any]], data.get("elicitations", []))
+        elicitations = [ElicitRequestURLParams.model_validate(e) for e in raw_elicitations]
+        return cls(elicitations, error.message)
diff --git a/tests/shared/test_exceptions.py b/tests/shared/test_exceptions.py
@@ -0,0 +1,159 @@
+"""Tests for MCP exception classes."""
+
+import pytest
+
+from mcp.shared.exceptions import McpError, UrlElicitationRequiredError
+from mcp.types import URL_ELICITATION_REQUIRED, ElicitRequestURLParams, ErrorData
+
+
+class TestUrlElicitationRequiredError:
+    """Tests for UrlElicitationRequiredError exception class."""
+
+    def test_create_with_single_elicitation(self) -> None:
+        """Test creating error with a single elicitation."""
+        elicitation = ElicitRequestURLParams(
+            mode="url",
+            message="Auth required",
+            url="https://example.com/auth",
+            elicitationId="test-123",
+        )
+        error = UrlElicitationRequiredError([elicitation])
+
+        assert error.error.code == URL_ELICITATION_REQUIRED
+        assert error.error.message == "URL elicitation required"
+        assert len(error.elicitations) == 1
+        assert error.elicitations[0].elicitationId == "test-123"
+
+    def test_create_with_multiple_elicitations(self) -> None:
+        """Test creating error with multiple elicitations uses plural message."""
+        elicitations = [
+            ElicitRequestURLParams(
+                mode="url",
+                message="Auth 1",
+                url="https://example.com/auth1",
+                elicitationId="test-1",
+            ),
+            ElicitRequestURLParams(
+                mode="url",
+                message="Auth 2",
+                url="https://example.com/auth2",
+                elicitationId="test-2",
+            ),
+        ]
+        error = UrlElicitationRequiredError(elicitations)
+
+        assert error.error.message == "URL elicitations required"  # Plural
+        assert len(error.elicitations) == 2
+
+    def test_custom_message(self) -> None:
+        """Test creating error with a custom message."""
+        elicitation = ElicitRequestURLParams(
+            mode="url",
+            message="Auth required",
+            url="https://example.com/auth",
+            elicitationId="test-123",
+        )
+        error = UrlElicitationRequiredError([elicitation], message="Custom message")
+
+        assert error.error.message == "Custom message"
+
+    def test_from_error_data(self) -> None:
+        """Test reconstructing error from ErrorData."""
+        error_data = ErrorData(
+            code=URL_ELICITATION_REQUIRED,
+            message="URL elicitation required",
+            data={
+                "elicitations": [
+                    {
+                        "mode": "url",
+                        "message": "Auth required",
+                        "url": "https://example.com/auth",
+                        "elicitationId": "test-123",
+                    }
+                ]
+            },
+        )
+
+        error = UrlElicitationRequiredError.from_error(error_data)
+
+        assert len(error.elicitations) == 1
+        assert error.elicitations[0].elicitationId == "test-123"
+        assert error.elicitations[0].url == "https://example.com/auth"
+
+    def test_from_error_data_wrong_code(self) -> None:
+        """Test that from_error raises ValueError for wrong error code."""
+        error_data = ErrorData(
+            code=-32600,  # Wrong code
+            message="Some other error",
+            data={},
+        )
+
+        with pytest.raises(ValueError, match="Expected error code"):
+            UrlElicitationRequiredError.from_error(error_data)
+
+    def test_serialization_roundtrip(self) -> None:
+        """Test that error can be serialized and reconstructed."""
+        original = UrlElicitationRequiredError(
+            [
+                ElicitRequestURLParams(
+                    mode="url",
+                    message="Auth required",
+                    url="https://example.com/auth",
+                    elicitationId="test-123",
+                )
+            ]
+        )
+
+        # Simulate serialization over wire
+        error_data = original.error
+
+        # Reconstruct
+        reconstructed = UrlElicitationRequiredError.from_error(error_data)
+
+        assert reconstructed.elicitations[0].elicitationId == original.elicitations[0].elicitationId
+        assert reconstructed.elicitations[0].url == original.elicitations[0].url
+        assert reconstructed.elicitations[0].message == original.elicitations[0].message
+
+    def test_error_data_contains_elicitations(self) -> None:
+        """Test that error data contains properly serialized elicitations."""
+        elicitation = ElicitRequestURLParams(
+            mode="url",
+            message="Please authenticate",
+            url="https://example.com/oauth",
+            elicitationId="oauth-flow-1",
+        )
+        error = UrlElicitationRequiredError([elicitation])
+
+        assert error.error.data is not None
+        assert "elicitations" in error.error.data
+        elicit_data = error.error.data["elicitations"][0]
+        assert elicit_data["mode"] == "url"
+        assert elicit_data["message"] == "Please authenticate"
+        assert elicit_data["url"] == "https://example.com/oauth"
+        assert elicit_data["elicitationId"] == "oauth-flow-1"
+
+    def test_inherits_from_mcp_error(self) -> None:
+        """Test that UrlElicitationRequiredError inherits from McpError."""
+        elicitation = ElicitRequestURLParams(
+            mode="url",
+            message="Auth required",
+            url="https://example.com/auth",
+            elicitationId="test-123",
+        )
+        error = UrlElicitationRequiredError([elicitation])
+
+        assert isinstance(error, McpError)
+        assert isinstance(error, Exception)
+
+    def test_exception_message(self) -> None:
+        """Test that exception message is set correctly."""
+        elicitation = ElicitRequestURLParams(
+            mode="url",
+            message="Auth required",
+            url="https://example.com/auth",
+            elicitationId="test-123",
+        )
+        error = UrlElicitationRequiredError([elicitation])
+
+        # The exception's string representation should match the message
+        assert str(error) == "URL elicitation required"
