View a markdown version of this page

Ative o streaming de resposta MCP para seu gateway AgentCore - Amazon Bedrock AgentCore
Benefícios do streaming de respostasAtivar streaming de respostasComo funciona o streaming de respostasRequisitos do clienteExemplos de código

Ative o streaming de resposta MCP para seu gateway AgentCore

O streaming de resposta do MCP permite que seu AgentCore gateway forneça Server-Sent eventos em tempo real (SSE) aos clientes durante a execução da ferramenta. Em vez de esperar que toda a chamada da ferramenta seja concluída antes de retornar uma resposta, o gateway transmite os eventos à medida que eles ocorrem, incluindo notificações de progresso, mensagens de log, solicitações de elicitação e solicitações de amostragem.

Benefícios do streaming de respostas

Real-time feedback

Os clientes recebem atualizações de progresso e mensagens de registro à medida que elas acontecem, em vez de esperar pela resposta completa da ferramenta.

Permite recursos MCP interativos

O streaming de resposta é um pré-requisito para elicitação, amostragem, notificações de progresso e registro de mensagens. Esses recursos exigem uma conexão SSE aberta para fornecer eventos iniciados pelo servidor durante a execução da ferramenta.

Melhor experiência do usuário para ferramentas de longa duração

Para ferramentas que levam segundos ou minutos para serem concluídas, o streaming mantém o cliente informado e responsivo.

Ativar streaming de respostas

Para ativar o streaming de respostas, streamingConfiguration.enableResponseStreaming defina como true no protocolConfiguration.mcp campo ao criar ou atualizar seu gateway:

{
  "protocolConfiguration": {
    "mcp": {
      "streamingConfiguration": {
        "enableResponseStreaming": true
      }
    }
  }
}
nota

A ativação do streaming de resposta introduz uma alteração no contrato de entrada do interceptor de resposta. Se você usa interceptores de resposta, revise a lógica do interceptor para garantir a compatibilidade com as respostas de streaming. Consulte Interceptores de resposta com streaming ativado para obter detalhes.

Como funciona o streaming de respostas

Quando o streaming de resposta está ativado e o cliente envia uma solicitaçãoAccept: text/event-stream, o gateway retorna um fluxo SSE em vez de uma única resposta JSON. Os eventos são entregues à medida que são recebidos do servidor MCP de destino.

O stream SSE pode incluir os seguintes tipos de eventos:

Tipo de evento Description

notifications/progress

Atualizações de progresso do alvo durante a execução da ferramenta. Consulte Receber notificações de progresso.

notifications/message

Registre mensagens do alvo. Consulte Receber mensagens de registro.

elicitation/create

Solicitações de elicitação do alvo solicitando a entrada do usuário. Consulte Usar elicitação.

sampling/createMessage

Solicitações de amostragem do alvo solicitando a conclusão do LLM. Consulte Usar amostragem.

Resultado final

O resultado da chamada da ferramenta, entregue como o último evento antes do encerramento da transmissão.

Se o cliente não enviarAccept: text/event-stream, o gateway armazena a resposta em buffer e retorna uma única resposta JSON após a conclusão da chamada da ferramenta. Eventos intermediários (progresso, registro) não são entregues nesse caso.

Requisitos do cliente

Para receber respostas de streaming, os clientes devem:

  • Envie o Accept: text/event-stream cabeçalho em suas solicitações.

  • Gerencie os eventos SSE à medida que eles chegam, analisando cada data: linha como uma JSON-RPC mensagem.

  • Mantenha a conexão aberta até que o evento do resultado final seja recebido.

Exemplos de código

exemplo
curl

  1. Envie uma solicitação de chamada de ferramenta com o cabeçalho SSE accept:

    curl -N -X POST \
  https://mygateway-abcdefghij.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -H "Authorization: Bearer YOUR_ACCESS_TOKEN" \
  -d '{
    "jsonrpc": "2.0",
    "id": "tool-call-1",
    "method": "tools/call",
    "params": {
      "name": "analyzeDataset",
      "arguments": {
        "datasetId": "ds-12345"
      }
    }
}'

    Exemplo de resposta de fluxo SSE:

    event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"auto-1","progress":1,"total":3,"message":"Loading data..."}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/message","params":{"level":"info","logger":"analyzer","data":"Processing 10,000 records"}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"auto-1","progress":2,"total":3,"message":"Analyzing..."}}

event: message
data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"auto-1","progress":3,"total":3,"message":"Complete"}}

event: message
data: {"jsonrpc":"2.0","id":"tool-call-1","result":{"content":[{"type":"text","text":"Analysis complete. Found 3 anomalies."}]}}
Python requests package

  1. 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"
}

response = requests.post(gateway_url, headers=headers, json={
    "jsonrpc": "2.0",
    "id": "tool-call-1",
    "method": "tools/call",
    "params": {
        "name": "analyzeDataset",
        "arguments": {"datasetId": "ds-12345"}
    }
}, stream=True)

client = sseclient.SSEClient(response)
for event in client.events():
    data = json.loads(event.data)
    method = data.get("method")

    if method == "notifications/progress":
        params = data["params"]
        print(f"Progress: {params['progress']}/{params.get('total', '?')} - {params.get('message', '')}")
    elif method == "notifications/message":
        params = data["params"]
        print(f"[{params['level'].upper()}] {params['data']}")
    elif "result" in data:
        print(f"Final result: {data['result']}")
        break
MCP Client

  1. from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
import asyncio

async def use_streaming(url, token):
    headers = {"Authorization": f"Bearer {token}"}

    # The MCP client uses streamable HTTP transport which handles SSE automatically
    async with streamablehttp_client(url=url, headers=headers) as (
        read_stream, write_stream, _
    ):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()

            # Tool calls automatically receive streaming events
            result = await session.call_tool(
                name="analyzeDataset",
                arguments={"datasetId": "ds-12345"}
            )
            print(f"Tool result: {result}")
            return result

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