wehub-resource-sync
49b9bb6724
Deploy Docs / deploy-docs (push) Failing after 1s
Conformance Tests / client-conformance (push) Failing after 3s
Conformance Tests / server-conformance (push) Failing after 1s
GitHub Actions Security Analysis / zizmor (push) Failing after 1s
CI / checks (push) Failing after 59m20s
CI / all-green (push) Waiting to run
319 lines
10 KiB
Python
319 lines
10 KiB
Python
"""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 webbrowser
|
|
from typing import Any
|
|
from urllib.parse import urlparse
|
|
|
|
import mcp_types as types
|
|
from mcp_types import URL_ELICITATION_REQUIRED
|
|
|
|
from mcp import ClientSession
|
|
from mcp.client.context import ClientRequestContext
|
|
from mcp.client.sse import sse_client
|
|
from mcp.shared.exceptions import MCPError, UrlElicitationRequiredError
|
|
|
|
|
|
async def handle_elicitation(
|
|
context: ClientRequestContext,
|
|
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}",
|
|
)
|
|
|
|
|
|
ALLOWED_SCHEMES = {"http", "https"}
|
|
|
|
|
|
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. Validate the URL scheme before prompting the user
|
|
2. Display the full URL and domain for user inspection
|
|
3. Show the server's reason for requesting this interaction
|
|
4. 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")
|
|
|
|
# Reject dangerous URL schemes before prompting the user
|
|
parsed = urlparse(str(url))
|
|
if parsed.scheme.lower() not in ALLOWED_SCHEMES:
|
|
print(f"\nRejecting URL with disallowed scheme '{parsed.scheme}': {url}")
|
|
return types.ElicitResult(action="decline")
|
|
|
|
# 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}")
|
|
try:
|
|
webbrowser.open(url)
|
|
except Exception as e:
|
|
print(f"Failed to open browser: {e}")
|
|
print(f"Please manually open: {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"
|
|
|
|
|
|
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.is_error:
|
|
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.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 <name> [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 <tool-name> [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()
|