mirror of
https://github.com/PrefectHQ/fastmcp.git
synced 2026-08-31 11:33:20 +02:00
150 lines
5.4 KiB
Python
150 lines
5.4 KiB
Python
"""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")
|