Use elicitation with your AgentCore gateway

import requests
import json
import sseclient

gateway_url = "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp"
headers = {
    "Content-Type": "application/json",
    "Accept": "text/event-stream",
    "Authorization": "Bearer YOUR_ACCESS_TOKEN"
}

# Step 1: Initialize with elicitation capability
init_response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "init-request",
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-06-18",
        "capabilities": {"elicitation": {"form": {}, "url": {}}},
        "clientInfo": {"name": "my-agent", "version": "1.0.0"}
    }
})
session_id = init_response.headers["Mcp-Session-Id"]
headers["Mcp-Session-Id"] = session_id

# Step 2: Call tool (streaming response)
response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "tool-call-1",
    "method": "tools/call",
    "params": {
        "name": "use_aws",
        "arguments": {"command": "aws s3 rm s3://my-bucket/important-file"}
    }
}, stream=True)

# Step 3: Process SSE events
client = sseclient.SSEClient(response)
for event in client.events():
    data = json.loads(event.data)
    if data.get("method") == "elicitation/create":
        elicitation_id = data["id"]
        print(f"Form elicitation received: {data['params']['message']}")

        # Step 4: Send elicitation response
        requests.post(gateway_url, headers=headers, json={
            "jsonrpc": "2.0",
            "id": elicitation_id,
            "result": {"action": "accept", "content": {"confirm": True}}
        })
    elif "result" in data:
        print(f"Tool result: {data['result']}")
        break

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
import asyncio

async def elicitation_handler(request):
    """Handle form elicitation requests from the server."""
    print(f"Elicitation: {request.params.message}")
    return {"action": "accept", "content": {"confirm": True}}

async def use_elicitation(url, token):
    headers = {"Authorization": f"Bearer {token}"}

    async with streamablehttp_client(url=url, headers=headers) as (
        read_stream, write_stream, _
    ):
        async with ClientSession(
            read_stream, write_stream,
            elicitation_handler=elicitation_handler
        ) as session:
            await session.initialize()
            result = await session.call_tool(
                name="use_aws",
                arguments={"command": "aws s3 rm s3://my-bucket/important-file"}
            )
            print(f"Tool result: {result}")
            return result

asyncio.run(use_elicitation(
    url="https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
    token="YOUR_ACCESS_TOKEN"
))

from mcp.client.streamable_http import streamablehttp_client
from mcp.types import ElicitResult
from strands import Agent
from strands.tools.mcp import MCPClient

mcp_url = "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp"
access_token = "YOUR_ACCESS_TOKEN"

async def elicitation_callback(context, params):
    """Handle form elicitation requests from the MCP server target."""
    print(f"Elicitation: {params.message}")
    user_response = get_user_input(params)  # Your UI logic
    return ElicitResult(action="accept", content=user_response)

mcp_client = MCPClient(
    lambda: streamablehttp_client(
        mcp_url, headers={"Authorization": f"Bearer {access_token}"}
    ),
    elicitation_callback=elicitation_callback,
)

with mcp_client:
    agent = Agent(tools=mcp_client.list_tools_sync())
    response = agent("Delete the file s3://my-bucket/important-file using AWS CLI")
    print(response)

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext
from langchain.agents import create_agent
from mcp.shared.context import RequestContext
from mcp.types import ElicitRequestParams, ElicitResult

async def on_elicitation(
    mcp_context: RequestContext,
    params: ElicitRequestParams,
    context: CallbackContext,
) -> ElicitResult:
    """Handle elicitation requests from MCP servers."""
    print(f"[{context.server_name}] Elicitation: {params.message}")
    # Prompt user for input based on params.requestedSchema
    return ElicitResult(
        action="accept",
        content={"confirm": True},
    )

client = MultiServerMCPClient(
    {
        "gateway": {
            "url": "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
            "transport": "http",
            "headers": {"Authorization": "Bearer YOUR_ACCESS_TOKEN"},
        }
    },
    callbacks=Callbacks(on_elicitation=on_elicitation),
)

tools = await client.get_tools()
agent = create_agent("claude-sonnet-4-20250514", tools)

result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "Delete s3://my-bucket/important-file"}]}
)

import requests
import json
import sseclient
import webbrowser

