AgentCore 게이트웨이에서 유도 사용

이러한 버전에서 클라이언트는 중 유도 기능을 선언initialize하며, 유도는 열린 SSE 스트림에 elicitation/create 요청으로 도착합니다. 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유도는 SSE 스트림에서 서버 시작 요청 대신 다중 왕복 요청 패턴을 사용합니다. 클라이언트는 각 요청에 대해 _meta의 유도 기능을 선언합니다. 도구에 입력이 필요한 경우 응답은 inputRequests 및 불투명가 포함된 input_required 결과입니다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

버전 에서는 SSE 스트림 대신 다중 왕복 요청 패턴을 통해 2026-07-28URL 유도가 input_required 결과로 도착합니다. 사용자는 URL에서 작업을 완료한 다음 클라이언트는 새 , inputResponses및 수정되지 않은 id로 원래 요청을 재시도합니다requestState. 세션은 사용되지 않습니다. 게이트웨이의 에는가 포함되어야 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"}]}
)