AgentCore 게이트웨이에서 샘플링 사용

이러한 버전에서 클라이언트는 중에 샘플링 기능을 선언initialize하고 샘플링 요청은 열린 SSE 스트림에 sampling/createMessage 요청으로 도착합니다. 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 sampling capability
init_response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "init-request",
    "method": "initialize",
    "params": {
        "protocolVersion": "2025-06-18",
        "capabilities": {"sampling": {}},
        "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": "summarizeDocument",
        "arguments": {"documentId": "doc-789"}
    }
}, 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") == "sampling/createMessage":
        sampling_id = data["id"]
        print(f"Sampling request: {data['params']['messages']}")

        # Step 4: Invoke your LLM and send result
        llm_result = invoke_your_model(data["params"])  # Your LLM invocation
        requests.post(gateway_url, headers=headers, json={
            "jsonrpc": "2.0",
            "id": sampling_id,
            "result": {
                "model": "claude-sonnet-4-20250514",
                "role": "assistant",
                "content": {"type": "text", "text": llm_result}
            }
        })
    elif "result" in data:
        print(f"Tool result: {data['result']}")
        break

버전에서 2026-07-28샘플링은 SSE 스트림에서 서버 시작 요청 대신 다중 왕복 요청 패턴을 사용합니다. 클라이언트는 각 요청에 대해 _meta의 샘플링 기능을 선언합니다. 도구에 완료가 필요한 경우 응답은의 sampling/createMessage 요청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": {"sampling": {}}
}
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": "summarizeDocument"
}
arguments = {"documentId": "doc-789"}

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

result = response["result"]
if result.get("resultType") == "input_required":
    # Step 2: Fulfill each sampling request by invoking your model
    input_responses = {}
    for key, input_request in result.get("inputRequests", {}).items():
        params = input_request["params"]
        print(f"Sampling request: {params['messages']}")
        llm_result = invoke_your_model(params)  # Your LLM invocation
        input_responses[key] = {
            "model": "claude-sonnet-4-20250514",
            "role": "assistant",
            "content": {"type": "text", "text": llm_result}
        }

    # Step 3: Retry the tool call with a new id, the input responses,
    # and the requestState echoed back unmodified
    retry_params = {"name": "summarizeDocument", "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 sampling_handler(request):
    """Handle sampling requests from the server by invoking an LLM."""
    messages = request.params.messages
    llm_response = await invoke_your_model(messages, max_tokens=request.params.maxTokens)
    return {
        "model": "claude-sonnet-4-20250514",
        "role": "assistant",
        "content": {"type": "text", "text": llm_response}
    }

async def use_sampling(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,
            sampling_handler=sampling_handler
        ) as session:
            await session.initialize()
            result = await session.call_tool(
                name="summarizeDocument",
                arguments={"documentId": "doc-789"}
            )
            print(f"Tool result: {result}")
            return result

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