gateway_url = "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp"
headers = {
    "Content-Type": "application/json",
    "Accept": "text/event-stream",
    "Authorization": "Bearer YOUR_ACCESS_TOKEN",
    "Mcp-Session-Id": "session-abc123def456"
}

# Call tool that triggers URL elicitation
response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "tool-call-2",
    "method": "tools/call",
    "params": {
        "name": "access_github_repo",
        "arguments": {"repo": "my-org/my-repo"}
    }
}, stream=True)

# Process SSE events
client = sseclient.SSEClient(response)
for event in client.events():
    data = json.loads(event.data)
    if data.get("method") == "elicitation/create":
        elicitation_id = data["id"]
        url = data["params"]["url"]
        print(f"URL elicitation: {data['params']['message']}")

        # Open browser for user to complete authentication
        webbrowser.open(url)
        input("Press Enter after completing authentication...")

        # Send elicitation response
        requests.post(gateway_url, headers=headers, json={
            "jsonrpc": "2.0",
            "id": elicitation_id,
            "result": {"action": "accept"}
        })
    elif "result" in data:
        print(f"Tool result: {data['result']}")
        break

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
import asyncio
import webbrowser

async def elicitation_handler(request):
    """Handle URL elicitation by opening the browser."""
    if hasattr(request.params, 'url') and request.params.url:
        print(f"Opening URL for authentication: {request.params.url}")
        webbrowser.open(request.params.url)
        input("Press Enter after completing authentication...")
        return {"action": "accept"}
    # Form mode fallback
    return {"action": "accept", "content": {}}

async def use_url_elicitation(url, token):
    headers = {"Authorization": f"Bearer {token}"}

    async with streamablehttp_client(url=url, headers=headers) as (
        read_stream, write_stream, _
    ):
        async with ClientSession(
            read_stream, write_stream,
            elicitation_handler=elicitation_handler
        ) as session:
            await session.initialize()
            result = await session.call_tool(
                name="access_github_repo",
                arguments={"repo": "my-org/my-repo"}
            )
            print(f"Tool result: {result}")
            return result

asyncio.run(use_url_elicitation(
    url="https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
    token="YOUR_ACCESS_TOKEN"
))

from mcp.client.streamable_http import streamablehttp_client
from mcp.types import ElicitResult
from strands import Agent
from strands.tools.mcp import MCPClient
import webbrowser

mcp_url = "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp"
access_token = "YOUR_ACCESS_TOKEN"

async def elicitation_callback(context, params):
    """Handle URL elicitation by opening the browser."""
    if hasattr(params, 'url') and params.url:
        print(f"Opening URL: {params.url}")
        webbrowser.open(params.url)
        input("Press Enter after completing authentication...")
        return ElicitResult(action="accept")
    return ElicitResult(action="accept", content={})

mcp_client = MCPClient(
    lambda: streamablehttp_client(
        mcp_url, headers={"Authorization": f"Bearer {access_token}"}
    ),
    elicitation_callback=elicitation_callback,
)

with mcp_client:
    agent = Agent(tools=mcp_client.list_tools_sync())
    response = agent("Access the my-org/my-repo GitHub repository")
    print(response)

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_mcp_adapters.callbacks import Callbacks, CallbackContext
from langchain.agents import create_agent
from mcp.shared.context import RequestContext
from mcp.types import ElicitRequestParams, ElicitResult
import webbrowser

async def on_elicitation(
    mcp_context: RequestContext,
    params: ElicitRequestParams,
    context: CallbackContext,
) -> ElicitResult:
    """Handle URL elicitation by opening the browser."""
    if params.url:
        print(f"[{context.server_name}] Opening URL: {params.url}")
        webbrowser.open(params.url)
        input("Press Enter after completing authentication...")
        return ElicitResult(action="accept")
    return ElicitResult(action="accept", content={})

client = MultiServerMCPClient(
    {
        "gateway": {
            "url": "https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp",
            "transport": "http",
            "headers": {"Authorization": "Bearer YOUR_ACCESS_TOKEN"},
        }
    },
    callbacks=Callbacks(on_elicitation=on_elicitation),
)

tools = await client.get_tools()
agent = create_agent("claude-sonnet-4-20250514", tools)

result = await agent.ainvoke(
    {"messages": [{"role": "user", "content": "Access the my-org/my-repo GitHub repository"}]}
)