Skip to content

Commit 049f4d2

Browse files
felixweinbergerfelixweinberger
committed
Add URL elicitation client example
Demonstrates how clients handle URL elicitation requests from servers: - Client elicitation capability declaration via elicitation_callback - Handling elicitation requests with security warnings - Catching UrlElicitationRequiredError from tool calls - Browser interaction for out-of-band authentication - Interactive CLI for testing with elicitation server
1 parent 3f0ac89 commit 049f4d2

File tree

2 files changed

+319
-0
lines changed

2 files changed

+319
-0
lines changed

‎examples/snippets/clients/url_elicitation_client.py‎

Lines changed: 318 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,318 @@
1+
"""URL Elicitation Client Example.
2+
3+
Demonstrates how clients handle URL elicitation requests from servers.
4+
This is the Python equivalent of TypeScript SDK's elicitationUrlExample.ts,
5+
focused on URL elicitation patterns without OAuth complexity.
6+
7+
Features demonstrated:
8+
1. Client elicitation capability declaration
9+
2. Handling elicitation requests from servers via callback
10+
3. Catching UrlElicitationRequiredError from tool calls
11+
4. Browser interaction with security warnings
12+
5. Interactive CLI for testing
13+
14+
Run with:
15+
cd examples/snippets
16+
uv run elicitation-client
17+
18+
Requires a server with URL elicitation tools running. Start the elicitation
19+
server first:
20+
uv run server elicitation sse
21+
"""
22+
23+
from __future__ import annotations
24+
25+
import asyncio
26+
import json
27+
import subprocess
28+
import sys
29+
import webbrowser
30+
from typing import Any
31+
from urllib.parse import urlparse
32+
33+
from mcp import ClientSession, types
34+
from mcp.client.sse import sse_client
35+
from mcp.shared.context import RequestContext
36+
from mcp.shared.exceptions import McpError, UrlElicitationRequiredError
37+
from mcp.types import URL_ELICITATION_REQUIRED
38+
39+
40+
async def handle_elicitation(
41+
context: RequestContext[ClientSession, Any],
42+
params: types.ElicitRequestParams,
43+
) -> types.ElicitResult | types.ErrorData:
44+
"""Handle elicitation requests from the server.
45+
46+
This callback is invoked when the server sends an elicitation/request.
47+
For URL mode, we prompt the user and optionally open their browser.
48+
"""
49+
if params.mode == "url":
50+
return await handle_url_elicitation(params)
51+
else:
52+
# We only support URL mode in this example
53+
return types.ErrorData(
54+
code=types.INVALID_REQUEST,
55+
message=f"Unsupported elicitation mode: {params.mode}",
56+
)
57+
58+
59+
async def handle_url_elicitation(
60+
params: types.ElicitRequestParams,
61+
) -> types.ElicitResult:
62+
"""Handle URL mode elicitation - show security warning and optionally open browser.
63+
64+
This function demonstrates the security-conscious approach to URL elicitation:
65+
1. Display the full URL and domain for user inspection
66+
2. Show the server's reason for requesting this interaction
67+
3. Require explicit user consent before opening any URL
68+
"""
69+
# Extract URL parameters - these are available on URL mode requests
70+
url = getattr(params, "url", None)
71+
elicitation_id = getattr(params, "elicitationId", None)
72+
message = params.message
73+
74+
if not url:
75+
print("Error: No URL provided in elicitation request")
76+
return types.ElicitResult(action="cancel")
77+
78+
# Extract domain for security display
79+
domain = extract_domain(url)
80+
81+
# Security warning - always show the user what they're being asked to do
82+
print("\n" + "=" * 60)
83+
print("SECURITY WARNING: External URL Request")
84+
print("=" * 60)
85+
print("\nThe server is requesting you to open an external URL.")
86+
print(f"\n Domain: {domain}")
87+
print(f" Full URL: {url}")
88+
print("\n Server's reason:")
89+
print(f" {message}")
90+
print(f"\n Elicitation ID: {elicitation_id}")
91+
print("\n" + "-" * 60)
92+
93+
# Get explicit user consent
94+
try:
95+
response = input("\nOpen this URL in your browser? (y/n): ").strip().lower()
96+
except EOFError:
97+
return types.ElicitResult(action="cancel")
98+
99+
if response in ("n", "no"):
100+
print("URL navigation declined.")
101+
return types.ElicitResult(action="decline")
102+
elif response not in ("y", "yes"):
103+
print("Invalid response. Cancelling.")
104+
return types.ElicitResult(action="cancel")
105+
106+
# Open the browser
107+
print(f"\nOpening browser to: {url}")
108+
open_browser(url)
109+
110+
print("Waiting for you to complete the interaction in your browser...")
111+
print("(The server will continue once you've finished)")
112+
113+
return types.ElicitResult(action="accept")
114+
115+
116+
def extract_domain(url: str) -> str:
117+
"""Extract domain from URL for security display."""
118+
try:
119+
return urlparse(url).netloc
120+
except Exception:
121+
return "unknown"
122+
123+
124+
def open_browser(url: str) -> None:
125+
"""Open URL in the default browser."""
126+
try:
127+
if sys.platform == "darwin":
128+
subprocess.run(["open", url], check=False)
129+
elif sys.platform == "win32":
130+
subprocess.run(["start", url], shell=True, check=False)
131+
else:
132+
webbrowser.open(url)
133+
except Exception as e:
134+
print(f"Failed to open browser: {e}")
135+
print(f"Please manually open: {url}")
136+
137+
138+
async def call_tool_with_error_handling(
139+
session: ClientSession,
140+
tool_name: str,
141+
arguments: dict[str, Any],
142+
) -> types.CallToolResult | None:
143+
"""Call a tool, handling UrlElicitationRequiredError if raised.
144+
145+
When a server tool needs URL elicitation before it can proceed,
146+
it can either:
147+
1. Send an elicitation request directly (handled by elicitation_callback)
148+
2. Return an error with code -32042 (URL_ELICITATION_REQUIRED)
149+
150+
This function demonstrates handling case 2 - catching the error
151+
and processing the required URL elicitations.
152+
"""
153+
try:
154+
result = await session.call_tool(tool_name, arguments)
155+
156+
# Check if the tool returned an error in the result
157+
if result.isError:
158+
print(f"Tool returned error: {result.content}")
159+
return None
160+
161+
return result
162+
163+
except McpError as e:
164+
# Check if this is a URL elicitation required error
165+
if e.error.code == URL_ELICITATION_REQUIRED:
166+
print("\n[Tool requires URL elicitation to proceed]")
167+
168+
# Convert to typed error to access elicitations
169+
url_error = UrlElicitationRequiredError.from_error(e.error)
170+
171+
# Process each required elicitation
172+
for elicitation in url_error.elicitations:
173+
await handle_url_elicitation(elicitation)
174+
175+
return None
176+
else:
177+
# Re-raise other MCP errors
178+
print(f"MCP Error: {e.error.message} (code: {e.error.code})")
179+
return None
180+
181+
182+
def print_help() -> None:
183+
"""Print available commands."""
184+
print("\nAvailable commands:")
185+
print(" list-tools - List available tools")
186+
print(" call <name> [json-args] - Call a tool with optional JSON arguments")
187+
print(" secure-payment - Test URL elicitation via ctx.elicit_url()")
188+
print(" connect-service - Test URL elicitation via UrlElicitationRequiredError")
189+
print(" help - Show this help")
190+
print(" quit - Exit the program")
191+
192+
193+
def print_tool_result(result: types.CallToolResult | None) -> None:
194+
"""Print a tool call result."""
195+
if not result:
196+
return
197+
print("\nTool result:")
198+
for content in result.content:
199+
if isinstance(content, types.TextContent):
200+
print(f" {content.text}")
201+
else:
202+
print(f" [{content.type}]")
203+
204+
205+
async def handle_list_tools(session: ClientSession) -> None:
206+
"""Handle the list-tools command."""
207+
tools = await session.list_tools()
208+
if tools.tools:
209+
print("\nAvailable tools:")
210+
for tool in tools.tools:
211+
print(f" - {tool.name}: {tool.description or 'No description'}")
212+
else:
213+
print("No tools available")
214+
215+
216+
async def handle_call_command(session: ClientSession, command: str) -> None:
217+
"""Handle the call command."""
218+
parts = command.split(maxsplit=2)
219+
if len(parts) < 2:
220+
print("Usage: call <tool-name> [json-args]")
221+
return
222+
223+
tool_name = parts[1]
224+
args: dict[str, Any] = {}
225+
if len(parts) > 2:
226+
try:
227+
args = json.loads(parts[2])
228+
except json.JSONDecodeError as e:
229+
print(f"Invalid JSON arguments: {e}")
230+
return
231+
232+
print(f"\nCalling tool '{tool_name}' with args: {args}")
233+
result = await call_tool_with_error_handling(session, tool_name, args)
234+
print_tool_result(result)
235+
236+
237+
async def process_command(session: ClientSession, command: str) -> bool:
238+
"""Process a single command. Returns False if should exit."""
239+
if command in {"quit", "exit"}:
240+
print("Goodbye!")
241+
return False
242+
243+
if command == "help":
244+
print_help()
245+
elif command == "list-tools":
246+
await handle_list_tools(session)
247+
elif command.startswith("call "):
248+
await handle_call_command(session, command)
249+
elif command == "secure-payment":
250+
print("\nTesting secure_payment tool (uses ctx.elicit_url())...")
251+
result = await call_tool_with_error_handling(session, "secure_payment", {"amount": 99.99})
252+
print_tool_result(result)
253+
elif command == "connect-service":
254+
print("\nTesting connect_service tool (raises UrlElicitationRequiredError)...")
255+
result = await call_tool_with_error_handling(session, "connect_service", {"service_name": "github"})
256+
print_tool_result(result)
257+
else:
258+
print(f"Unknown command: {command}")
259+
print("Type 'help' for available commands.")
260+
261+
return True
262+
263+
264+
async def run_command_loop(session: ClientSession) -> None:
265+
"""Run the interactive command loop."""
266+
while True:
267+
try:
268+
command = input("> ").strip()
269+
except EOFError:
270+
break
271+
except KeyboardInterrupt:
272+
print("\n")
273+
break
274+
275+
if not command:
276+
continue
277+
278+
if not await process_command(session, command):
279+
break
280+
281+
282+
async def main() -> None:
283+
"""Run the interactive URL elicitation client."""
284+
server_url = "http://localhost:8000/sse"
285+
286+
print("=" * 60)
287+
print("URL Elicitation Client Example")
288+
print("=" * 60)
289+
print(f"\nConnecting to: {server_url}")
290+
print("(Start server with: cd examples/snippets && uv run server elicitation sse)")
291+
292+
try:
293+
async with sse_client(server_url) as (read, write):
294+
async with ClientSession(
295+
read,
296+
write,
297+
elicitation_callback=handle_elicitation,
298+
) as session:
299+
await session.initialize()
300+
print("\nConnected! Type 'help' for available commands.\n")
301+
await run_command_loop(session)
302+
303+
except ConnectionRefusedError:
304+
print(f"\nError: Could not connect to {server_url}")
305+
print("Make sure the elicitation server is running:")
306+
print(" cd examples/snippets && uv run server elicitation sse")
307+
except Exception as e:
308+
print(f"\nError: {e}")
309+
raise
310+
311+
312+
def run() -> None:
313+
"""Entry point for the client script."""
314+
asyncio.run(main())
315+
316+
317+
if __name__ == "__main__":
318+
run()

‎examples/snippets/pyproject.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,3 +21,4 @@ completion-client = "clients.completion_client:main"
2121
direct-execution-server = "servers.direct_execution:main"
2222
display-utilities-client = "clients.display_utilities:main"
2323
oauth-client = "clients.oauth_client:run"
24+
elicitation-client = "clients.url_elicitation_client:run"

0 commit comments

Comments
 (0)