Gunakan elicitation dengan gateway Anda AgentCore

Pada versi ini, klien menyatakan kemampuan elicitasi selamainitialize, dan elisitasi tiba sebagai elicitation/create permintaan pada aliran SSE terbuka. Atur MCP-Protocol-Version header ke versi yang didukung gateway Anda.

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

Pada versi2026-07-28, elicitation menggunakan pola permintaan multi pulang-pergi alih-alih permintaan yang diprakarsai server pada aliran SSE. Klien menyatakan kemampuan elicitasi _meta pada setiap permintaan. Jika alat membutuhkan input, responsnya adalah input_required hasil yang berisi inputRequests dan buramrequestState. Klien mengumpulkan input dan mencoba kembali permintaan asli dengan yang baruid, yanginputResponses, dan yang tidak dimodifikasi. requestState Sesi dan initialize jabat tangan tidak digunakan. Gateway Anda supportedVersions harus disertakan2026-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"}]}
)

Pada versi ini, elicitasi URL tiba sebagai elicitation/create permintaan pada aliran SSE terbuka dalam sesi MCP. Atur MCP-Protocol-Version header ke versi yang didukung gateway Anda.

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

Pada versi2026-07-28, elicitasi URL tiba sebagai input_required hasilnya melalui pola permintaan multi pulang-pergi alih-alih pada aliran SSE. Pengguna menyelesaikan tindakan di URL, kemudian klien mencoba kembali permintaan asli dengan yang baruid, yanginputResponses, dan yang tidak requestState dimodifikasi. Sesi tidak digunakan. Gateway Anda supportedVersions harus disertakan2026-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"}]}
)