diff --git a/README.md b/README.md index 1abbb742b9..ca0655f579 100644 --- a/README.md +++ b/README.md @@ -808,10 +808,21 @@ Request additional information from users. This example shows an Elicitation dur ```python +"""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") @@ -828,7 +839,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 @@ -845,6 +859,54 @@ 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, + ) + ] + ) ``` _Full example: [examples/snippets/servers/elicitation.py](https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/elicitation.py)_ diff --git a/examples/snippets/clients/url_elicitation_client.py b/examples/snippets/clients/url_elicitation_client.py new file mode 100644 index 0000000000..56457512c6 --- /dev/null +++ b/examples/snippets/clients/url_elicitation_client.py @@ -0,0 +1,318 @@ +"""URL Elicitation Client Example. + +Demonstrates how clients handle URL elicitation requests from servers. +This is the Python equivalent of TypeScript SDK's elicitationUrlExample.ts, +focused on URL elicitation patterns without OAuth complexity. + +Features demonstrated: +1. Client elicitation capability declaration +2. Handling elicitation requests from servers via callback +3. Catching UrlElicitationRequiredError from tool calls +4. Browser interaction with security warnings +5. Interactive CLI for testing + +Run with: + cd examples/snippets + uv run elicitation-client + +Requires a server with URL elicitation tools running. Start the elicitation +server first: + uv run server elicitation sse +""" + +from __future__ import annotations + +import asyncio +import json +import subprocess +import sys +import webbrowser +from typing import Any +from urllib.parse import urlparse + +from mcp import ClientSession, types +from mcp.client.sse import sse_client +from mcp.shared.context import RequestContext +from mcp.shared.exceptions import McpError, UrlElicitationRequiredError +from mcp.types import URL_ELICITATION_REQUIRED + + +async def handle_elicitation( + context: RequestContext[ClientSession, Any], + params: types.ElicitRequestParams, +) -> types.ElicitResult | types.ErrorData: + """Handle elicitation requests from the server. + + This callback is invoked when the server sends an elicitation/request. + For URL mode, we prompt the user and optionally open their browser. + """ + if params.mode == "url": + return await handle_url_elicitation(params) + else: + # We only support URL mode in this example + return types.ErrorData( + code=types.INVALID_REQUEST, + message=f"Unsupported elicitation mode: {params.mode}", + ) + + +async def handle_url_elicitation( + params: types.ElicitRequestParams, +) -> types.ElicitResult: + """Handle URL mode elicitation - show security warning and optionally open browser. + + This function demonstrates the security-conscious approach to URL elicitation: + 1. Display the full URL and domain for user inspection + 2. Show the server's reason for requesting this interaction + 3. Require explicit user consent before opening any URL + """ + # Extract URL parameters - these are available on URL mode requests + url = getattr(params, "url", None) + elicitation_id = getattr(params, "elicitationId", None) + message = params.message + + if not url: + print("Error: No URL provided in elicitation request") + return types.ElicitResult(action="cancel") + + # Extract domain for security display + domain = extract_domain(url) + + # Security warning - always show the user what they're being asked to do + print("\n" + "=" * 60) + print("SECURITY WARNING: External URL Request") + print("=" * 60) + print("\nThe server is requesting you to open an external URL.") + print(f"\n Domain: {domain}") + print(f" Full URL: {url}") + print("\n Server's reason:") + print(f" {message}") + print(f"\n Elicitation ID: {elicitation_id}") + print("\n" + "-" * 60) + + # Get explicit user consent + try: + response = input("\nOpen this URL in your browser? (y/n): ").strip().lower() + except EOFError: + return types.ElicitResult(action="cancel") + + if response in ("n", "no"): + print("URL navigation declined.") + return types.ElicitResult(action="decline") + elif response not in ("y", "yes"): + print("Invalid response. Cancelling.") + return types.ElicitResult(action="cancel") + + # Open the browser + print(f"\nOpening browser to: {url}") + open_browser(url) + + print("Waiting for you to complete the interaction in your browser...") + print("(The server will continue once you've finished)") + + return types.ElicitResult(action="accept") + + +def extract_domain(url: str) -> str: + """Extract domain from URL for security display.""" + try: + return urlparse(url).netloc + except Exception: + return "unknown" + + +def open_browser(url: str) -> None: + """Open URL in the default browser.""" + try: + if sys.platform == "darwin": + subprocess.run(["open", url], check=False) + elif sys.platform == "win32": + subprocess.run(["start", url], shell=True, check=False) + else: + webbrowser.open(url) + except Exception as e: + print(f"Failed to open browser: {e}") + print(f"Please manually open: {url}") + + +async def call_tool_with_error_handling( + session: ClientSession, + tool_name: str, + arguments: dict[str, Any], +) -> types.CallToolResult | None: + """Call a tool, handling UrlElicitationRequiredError if raised. + + When a server tool needs URL elicitation before it can proceed, + it can either: + 1. Send an elicitation request directly (handled by elicitation_callback) + 2. Return an error with code -32042 (URL_ELICITATION_REQUIRED) + + This function demonstrates handling case 2 - catching the error + and processing the required URL elicitations. + """ + try: + result = await session.call_tool(tool_name, arguments) + + # Check if the tool returned an error in the result + if result.isError: + print(f"Tool returned error: {result.content}") + return None + + return result + + except McpError as e: + # Check if this is a URL elicitation required error + if e.error.code == URL_ELICITATION_REQUIRED: + print("\n[Tool requires URL elicitation to proceed]") + + # Convert to typed error to access elicitations + url_error = UrlElicitationRequiredError.from_error(e.error) + + # Process each required elicitation + for elicitation in url_error.elicitations: + await handle_url_elicitation(elicitation) + + return None + else: + # Re-raise other MCP errors + print(f"MCP Error: {e.error.message} (code: {e.error.code})") + return None + + +def print_help() -> None: + """Print available commands.""" + print("\nAvailable commands:") + print(" list-tools - List available tools") + print(" call [json-args] - Call a tool with optional JSON arguments") + print(" secure-payment - Test URL elicitation via ctx.elicit_url()") + print(" connect-service - Test URL elicitation via UrlElicitationRequiredError") + print(" help - Show this help") + print(" quit - Exit the program") + + +def print_tool_result(result: types.CallToolResult | None) -> None: + """Print a tool call result.""" + if not result: + return + print("\nTool result:") + for content in result.content: + if isinstance(content, types.TextContent): + print(f" {content.text}") + else: + print(f" [{content.type}]") + + +async def handle_list_tools(session: ClientSession) -> None: + """Handle the list-tools command.""" + tools = await session.list_tools() + if tools.tools: + print("\nAvailable tools:") + for tool in tools.tools: + print(f" - {tool.name}: {tool.description or 'No description'}") + else: + print("No tools available") + + +async def handle_call_command(session: ClientSession, command: str) -> None: + """Handle the call command.""" + parts = command.split(maxsplit=2) + if len(parts) < 2: + print("Usage: call [json-args]") + return + + tool_name = parts[1] + args: dict[str, Any] = {} + if len(parts) > 2: + try: + args = json.loads(parts[2]) + except json.JSONDecodeError as e: + print(f"Invalid JSON arguments: {e}") + return + + print(f"\nCalling tool '{tool_name}' with args: {args}") + result = await call_tool_with_error_handling(session, tool_name, args) + print_tool_result(result) + + +async def process_command(session: ClientSession, command: str) -> bool: + """Process a single command. Returns False if should exit.""" + if command in {"quit", "exit"}: + print("Goodbye!") + return False + + if command == "help": + print_help() + elif command == "list-tools": + await handle_list_tools(session) + elif command.startswith("call "): + await handle_call_command(session, command) + elif command == "secure-payment": + print("\nTesting secure_payment tool (uses ctx.elicit_url())...") + result = await call_tool_with_error_handling(session, "secure_payment", {"amount": 99.99}) + print_tool_result(result) + elif command == "connect-service": + print("\nTesting connect_service tool (raises UrlElicitationRequiredError)...") + result = await call_tool_with_error_handling(session, "connect_service", {"service_name": "github"}) + print_tool_result(result) + else: + print(f"Unknown command: {command}") + print("Type 'help' for available commands.") + + return True + + +async def run_command_loop(session: ClientSession) -> None: + """Run the interactive command loop.""" + while True: + try: + command = input("> ").strip() + except EOFError: + break + except KeyboardInterrupt: + print("\n") + break + + if not command: + continue + + if not await process_command(session, command): + break + + +async def main() -> None: + """Run the interactive URL elicitation client.""" + server_url = "http://localhost:8000/sse" + + print("=" * 60) + print("URL Elicitation Client Example") + print("=" * 60) + print(f"\nConnecting to: {server_url}") + print("(Start server with: cd examples/snippets && uv run server elicitation sse)") + + try: + async with sse_client(server_url) as (read, write): + async with ClientSession( + read, + write, + elicitation_callback=handle_elicitation, + ) as session: + await session.initialize() + print("\nConnected! Type 'help' for available commands.\n") + await run_command_loop(session) + + except ConnectionRefusedError: + print(f"\nError: Could not connect to {server_url}") + print("Make sure the elicitation server is running:") + print(" cd examples/snippets && uv run server elicitation sse") + except Exception as e: + print(f"\nError: {e}") + raise + + +def run() -> None: + """Entry point for the client script.""" + asyncio.run(main()) + + +if __name__ == "__main__": + run() diff --git a/examples/snippets/pyproject.toml b/examples/snippets/pyproject.toml index 76791a55a7..4e68846a09 100644 --- a/examples/snippets/pyproject.toml +++ b/examples/snippets/pyproject.toml @@ -21,3 +21,4 @@ completion-client = "clients.completion_client:main" direct-execution-server = "servers.direct_execution:main" display-utilities-client = "clients.display_utilities:main" oauth-client = "clients.oauth_client:run" +elicitation-client = "clients.url_elicitation_client:run" diff --git a/examples/snippets/servers/elicitation.py b/examples/snippets/servers/elicitation.py index 2c8a3b35ac..a1a65fb32c 100644 --- 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 index 077ff9af64..203a516613 100644 --- 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/client/session.py b/src/mcp/client/session.py index 3817ca6b5d..be47d681fb 100644 --- a/src/mcp/client/session.py +++ b/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 @@ -552,5 +557,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 diff --git a/src/mcp/server/elicitation.py b/src/mcp/server/elicitation.py index c2d98de384..49195415bf 100644 --- a/src/mcp/server/elicitation.py +++ b/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) @@ -99,20 +108,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, @@ -129,3 +140,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. When the elicitation completes, + the server should send an ElicitCompleteNotification to notify the client. + + 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}") diff --git a/src/mcp/server/fastmcp/server.py b/src/mcp/server/fastmcp/server.py index 5d6781f83d..2e596c9f9a 100644 --- 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 @@ -1204,6 +1208,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/server/session.py b/src/mcp/server/session.py index 677ffef89f..b116fbe384 100644 --- a/src/mcp/server/session.py +++ b/src/mcp/server/session.py @@ -335,19 +335,42 @@ 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( + params=types.ElicitRequestFormParams( message=message, requestedSchema=requestedSchema, ), @@ -357,6 +380,41 @@ 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.ElicitRequestURLParams( + 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 +457,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) diff --git a/src/mcp/shared/exceptions.py b/src/mcp/shared/exceptions.py index 97a1c09a9f..4943114912 100644 --- 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/src/mcp/types.py b/src/mcp/types.py index 8955de694e..dd9775f8c8 100644 --- a/src/mcp/types.py +++ b/src/mcp/types.py @@ -146,6 +146,10 @@ class JSONRPCResponse(BaseModel): model_config = ConfigDict(extra="allow") +# MCP-specific error codes in the range [-32000, -32099] +URL_ELICITATION_REQUIRED = -32042 +"""Error code indicating that a URL mode elicitation is required before the request can be processed.""" + # SDK error codes CONNECTION_CLOSED = -32000 # REQUEST_TIMEOUT = -32001 # the typescript sdk uses this @@ -272,8 +276,29 @@ class SamplingToolsCapability(BaseModel): model_config = ConfigDict(extra="allow") +class FormElicitationCapability(BaseModel): + """Capability for form mode elicitation.""" + + model_config = ConfigDict(extra="allow") + + +class UrlElicitationCapability(BaseModel): + """Capability for URL mode elicitation.""" + + model_config = ConfigDict(extra="allow") + + class ElicitationCapability(BaseModel): - """Capability for elicitation operations.""" + """Capability for elicitation operations. + + Clients must support at least one mode (form or url). + """ + + form: FormElicitationCapability | None = None + """Present if the client supports form mode elicitation.""" + + url: UrlElicitationCapability | None = None + """Present if the client supports URL mode elicitation.""" model_config = ConfigDict(extra="allow") @@ -1411,6 +1436,32 @@ class CancelledNotification(Notification[CancelledNotificationParams, Literal["n params: CancelledNotificationParams +class ElicitCompleteNotificationParams(NotificationParams): + """Parameters for elicitation completion notifications.""" + + elicitationId: str + """The unique identifier of the elicitation that was completed.""" + + model_config = ConfigDict(extra="allow") + + +class ElicitCompleteNotification( + Notification[ElicitCompleteNotificationParams, Literal["notifications/elicitation/complete"]] +): + """ + A notification from the server to the client, informing it that a URL mode + elicitation has been completed. + + Clients MAY use the notification to automatically retry requests that received a + URLElicitationRequiredError, update the user interface, or otherwise continue + an interaction. However, because delivery of the notification is not guaranteed, + clients must not wait indefinitely for a notification from the server. + """ + + method: Literal["notifications/elicitation/complete"] = "notifications/elicitation/complete" + params: ElicitCompleteNotificationParams + + class ClientRequest( RootModel[ PingRequest @@ -1442,14 +1493,58 @@ class ClientNotification( """Schema for elicitation requests.""" -class ElicitRequestParams(RequestParams): - """Parameters for elicitation requests.""" +class ElicitRequestFormParams(RequestParams): + """Parameters for form mode elicitation requests. + + Form mode collects non-sensitive information from the user via an in-band form + rendered by the client. + """ + + mode: Literal["form"] = "form" + """The elicitation mode (always "form" for this type).""" message: str + """The message to present to the user describing what information is being requested.""" + requestedSchema: ElicitRequestedSchema + """ + A restricted subset of JSON Schema defining the structure of expected response. + Only top-level properties are allowed, without nesting. + """ + model_config = ConfigDict(extra="allow") +class ElicitRequestURLParams(RequestParams): + """Parameters for URL mode elicitation requests. + + URL mode directs users to external URLs for sensitive out-of-band interactions + like OAuth flows, credential collection, or payment processing. + """ + + mode: Literal["url"] = "url" + """The elicitation mode (always "url" for this type).""" + + message: str + """The message to present to the user explaining why the interaction is needed.""" + + url: str + """The URL that the user should navigate to.""" + + elicitationId: str + """ + The ID of the elicitation, which must be unique within the context of the server. + The client MUST treat this ID as an opaque value. + """ + + model_config = ConfigDict(extra="allow") + + +# Union type for elicitation request parameters +ElicitRequestParams: TypeAlias = ElicitRequestURLParams | ElicitRequestFormParams +"""Parameters for elicitation requests - either form or URL mode.""" + + class ElicitRequest(Request[ElicitRequestParams, Literal["elicitation/create"]]): """A request from the server to elicit information from the client.""" @@ -1463,18 +1558,33 @@ class ElicitResult(Result): action: Literal["accept", "decline", "cancel"] """ The user action in response to the elicitation. - - "accept": User submitted the form/confirmed the action + - "accept": User submitted the form/confirmed the action (or consented to URL navigation) - "decline": User explicitly declined the action - "cancel": User dismissed without making an explicit choice """ content: dict[str, str | int | float | bool | list[str] | None] | None = None """ - The submitted form data, only present when action is "accept". - Contains values matching the requested schema. + The submitted form data, only present when action is "accept" in form mode. + Contains values matching the requested schema. Values can be strings, integers, + booleans, or arrays of strings. + For URL mode, this field is omitted. """ +class ElicitationRequiredErrorData(BaseModel): + """Error data for URLElicitationRequiredError. + + Servers return this when a request cannot be processed until one or more + URL mode elicitations are completed. + """ + + elicitations: list[ElicitRequestURLParams] + """List of URL mode elicitations that must be completed.""" + + model_config = ConfigDict(extra="allow") + + class ClientResult(RootModel[EmptyResult | CreateMessageResult | ListRootsResult | ElicitResult]): pass @@ -1492,6 +1602,7 @@ class ServerNotification( | ResourceListChangedNotification | ToolListChangedNotification | PromptListChangedNotification + | ElicitCompleteNotification ] ): pass diff --git a/tests/server/fastmcp/test_elicitation.py b/tests/server/fastmcp/test_elicitation.py index 359fea6197..597b291785 100644 --- a/tests/server/fastmcp/test_elicitation.py +++ b/tests/server/fastmcp/test_elicitation.py @@ -7,6 +7,7 @@ import pytest from pydantic import BaseModel, Field +from mcp import types from mcp.client.session import ClientSession, ElicitationFnT from mcp.server.fastmcp import Context, FastMCP from mcp.server.session import ServerSession @@ -288,6 +289,7 @@ async def defaults_tool(ctx: Context[ServerSession, None]) -> str: # First verify that defaults are present in the JSON schema sent to clients async def callback_schema_verify(context: RequestContext[ClientSession, None], params: ElicitRequestParams): # Verify the schema includes defaults + assert isinstance(params, types.ElicitRequestFormParams), "Expected form mode elicitation" schema = params.requestedSchema props = schema["properties"] diff --git a/tests/server/fastmcp/test_url_elicitation.py b/tests/server/fastmcp/test_url_elicitation.py new file mode 100644 index 0000000000..a4d3b2e643 --- /dev/null +++ b/tests/server/fastmcp/test_url_elicitation.py @@ -0,0 +1,394 @@ +"""Test URL mode elicitation feature (SEP 1036).""" + +import anyio +import pytest + +from mcp import types +from mcp.client.session import ClientSession +from mcp.server.elicitation import CancelledElicitation, DeclinedElicitation +from mcp.server.fastmcp import Context, FastMCP +from mcp.server.session import ServerSession +from mcp.shared.context import RequestContext +from mcp.shared.memory import create_connected_server_and_client_session +from mcp.types import ElicitRequestParams, ElicitResult, TextContent + + +@pytest.mark.anyio +async def test_url_elicitation_accept(): + """Test URL mode elicitation with user acceptance.""" + mcp = FastMCP(name="URLElicitationServer") + + @mcp.tool(description="A tool that uses URL elicitation") + async def request_api_key(ctx: Context[ServerSession, None]) -> str: + result = await ctx.session.elicit_url( + message="Please provide your API key to continue.", + url="https://example.com/api_key_setup", + elicitation_id="test-elicitation-001", + ) + # Test only checks accept path + return f"User {result.action}" + + # Create elicitation callback that accepts URL mode + async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + assert params.mode == "url" + assert params.url == "https://example.com/api_key_setup" + assert params.elicitationId == "test-elicitation-001" + assert params.message == "Please provide your API key to continue." + return ElicitResult(action="accept") + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=elicitation_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("request_api_key", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "User accept" + + +@pytest.mark.anyio +async def test_url_elicitation_decline(): + """Test URL mode elicitation with user declining.""" + mcp = FastMCP(name="URLElicitationDeclineServer") + + @mcp.tool(description="A tool that uses URL elicitation") + async def oauth_flow(ctx: Context[ServerSession, None]) -> str: + result = await ctx.session.elicit_url( + message="Authorize access to your files.", + url="https://example.com/oauth/authorize", + elicitation_id="oauth-001", + ) + # Test only checks decline path + return f"User {result.action} authorization" + + async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + assert params.mode == "url" + return ElicitResult(action="decline") + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=elicitation_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("oauth_flow", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "User decline authorization" + + +@pytest.mark.anyio +async def test_url_elicitation_cancel(): + """Test URL mode elicitation with user cancelling.""" + mcp = FastMCP(name="URLElicitationCancelServer") + + @mcp.tool(description="A tool that uses URL elicitation") + async def payment_flow(ctx: Context[ServerSession, None]) -> str: + result = await ctx.session.elicit_url( + message="Complete payment to proceed.", + url="https://example.com/payment", + elicitation_id="payment-001", + ) + # Test only checks cancel path + return f"User {result.action} payment" + + async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + assert params.mode == "url" + return ElicitResult(action="cancel") + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=elicitation_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("payment_flow", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "User cancel payment" + + +@pytest.mark.anyio +async def test_url_elicitation_helper_function(): + """Test the elicit_url helper function.""" + from mcp.server.elicitation import elicit_url + + mcp = FastMCP(name="URLElicitationHelperServer") + + @mcp.tool(description="Tool using elicit_url helper") + async def setup_credentials(ctx: Context[ServerSession, None]) -> str: + result = await elicit_url( + session=ctx.session, + message="Set up your credentials", + url="https://example.com/setup", + elicitation_id="setup-001", + ) + # Test only checks accept path - return the type name + return type(result).__name__ + + async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + return ElicitResult(action="accept") + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=elicitation_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("setup_credentials", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "AcceptedUrlElicitation" + + +@pytest.mark.anyio +async def test_url_no_content_in_response(): + """Test that URL mode elicitation responses don't include content field.""" + mcp = FastMCP(name="URLContentCheckServer") + + @mcp.tool(description="Check URL response format") + async def check_url_response(ctx: Context[ServerSession, None]) -> str: + result = await ctx.session.elicit_url( + message="Test message", + url="https://example.com/test", + elicitation_id="test-001", + ) + + # URL mode responses should not have content + assert result.content is None + return f"Action: {result.action}, Content: {result.content}" + + async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + # Verify that this is URL mode + assert params.mode == "url" + assert isinstance(params, types.ElicitRequestURLParams) + # URL params have url and elicitationId, not requestedSchema + assert params.url == "https://example.com/test" + assert params.elicitationId == "test-001" + # Return without content - this is correct for URL mode + return ElicitResult(action="accept") + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=elicitation_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("check_url_response", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert "Content: None" in result.content[0].text + + +@pytest.mark.anyio +async def test_form_mode_still_works(): + """Ensure form mode elicitation still works after SEP 1036.""" + from pydantic import BaseModel, Field + + mcp = FastMCP(name="FormModeBackwardCompatServer") + + class NameSchema(BaseModel): + name: str = Field(description="Your name") + + @mcp.tool(description="Test form mode") + async def ask_name(ctx: Context[ServerSession, None]) -> str: + result = await ctx.elicit(message="What is your name?", schema=NameSchema) + # Test only checks accept path with data + assert result.action == "accept" + assert result.data is not None + return f"Hello, {result.data.name}!" + + async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + # Verify form mode parameters + assert params.mode == "form" + assert isinstance(params, types.ElicitRequestFormParams) + # Form params have requestedSchema, not url/elicitationId + assert params.requestedSchema is not None + return ElicitResult(action="accept", content={"name": "Alice"}) + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=elicitation_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("ask_name", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "Hello, Alice!" + + +@pytest.mark.anyio +async def test_elicit_complete_notification(): + """Test that elicitation completion notifications can be sent and received.""" + mcp = FastMCP(name="ElicitCompleteServer") + + # Track if the notification was sent + notification_sent = False + + @mcp.tool(description="Tool that sends completion notification") + async def trigger_elicitation(ctx: Context[ServerSession, None]) -> str: + nonlocal notification_sent + + # Simulate an async operation (e.g., user completing auth in browser) + elicitation_id = "complete-test-001" + + # Send completion notification + await ctx.session.send_elicit_complete(elicitation_id) + notification_sent = True + + return "Elicitation completed" + + async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + return ElicitResult(action="accept") # pragma: no cover + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=elicitation_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("trigger_elicitation", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "Elicitation completed" + + # Give time for notification to be processed + await anyio.sleep(0.1) + + # Verify the notification was sent + assert notification_sent + + +@pytest.mark.anyio +async def test_url_elicitation_required_error_code(): + """Test that the URL_ELICITATION_REQUIRED error code is correct.""" + # Verify the error code matches the specification (SEP 1036) + assert types.URL_ELICITATION_REQUIRED == -32042, ( + "URL_ELICITATION_REQUIRED error code must be -32042 per SEP 1036 specification" + ) + + +@pytest.mark.anyio +async def test_elicit_url_typed_results(): + """Test that elicit_url returns properly typed result objects.""" + from mcp.server.elicitation import elicit_url + + mcp = FastMCP(name="TypedResultsServer") + + @mcp.tool(description="Test declined result") + async def test_decline(ctx: Context[ServerSession, None]) -> str: + result = await elicit_url( + session=ctx.session, + message="Test decline", + url="https://example.com/decline", + elicitation_id="decline-001", + ) + + if isinstance(result, DeclinedElicitation): + return "Declined" + return "Not declined" # pragma: no cover + + @mcp.tool(description="Test cancelled result") + async def test_cancel(ctx: Context[ServerSession, None]) -> str: + result = await elicit_url( + session=ctx.session, + message="Test cancel", + url="https://example.com/cancel", + elicitation_id="cancel-001", + ) + + if isinstance(result, CancelledElicitation): + return "Cancelled" + return "Not cancelled" # pragma: no cover + + # Test declined result + async def decline_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + return ElicitResult(action="decline") + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=decline_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("test_decline", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "Declined" + + # Test cancelled result + async def cancel_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + return ElicitResult(action="cancel") + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=cancel_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("test_cancel", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "Cancelled" + + +@pytest.mark.anyio +async def test_deprecated_elicit_method(): + """Test the deprecated elicit() method for backward compatibility.""" + from pydantic import BaseModel, Field + + mcp = FastMCP(name="DeprecatedElicitServer") + + class EmailSchema(BaseModel): + email: str = Field(description="Email address") + + @mcp.tool(description="Test deprecated elicit method") + async def use_deprecated_elicit(ctx: Context[ServerSession, None]) -> str: + # Use the deprecated elicit() method which should call elicit_form() + result = await ctx.session.elicit( + message="Enter your email", + requestedSchema=EmailSchema.model_json_schema(), + ) + + if result.action == "accept" and result.content: + return f"Email: {result.content.get('email', 'none')}" + return "No email provided" # pragma: no cover + + async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + # Verify this is form mode + assert params.mode == "form" + assert params.requestedSchema is not None + return ElicitResult(action="accept", content={"email": "test@example.com"}) + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=elicitation_callback + ) as client_session: + await client_session.initialize() + + result = await client_session.call_tool("use_deprecated_elicit", {}) + assert len(result.content) == 1 + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "Email: test@example.com" + + +@pytest.mark.anyio +async def test_ctx_elicit_url_convenience_method(): + """Test the ctx.elicit_url() convenience method (vs ctx.session.elicit_url()).""" + mcp = FastMCP(name="CtxElicitUrlServer") + + @mcp.tool(description="A tool that uses ctx.elicit_url() directly") + async def direct_elicit_url(ctx: Context[ServerSession, None]) -> str: + # Use ctx.elicit_url() directly instead of ctx.session.elicit_url() + result = await ctx.elicit_url( + message="Test the convenience method", + url="https://example.com/test", + elicitation_id="ctx-test-001", + ) + return f"Result: {result.action}" + + async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams): + assert params.mode == "url" + assert params.elicitationId == "ctx-test-001" + return ElicitResult(action="accept") + + async with create_connected_server_and_client_session( + mcp._mcp_server, elicitation_callback=elicitation_callback + ) as client_session: + await client_session.initialize() + result = await client_session.call_tool("direct_elicit_url", {}) + assert isinstance(result.content[0], TextContent) + assert result.content[0].text == "Result: accept" diff --git a/tests/shared/test_exceptions.py b/tests/shared/test_exceptions.py new file mode 100644 index 0000000000..8845dfe781 --- /dev/null +++ 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"