Utilisez l'échantillonnage avec votre AgentCore passerelle

Sur ces versions, le client déclare la capacité d'échantillonnage pendantinitialize, et la demande d'échantillonnage arrive en tant que sampling/createMessage 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 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

Dans la version2026-07-28, l'échantillonnage utilise le modèle de demandes aller-retour multiples au lieu d'une demande initiée par le serveur sur le flux SSE. Le client déclare la capacité d'échantillonnage _meta à chaque demande. Si l'outil doit être complété, la réponse est un input_required résultat contenant une sampling/createMessage requête inputRequests et un opaquerequestState. Le client invoque son modèle et tente à nouveau la demande d'origine avec un nouveau idinputResponses, le et le non modifié. 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": {"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"
))