在网关中使用激励 AgentCore

在这些版本中，客户端在期间声明了激发能力initialize，并以elicitation/create请求的形式在开放的 SSE 数据流中到达。将标MCP-Protocol-Version头设置为您的网关支持的版本。

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

在版本上2026-07-28，Elicitation 使用多往返请求模式，而不是服务器在 SSE 流上发起的请求。客户在每个请求中_meta都声明了激发能力。如果该工具需要输入，则响应是包含input_required结果inputRequests且不透明requestState。客户端收集输入并使用新的id、和未修改的重试原始请求。inputResponses requestState不使用会话和initialize握手。您的网关supportedVersions必须包括2026-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"}]}
)

在这些版本中，URL 引用在 MCP 会话内以开放的 SSE 流中的elicitation/create请求形式到达。将标MCP-Protocol-Version头设置为您的网关支持的版本。

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

在版本上2026-07-28，URL 引用是input_required通过多往返请求模式而不是在 SSE 流中获得的。用户在 URL 上完成操作，然后客户端使用新的id、未requestState修改的inputResponses重试原始请求。不使用会话。您的网关supportedVersions必须包括2026-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"}]}
)