diff --git a/docs/clients/elicitation.mdx b/docs/clients/elicitation.mdx index dc0135ae3..08cdfd448 100644 --- a/docs/clients/elicitation.mdx +++ b/docs/clients/elicitation.mdx @@ -112,6 +112,34 @@ async def elicitation_handler(message, response_type, params, context): - **`decline`**: User chose not to provide the requested information. Omit `content`. - **`cancel`**: User cancelled the entire operation. Omit `content`. +## URL Elicitation + +Servers can request a different kind of interaction: instead of asking the user to fill out a form, they can direct the user to visit a URL out-of-band—for OAuth consent, API-key entry, or payment. This keeps sensitive data out of the LLM context. See [URL Elicitation](/servers/elicitation#url-elicitation) on the server side for why this matters. + +A URL elicitation is distinguishable from a form elicitation in two ways: `response_type` is `None` (there is no schema to fill out), and `params` is an `ElicitRequestURLParams` carrying the URL the user should visit. Check for it with `isinstance`, then present the URL and return an accept/decline/cancel action—no `content` is needed for any of them. + +```python +from fastmcp import Client +from fastmcp.client.elicitation import ElicitResult, ElicitRequestURLParams + +async def elicitation_handler(message, response_type, params, context): + if isinstance(params, ElicitRequestURLParams): + print(f"{message}\nVisit: {params.url}") + approved = input("Open this URL? [y/N] ").lower() == "y" + return ElicitResult(action="accept" if approved else "decline") + + # Otherwise this is a normal form elicitation + user_input = input(f"{message}: ") + return response_type(value=user_input) + +client = Client( + "my_mcp_server.py", + elicitation_handler=elicitation_handler, +) +``` + +Returning `action="accept"` signals that the user consented to navigate to the URL. The actual interaction completes out-of-band in the user's browser, so your handler never sees the secrets exchanged there. + ## Example A file management tool might ask which directory to create: diff --git a/docs/servers/elicitation.mdx b/docs/servers/elicitation.mdx index 923e704c6..a65095262 100644 --- a/docs/servers/elicitation.mdx +++ b/docs/servers/elicitation.mdx @@ -116,6 +116,45 @@ Elicitation requires the client to implement an elicitation handler. If a client See [Client Elicitation](/clients/elicitation) for details on how clients handle these requests. +## URL Elicitation + +Some interactions should never pass through the model. OAuth consent screens, API-key entry, and payment flows all involve secrets or credentials that you don't want flowing through the LLM context or the MCP client's form UI. URL elicitation solves this by directing the user to visit a URL out-of-band, where the sensitive interaction happens in a real browser session. The server learns only whether the user agreed to navigate—never what they entered. + +Call `ctx.elicit_url()` with a message explaining why the interaction is needed and the URL the user should visit. FastMCP generates a unique elicitation ID for each call automatically. + +```python +from fastmcp import FastMCP, Context +from fastmcp.server.elicitation import ( + AcceptedUrlElicitation, + DeclinedElicitation, + CancelledElicitation, +) + +mcp = FastMCP("Payments Server") + +@mcp.tool +async def connect_account(ctx: Context) -> str: + """Connect the user's account via an out-of-band OAuth flow.""" + result = await ctx.elicit_url( + message="Authorize access to your account to continue", + url="https://example.com/oauth/authorize", + ) + + match result: + case AcceptedUrlElicitation(): + return "Account authorization started—complete it in your browser." + case DeclinedElicitation(): + return "Authorization declined" + case CancelledElicitation(): + return "Authorization cancelled" +``` + +The result is one of `AcceptedUrlElicitation`, `DeclinedElicitation`, or `CancelledElicitation`. An accepted result means the user consented to navigate to the URL—the actual interaction (signing in, entering a key, paying) completes out-of-band in the browser, not through MCP. + + +URL elicitation is not yet supported from [background tasks](/servers/tasks). Calling `ctx.elicit_url()` inside a `@mcp.tool(task=True)` tool raises `NotImplementedError`; use it from a regular tool instead. + + ## Schema and Response Types The server must send a schema to the client indicating the type of data it expects in response to the elicitation request. The MCP spec only supports a limited subset of JSON Schema types for elicitation responses—specifically JSON **objects** with **primitive** properties including `string`, `number` (or `integer`), `boolean`, and `enum` fields. diff --git a/fastmcp_slim/fastmcp/client/elicitation.py b/fastmcp_slim/fastmcp/client/elicitation.py index 2b9ce5db4..2d816396a 100644 --- a/fastmcp_slim/fastmcp/client/elicitation.py +++ b/fastmcp_slim/fastmcp/client/elicitation.py @@ -6,7 +6,11 @@ from typing import Any, Generic, TypeAlias import mcp_types from mcp import ClientSession from mcp.client.session import ClientRequestContext, ElicitationFnT -from mcp_types import ElicitRequestFormParams, ElicitRequestParams +from mcp_types import ( + ElicitRequestFormParams, + ElicitRequestParams, + ElicitRequestURLParams, +) from mcp_types import ElicitResult as MCPElicitResult from pydantic_core import to_jsonable_python from typing_extensions import TypeVar @@ -14,7 +18,12 @@ from typing_extensions import TypeVar from fastmcp.client._sdk_context_shim import LifespanContextT, RequestContext from fastmcp.utilities.json_schema_type import json_schema_to_type -__all__ = ["ElicitRequestParams", "ElicitResult", "ElicitationHandler"] +__all__ = [ + "ElicitRequestParams", + "ElicitRequestURLParams", + "ElicitResult", + "ElicitationHandler", +] T = TypeVar("T", default=Any) diff --git a/fastmcp_slim/fastmcp/server/context.py b/fastmcp_slim/fastmcp/server/context.py index 3d0709736..ef356b4a7 100644 --- a/fastmcp_slim/fastmcp/server/context.py +++ b/fastmcp_slim/fastmcp/server/context.py @@ -1,6 +1,7 @@ from __future__ import annotations import logging +import uuid import warnings import weakref from collections.abc import Callable, Generator, Mapping, Sequence @@ -31,8 +32,10 @@ from fastmcp.resources.base import ResourceResult from fastmcp.server.dependencies import FastMCPRequestContext, fastmcp_request_ctx from fastmcp.server.elicitation import ( AcceptedElicitation, + AcceptedUrlElicitation, CancelledElicitation, DeclinedElicitation, + UrlElicitationResult, handle_elicit_accept, parse_elicit_response_type, ) @@ -1229,6 +1232,61 @@ class Context: else: raise ValueError(f"Unexpected elicitation action: {result.action}") + async def elicit_url( + self, + message: str, + url: str, + ) -> UrlElicitationResult: + """Direct the user to an external URL for an out-of-band interaction. + + URL-mode elicitation (SEP-1036) asks the user to visit a URL instead of + answering a schema form. The interaction happens outside the MCP client, + so sensitive data never enters the LLM context. Use it for OAuth consent, + API-key entry, payment, or any flow where credentials must not pass + through the model. The client reports only whether the user consented to + navigate; the actual result of the interaction is obtained out-of-band. + + A unique ``elicitation_id`` is generated automatically for each call. + + Args: + message: A human-readable explanation of why the interaction is needed. + url: The URL the user should navigate to. + + Returns: + An ``AcceptedUrlElicitation`` if the user consented to navigate, a + ``DeclinedElicitation`` if they explicitly declined, or a + ``CancelledElicitation`` if they dismissed the request. + + Note: + URL-mode elicitation is not yet supported from background tasks + (``@server.tool(task=True)``); calling it there raises + ``NotImplementedError``. + """ + if self.is_background_task: + raise NotImplementedError( + "URL elicitation (ctx.elicit_url) is not yet supported from " + "background tasks. The background-task elicitation relay only " + "carries form-mode requests. Use ctx.elicit_url from a regular " + "(foreground) request context instead." + ) + + elicitation_id = str(uuid.uuid4()) + result = await self.session.elicit_url( + message=message, + url=url, + elicitation_id=elicitation_id, + related_request_id=self.request_id, + ) + + if result.action == "accept": + return AcceptedUrlElicitation() + elif result.action == "decline": + return DeclinedElicitation() + elif result.action == "cancel": + return CancelledElicitation() + else: + raise ValueError(f"Unexpected elicitation action: {result.action}") + async def _elicit_for_task( self, message: str, diff --git a/fastmcp_slim/fastmcp/server/elicitation.py b/fastmcp_slim/fastmcp/server/elicitation.py index 5947d223b..f068f258c 100644 --- a/fastmcp_slim/fastmcp/server/elicitation.py +++ b/fastmcp_slim/fastmcp/server/elicitation.py @@ -5,8 +5,10 @@ from enum import Enum from typing import Any, Generic, Literal, cast, get_origin from mcp.server.elicitation import ( + AcceptedUrlElicitation, CancelledElicitation, DeclinedElicitation, + UrlElicitationResult, ) from pydantic import BaseModel from pydantic.json_schema import GenerateJsonSchema, JsonSchemaValue @@ -19,10 +21,12 @@ from fastmcp.utilities.types import get_cached_typeadapter __all__ = [ "AcceptedElicitation", + "AcceptedUrlElicitation", "CancelledElicitation", "DeclinedElicitation", "ElicitConfig", "ScalarElicitationType", + "UrlElicitationResult", "get_elicitation_schema", "handle_elicit_accept", "parse_elicit_response_type", diff --git a/tests/client/test_url_elicitation.py b/tests/client/test_url_elicitation.py new file mode 100644 index 000000000..651e08e9e --- /dev/null +++ b/tests/client/test_url_elicitation.py @@ -0,0 +1,150 @@ +"""End-to-end tests for URL-mode elicitation (SEP-1036). + +URL elicitation directs the user to an external URL for an out-of-band +interaction (OAuth consent, API-key entry, payment) instead of answering a +schema form, so sensitive data never enters the LLM context. +""" + +from typing import Any + +import pytest +from mcp_types import ElicitRequestURLParams + +from fastmcp import Context, FastMCP +from fastmcp.client.client import Client +from fastmcp.client.elicitation import ElicitResult +from fastmcp.server.elicitation import ( + AcceptedUrlElicitation, + CancelledElicitation, + DeclinedElicitation, +) + + +@pytest.fixture +def url_server(): + mcp = FastMCP("UrlElicitationServer") + + @mcp.tool + async def connect_account(context: Context) -> str: + result = await context.elicit_url( + message="Authorize access to your account", + url="https://example.com/oauth/authorize", + ) + if isinstance(result, AcceptedUrlElicitation): + return "authorized" + elif isinstance(result, DeclinedElicitation): + return "declined" + else: + assert isinstance(result, CancelledElicitation) + return "cancelled" + + return mcp + + +async def test_url_elicitation_accept(url_server): + """When the user consents to navigate, the tool proceeds.""" + + async def elicitation_handler(message, response_type, params, ctx): + return ElicitResult(action="accept") + + async with Client(url_server, elicitation_handler=elicitation_handler) as client: + result = await client.call_tool("connect_account") + assert result.data == "authorized" + + +async def test_url_elicitation_decline(url_server): + """A declined URL elicitation returns DeclinedElicitation to the server.""" + + async def elicitation_handler(message, response_type, params, ctx): + return ElicitResult(action="decline") + + async with Client(url_server, elicitation_handler=elicitation_handler) as client: + result = await client.call_tool("connect_account") + assert result.data == "declined" + + +async def test_url_elicitation_cancel(url_server): + """A cancelled URL elicitation returns CancelledElicitation to the server.""" + + async def elicitation_handler(message, response_type, params, ctx): + return ElicitResult(action="cancel") + + async with Client(url_server, elicitation_handler=elicitation_handler) as client: + result = await client.call_tool("connect_account") + assert result.data == "cancelled" + + +async def test_url_elicitation_handler_receives_url_params(url_server): + """The handler receives ElicitRequestURLParams with the url and message. + + For URL mode, `response_type` is None (there is no schema), and the params + are an ElicitRequestURLParams carrying the url the user should visit. + """ + captured: dict[str, Any] = {} + + async def elicitation_handler(message, response_type, params, ctx): + captured["message"] = message + captured["response_type"] = response_type + captured["params"] = params + return ElicitResult(action="accept") + + async with Client(url_server, elicitation_handler=elicitation_handler) as client: + await client.call_tool("connect_account") + + assert captured["message"] == "Authorize access to your account" + assert captured["response_type"] is None + assert isinstance(captured["params"], ElicitRequestURLParams) + assert captured["params"].url == "https://example.com/oauth/authorize" + assert captured["params"].mode == "url" + # An elicitation_id is auto-generated by the server. + assert captured["params"].elicitation_id + + +async def test_url_elicitation_handler_returning_none_accepts(url_server): + """Returning None (no explicit ElicitResult) is treated as acceptance.""" + + async def elicitation_handler(message, response_type, params, ctx): + return None + + async with Client(url_server, elicitation_handler=elicitation_handler) as client: + result = await client.call_tool("connect_account") + assert result.data == "authorized" + + +async def test_url_elicitation_generates_unique_ids(): + """Each elicit_url call auto-generates a distinct elicitation_id.""" + mcp = FastMCP("UrlElicitationServer") + ids: list[str | None] = [] + + @mcp.tool + async def connect_twice(context: Context) -> str: + await context.elicit_url(message="first", url="https://example.com/a") + await context.elicit_url(message="second", url="https://example.com/b") + return "done" + + async def elicitation_handler(message, response_type, params, ctx): + ids.append(params.elicitation_id) + return ElicitResult(action="accept") + + async with Client(mcp, elicitation_handler=elicitation_handler) as client: + await client.call_tool("connect_twice") + + assert len(ids) == 2 + assert ids[0] is not None + assert ids[1] is not None + assert ids[0] != ids[1] + + +async def test_url_elicitation_not_supported_in_background_task(): + """URL elicitation from a background task raises NotImplementedError. + + The background-task elicitation relay only carries form-mode requests, so + URL mode is explicitly unsupported there (rather than silently degrading). + """ + mcp = FastMCP("UrlElicitationServer") + ctx = Context(mcp, task_id="task-123") + + assert ctx.is_background_task + + with pytest.raises(NotImplementedError, match="not yet supported from"): + await ctx.elicit_url(message="hi", url="https://example.com")