Utilisez l'élicitation avec votre passerelle AgentCore

Sur ces versions, le client déclare la capacité d'élicitation pendantinitialize, et l'élicitation arrive sous forme de elicitation/create requête sur le flux SSE ouvert. Définissez l'MCP-Protocol-Versionen-tête sur une version prise en charge par votre passerelle.

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
headers["MCP-Protocol-Version"] = "2025-06-18"

# 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

Dans la version2026-07-28, l'élicitation utilise le modèle de requêtes aller-retour multiples au lieu d'une demande initiée par le serveur sur le flux SSE. Le client déclare la capacité d'élicitation à chaque _meta demande. Si l'outil nécessite une saisie, la réponse est un input_required résultat contenant des résultats inputRequests et est opaquerequestState. Le client rassemble l'entrée et réessaie la demande d'origine avec une nouvelle demande idinputResponses, la et la non modifiée. requestState Les sessions et la initialize poignée de main ne sont pas utilisées. Votre passerelle supportedVersions doit inclure2026-07-28.

import requests

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

META = {
    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
    "io.modelcontextprotocol/clientInfo": {"name": "my-agent", "version": "1.0.0"},
    "io.modelcontextprotocol/clientCapabilities": {"elicitation": {"form": {}, "url": {}}}
}
headers = {
    "Content-Type": "application/json",
    "Accept": "application/json, text/event-stream",
    "Authorization": "Bearer YOUR_ACCESS_TOKEN",
    "MCP-Protocol-Version": "2026-07-28",
    "Mcp-Method": "tools/call",
    "Mcp-Name": "use_aws"
}
arguments = {"command": "aws s3 rm s3://my-bucket/important-file"}

# Step 1: Call the tool, declaring the elicitation capability in _meta
response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "tool-call-1",
    "method": "tools/call",
    "params": {"name": "use_aws", "arguments": arguments, "_meta": META}
}).json()

result = response["result"]
if result.get("resultType") == "input_required":
    # Step 2: Fulfill each input request (form elicitation)
    input_responses = {}
    for key, input_request in result.get("inputRequests", {}).items():
        print(f"Form elicitation received: {input_request['params']['message']}")
        input_responses[key] = {"action": "accept", "content": {"confirm": True}}

    # Step 3: Retry the tool call with a new id, the input responses,
    # and the requestState echoed back unmodified
    retry_params = {"name": "use_aws", "arguments": arguments, "_meta": META,
                    "inputResponses": input_responses}
    if "requestState" in result:
        retry_params["requestState"] = result["requestState"]
    response = requests.post(gateway_url, headers=headers, json={
        "jsonrpc": "2.0",
        "id": "tool-call-2",
        "method": "tools/call",
        "params": retry_params
    }).json()
    result = response["result"]

print(f"Tool result: {result}")

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"}]}
)

Sur ces versions, l'élicitation d'URL arrive sous la forme d'une elicitation/create requête sur le flux SSE ouvert au sein d'une session MCP. Définissez l'MCP-Protocol-Versionen-tête sur une version prise en charge par votre passerelle.

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-Protocol-Version": "2025-11-25",
    "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

Sur la version2026-07-28, l'élicitation d'URL arrive en input_required conséquence via le modèle de demandes aller-retour multiples plutôt que sur le flux SSE. L'utilisateur effectue l'action sur l'URL, puis le client tente à nouveau la demande d'origine avec une nouvelle demande idinputResponses, la et la non requestState modifiée. Les sessions ne sont pas utilisées. Votre passerelle supportedVersions doit inclure2026-07-28.

import requests
import webbrowser

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

META = {
    "io.modelcontextprotocol/protocolVersion": "2026-07-28",
    "io.modelcontextprotocol/clientInfo": {"name": "my-agent", "version": "1.0.0"},
    "io.modelcontextprotocol/clientCapabilities": {"elicitation": {"form": {}, "url": {}}}
}
headers = {
    "Content-Type": "application/json",
    "Accept": "application/json, text/event-stream",
    "Authorization": "Bearer YOUR_ACCESS_TOKEN",
    "MCP-Protocol-Version": "2026-07-28",
    "Mcp-Method": "tools/call",
    "Mcp-Name": "access_github_repo"
}
arguments = {"repo": "my-org/my-repo"}

# Step 1: Call the tool, declaring the elicitation capability in _meta
response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "tool-call-1",
    "method": "tools/call",
    "params": {"name": "access_github_repo", "arguments": arguments, "_meta": META}
}).json()

result = response["result"]
if result.get("resultType") == "input_required":
    # Step 2: Open the browser for the user to complete each URL elicitation
    input_responses = {}
    for key, input_request in result.get("inputRequests", {}).items():
        params = input_request["params"]
        print(f"URL elicitation: {params['message']}")
        webbrowser.open(params["url"])
        input("Press Enter after completing authentication...")
        input_responses[key] = {"action": "accept"}

    # Step 3: Retry the tool call with a new id, the input responses,
    # and the requestState echoed back unmodified
    retry_params = {"name": "access_github_repo", "arguments": arguments, "_meta": META,
                    "inputResponses": input_responses}
    if "requestState" in result:
        retry_params["requestState"] = result["requestState"]
    response = requests.post(gateway_url, headers=headers, json={
        "jsonrpc": "2.0",
        "id": "tool-call-2",
        "method": "tools/call",
        "params": retry_params
    }).json()
    result = response["result"]

print(f"Tool result: {result}")

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"}]}
)