Interatividade Bidimensional em Agentes de IA
A construção de agentes de IA oferece desafios significativos quando os fluxos de trabalho precisam pausar durante a execução para solicitar esclarecimentos ao usuário, gerar conteúdo via modelo de linguagem ou fornecer atualizações de progresso em operações de longa duração. Até recentemente, servidores MCP sem estado não conseguiam lidar com esses cenários, criando uma limitação fundamental para aplicações reais.
A AWS identificou essa lacuna e anunciou novidades importantes. O Amazon Bedrock AgentCore Runtime agora oferece suporte a três capacidades do cliente definidas na especificação MCP (Protocolo de Contexto de Modelo):
- Elicitação: solicitação de entrada do usuário durante a execução
- Amostragem: solicitação de conteúdo gerado por LLM do cliente
- Notificações de progresso: transmissão de atualizações em tempo real
Essas capacidades transformam a execução de ferramentas em sentido único em conversas bidirecionais entre o servidor MCP e os clientes.
Do Stateless ao Stateful: Uma Evolução Necessária
O Modelo Anterior: Stateless MCP
A implementação anterior do suporte MCP no AgentCore funcionava em modo sem estado: cada requisição HTTP era independente, sem contexto compartilhado entre chamadas. Esse modelo é simples de implementar e funciona bem para servidores de ferramentas que recebem entradas e retornam saídas.
Porém, possui uma restrição fundamental. O servidor não consegue manter uma conversa entre requisições, solicitar esclarecimentos do usuário durante a execução de uma ferramenta ou reportar progresso conforme o trabalho ocorre.
O Novo Paradigma: Stateful MCP
O modo stateful remove essas limitações. Quando você executa seu servidor MCP com stateless_http=False, o AgentCore Runtime provisiona uma máquina virtual dedicada para cada sessão de usuário. Essa máquina virtual persiste pela duração da sessão (até 8 horas, ou 15 minutos de inatividade conforme a configuração idleRuntimeSessionTimeout), com isolamento de CPU, memória e sistema de arquivos entre sessões.
O protocolo mantém a continuidade através de um header Mcp-Session-Id: o servidor retorna este identificador durante o handshake de inicialização, e o cliente o inclui em toda requisição subsequente para rotear de volta à mesma sessão.
A mudança de configuração para ativar o modo stateful é um único parâmetro na inicialização do servidor:
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
stateless_http=False # Ativa modo stateful
)Além desse parâmetro, as três capacidades do cliente ficam disponíveis automaticamente quando o cliente MCP declara suporte durante o handshake de inicialização.
Três Capacidades Essenciais para Workflows Produtivos
Elicitação: Coletando Entrada do Usuário em Tempo Real
A elicitação permite que um servidor pause a execução e solicite entrada estruturada do usuário através do cliente. A ferramenta consegue fazer perguntas direcionadas no momento certo do fluxo, coletando preferências, confirmando decisões ou reunindo valores que dependem de resultados anteriores.
O servidor envia uma requisição elicitation/create com uma mensagem e um esquema JSON opcional descrevendo a estrutura de resposta esperada. O cliente renderiza uma interface de entrada apropriada, e o usuário pode aceitar (fornecendo os dados), declinar ou cancelar.
Um exemplo prático é a ferramenta add_expense_interactive, que coleta uma despesa através de quatro etapas sequenciais de elicitação: valor, descrição, categoria e confirmação final antes de escrever no Amazon DynamoDB.
import os
from pydantic import BaseModel
from fastmcp import FastMCP, Context
from fastmcp.server.elicitation import AcceptedElicitation
from dynamo_utils import FinanceDB
mcp = FastMCP(name='ElicitationMCP')
_region = os.environ.get('AWS_REGION') or os.environ.get('AWS_DEFAULT_REGION') or 'us-east-1'
db = FinanceDB(region_name=_region)
class AmountInput(BaseModel):
amount: float
class DescriptionInput(BaseModel):
description: str
class CategoryInput(BaseModel):
category: str
class ConfirmInput(BaseModel):
confirm: str
@mcp.tool()
async def add_expense_interactive(user_alias: str, ctx: Context) -> str:
"""Interactively add a new expense using elicitation."""
result = await ctx.elicit('How much did you spend?', AmountInput)
if not isinstance(result, AcceptedElicitation):
return 'Expense entry cancelled.'
amount = result.data.amount
result = await ctx.elicit('What was it for?', DescriptionInput)
if not isinstance(result, AcceptedElicitation):
return 'Expense entry cancelled.'
description = result.data.description
result = await ctx.elicit(
'Select a category (food, transport, bills, entertainment, other):',
CategoryInput
)
if not isinstance(result, AcceptedElicitation):
return 'Expense entry cancelled.'
category = result.data.category
confirm_msg = (
f'Confirm: add expense of ${amount:.2f} for {description}'
f' (category: {category})? Reply Yes or No'
)
result = await ctx.elicit(confirm_msg, ConfirmInput)
if not isinstance(result, AcceptedElicitation) or result.data.confirm != 'Yes':
return 'Expense entry cancelled.'
return db.add_transaction(user_alias, 'expense', -abs(amount), description, category)
if __name__ == '__main__':
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
stateless_http=False
)No cliente, registrar um elicitation_handler tanto ativa o handler quanto declara suporte ao servidor durante a inicialização. Cada await ctx.elicit() suspende a ferramenta e envia uma requisição elicitation/create sobre a sessão ativa. A verificação isinstance(result, AcceptedElicitation) trata declínio e cancelamento uniformemente em cada etapa.
Use elicitação quando sua ferramenta precisa de informações que dependem de resultados anteriores, são melhor coletadas interativamente do que antecipadamente, ou variam entre usuários de forma que não pode ser parametrizada de antemão.
Amostragem: Delegando Geração de Conteúdo ao LLM do Cliente
A amostragem permite que um servidor MCP solicite uma conclusão gerada por LLM do cliente. O servidor envia uma requisição sampling/createMessage contendo uma lista de mensagens de conversa, um prompt do sistema e preferências opcionais de modelo. O cliente encaminha a requisição para seu modelo de linguagem conectado e retorna a resposta gerada.
O grande diferencial é que o servidor não precisa de chaves de API ou integração direta com modelo. O cliente retém controle total sobre qual modelo é usado, e a especificação MCP prevê uma etapa de verificação humana onde usuários podem revisar e aprovar requisições de amostragem antes do envio.
Um exemplo prático é a ferramenta analyze_spending, que recupera transações do DynamoDB, constrói um prompt com os dados estruturados e delega a análise ao LLM do cliente:
@mcp.tool()
async def analyze_spending(user_alias: str, ctx: Context) -> str:
"""Fetch expenses from DynamoDB and ask the client's LLM to analyse them."""
transactions = db.get_transactions(user_alias)
if not transactions:
return f'No transactions found for {user_alias}.'
lines = '\n'.join(
f"- {t['description']} (${abs(float(t['amount'])):.2f}, {t['category']})"
for t in transactions
)
prompt = (
f'Here are the recent expenses for a user:\n{lines}\n\n'
f'Please analyse the spending patterns and give 3 concise, '
f'actionable recommendations to improve their finances. '
f'Keep the response under 120 words.'
)
ai_analysis = 'Analysis unavailable.'
try:
response = await ctx.sample(messages=prompt, max_tokens=300)
if hasattr(response, 'text') and response.text:
ai_analysis = response.text
except Exception:
pass
return f'Spending Analysis for {user_alias}:\n\n{ai_analysis}'No cliente, o sampling_handler recebe o prompt do servidor e o encaminha a um modelo de linguagem. Registrar o handler também é como o cliente declara suporte de amostragem ao servidor durante inicialização.
Use amostragem quando sua ferramenta deve produzir saída em linguagem natural que se beneficia das capacidades de um modelo de linguagem. Um exemplo: ferramenta que coletou preferências de viagem do usuário e quer gerar uma narrativa de itinerário personalizada. Amostragem não é apropriada para operações determinísticas como consultas a banco de dados ou cálculos com saídas bem definidas — use lógica de ferramenta para esses casos.
Notificações de Progresso: Visibilidade em Tempo Real
As notificações de progresso são eventos que um servidor emite durante operações de longa duração para manter o cliente e o usuário informados sobre quanto trabalho foi concluído. await ctx.report_progress(progress, total) emite uma mensagem notifications/progress e retorna imediatamente. O servidor não espera por resposta — é fire-and-forget em ambas as direções.
O padrão é chamar report_progress em cada etapa lógica de uma operação multi-estágio, com o progresso incrementando em direção ao total. A ferramenta generate_report constrói um relatório financeiro mensal em cinco etapas, emitindo notificação de progresso no início de cada uma:
import os
from fastmcp import FastMCP, Context
from dynamo_utils import FinanceDB
mcp = FastMCP(name='Progress-MCP-Server')
_region = os.environ.get('AWS_REGION') or os.environ.get('AWS_DEFAULT_REGION') or 'us-east-1'
db = FinanceDB(region_name=_region)
@mcp.tool()
async def generate_report(user_alias: str, ctx: Context) -> str:
"""Generate a monthly financial report, streaming progress at each stage."""
total = 5
# Step 1: Fetch transactions
await ctx.report_progress(progress=1, total=total)
transactions = db.get_transactions(user_alias)
# Step 2: Group by category
await ctx.report_progress(progress=2, total=total)
by_category = {}
for t in transactions:
cat = t['category']
by_category[cat] = by_category.get(cat, 0) + abs(float(t['amount']))
# Step 3: Fetch budgets
await ctx.report_progress(progress=3, total=total)
budgets = {b['category']: float(b['monthly_limit']) for b in db.get_budgets(user_alias)}
# Step 4: Compare spending vs budgets
await ctx.report_progress(progress=4, total=total)
lines = []
for cat, spent in sorted(by_category.items(), key=lambda x: -x[1]):
limit = budgets.get(cat)
if limit:
pct = (spent / limit) * 100
status = 'OVER' if spent > limit else 'OK'
lines.append(f' {cat:<15} ${spent:>8.2f} / ${limit:.2f} [{pct:.0f}%] {status}')
else:
lines.append(f' {cat:<15} ${spent:>8.2f} (no budget set)')
# Step 5: Format and return
await ctx.report_progress(progress=5, total=total)
total_spent = sum(by_category.values())
return (
f'Monthly Report for {user_alias}\n'
f'{"=" * 50}\n'
f' {"Category":<15} {"Spent":>10} {"Budget":>8} Status\n'
f'{"-" * 50}\n'
+ '\n'.join(lines)
+ f'\n{"-" * 50}\n'
f' {"TOTAL":<15} ${total_spent:>8.2f}\n'
)
if __name__ == '__main__':
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
stateless_http=False
)No cliente, o progress_handler recebe progresso, total e uma mensagem opcional cada vez que o servidor emite notificação. O cliente consegue renderizar uma barra de progresso ou indicador de status, mantendo o usuário informado em vez de observar uma tela em branco.
Use notificações de progresso para qualquer chamada de ferramenta que leva mais alguns segundos e envolve etapas discretas mensuráveis. Operações como buscar em múltiplas fontes de dados, executar sequências de chamadas de API, processar lotes de registros ou executar fluxos de reserva multi-etapa são bons candidatos. Uma ferramenta que completa em menos de um segundo geralmente não precisa de relatório de progresso.
Primeiros Passos na Prática
Para começar a explorar essas capacidades, a AWS disponibiliza recursos técnicos de referência. O código de exemplo no GitHub inclui configuração completa do DynamoDB e deployment no AgentCore. A documentação do Amazon Bedrock AgentCore Runtime oferece guias detalhados, enquanto a documentação de funcionalidades stateful do MCP cobre especificações técnicas. Também recomenda-se consultar a especificação completa do MCP para entender o protocolo subjacente.
A implementação dessas capacidades representa evolução significativa na construção de agentes de IA interativos, oferecendo infraestrutura gerenciada e isolada para workflows bidirecionais.
Fonte
Introducing stateful MCP client capabilities on Amazon Bedrock AgentCore Runtime (https://aws.amazon.com/blogs/machine-learning/introducing-stateful-mcp-client-capabilities-on-amazon-bedrock-agentcore-runtime/)
Leave a Reply