Искусственный интеллект
Stateful MCP client capabilities в Amazon Bedrock AgentCore Runtime теперь позволяют строить интерактивные многоходовые агентные workflows, которые раньше были невозможны в stateless-реализациях. Разработчики AI-агентов часто сталкиваются с тем, что их workflows должны прерываться в середине выполнения, чтобы уточнить данные у пользователя, запросить сгенерированный large language model (LLM) контент или показать обновления прогресса для длительных операций; stateless MCP servers с такими сценариями не справляются. Это ограничение снимается за счет трех client capabilities из спецификации MCP:
- Elicitation (запрос пользовательского ввода в середине выполнения)
- Sampling (запрос LLM-generated content от client)
- Progress notification (потоковая передача обновлений в реальном времени)
Эти возможности превращают одностороннее выполнение tools в двусторонний диалог между вашим MCP server и clients.
Model Context Protocol (MCP) — это открытый стандарт, определяющий, как LLM-приложения подключаются к внешним tools и data sources. Спецификация описывает server capabilities (tools, prompts и resources, которые server exposes) и client capabilities (возможности, которые clients возвращают server). Если предыдущий релиз AWS был посвящен размещению stateless MCP servers на AgentCore Runtime, то новая возможность завершает реализацию двустороннего протокола. Clients, подключающиеся к MCP servers, размещенным на AgentCore, теперь могут отвечать на server-initiated requests. В этом посте вы узнаете, как строить stateful MCP servers, которые запрашивают пользовательский ввод во время выполнения, вызывают LLM sampling для динамической генерации контента и отправляют потоковые обновления прогресса для длительных задач. Вы увидите примеры кода для каждой возможности и развернете рабочий stateful MCP server в Amazon Bedrock AgentCore Runtime.
От stateless к stateful MCP
Изначальная поддержка MCP server в AgentCore использовала stateless mode: каждый входящий HTTP request был независимым, без общего контекста между вызовами. Эта модель проста в развертывании и понимании и хорошо подходит для tool servers, которые принимают входные данные и возвращают результат. Однако у нее есть фундаментальное ограничение. Server не может сохранять thread разговора между запросами, просить пользователя уточнить данные в середине tool call или сообщать client о ходе работы по мере выполнения.
Stateful mode снимает это ограничение. Когда вы запускаете MCP server с stateless_http=False, AgentCore Runtime выделяет отдельный microVM для каждой пользовательской session. MicroVM сохраняется на протяжении всей жизни session (до 8 часов либо 15 минут неактивности в соответствии с параметром idleRuntimeSessionTimeout setting), при этом между sessions обеспечивается изоляция CPU, памяти и файловой системы. Протокол поддерживает непрерывность через заголовок Mcp-Session-Id: server возвращает этот идентификатор во время initialize handshake, а client включает его в каждый последующий request, чтобы маршрутизировать его обратно в ту же session.
Следующая таблица показывает ключевые различия:
| Stateless mode | Stateful mode | |
| stateless_http setting | TRUE | FALSE |
| Session isolation | Выделенный microVM на session | Выделенный microVM на session |
| Session lifetime | До 8 часов; 15-минутный idle timeout | До 8 часов; 15-минутный idle timeout |
| Client capabilities | Не поддерживаются | Elicitation, sampling, progress notifications |
| Recommended for | Простое обслуживание tools | Интерактивные многоходовые workflows |
Когда session истекает или server перезапускается, последующие requests со старым session ID возвращают 404. В этот момент clients должны заново инициализировать соединение, чтобы получить новый session ID и начать новую session. Изменение конфигурации для включения stateful mode сводится к одному флагу при запуске server:
mcp.run( transport="streamable-http", host="0.0.0.0", port=8000, stateless_http=False # Enable stateful mode)
Кроме этого флага, три client capabilities становятся доступны автоматически, когда MCP client объявляет поддержку ими во время initialization handshake.
Три новые client capabilities
Stateful mode приносит три client capabilities из спецификации MCP. Каждая из них решает отдельный сценарий взаимодействия, с которым агенты сталкиваются в production workflows.
Elicitation позволяет server приостановить выполнение и запросить у user структурированный ввод через client. Tool может задать точечные вопросы в нужный момент workflow, собрать предпочтение, подтвердить решение или получить значение, зависящее от предыдущих результатов. Server отправляет request elicitation/create с сообщением и необязательной JSON schema, описывающей ожидаемую структуру ответа. Client отображает подходящий интерфейс ввода, а user может согласиться, отказать или отменить действие.
Sampling позволяет server запросить у client LLM-generated completion через sampling/createMessage. Именно это делает возможным использование language model capabilities в server-side tool logic без хранения собственных model credentials. Server передает prompt и необязательные model preferences; client пересылает request подключенной LLM и возвращает сгенерированный ответ. Практические сценарии включают генерацию персонализированных summaries, создание текстовых объяснений структурированных данных и выдачу рекомендаций на основе предыдущего контекста разговора.
Progress notifications позволяют server сообщать о промежуточном прогрессе во время длительных операций. Используя ctx.report_progress(progress, total), server отправляет обновления, которые clients могут показывать как progress bar или status indicator. Для операций, состоящих из нескольких шагов, например поиска по нескольким data sources, это помогает держать user в курсе, а не смотреть на пустой экран.
Все три возможности включаются на уровне client: client объявляет, какие capabilities он поддерживает, во время initialization, а server должен использовать только те capabilities, которые client заявил.
Elicitation: пользовательский ввод, инициируемый server
Elicitation — это механизм, при котором MCP server приостанавливает выполнение в середине работы и просит client собрать у user конкретную информацию. Server отправляет JSON-RPC request elicitation/create, содержащий понятное человеку сообщение и requestedSchema, описывающую ожидаемый ответ. Client отображает это как форму или prompt, а ответ user (или явный отказ) возвращается server, чтобы выполнение могло продолжиться. Спецификация MCP поддерживает два режима elicitation:
- Form mode: структурированный сбор данных непосредственно через MCP client. Подходит для предпочтений, конфигурационных вводов и подтверждений, не связанных с чувствительными данными.
- URL mode: направляет user на внешний URL для взаимодействий, которые не должны проходить через MCP client, например OAuth flows, платежную обработку или ввод учетных данных.
Ответ использует модель из трех действий: accept (user предоставил данные), decline (user явно отклонил запрос) или cancel (user закрыл окно, не выбрав действие). Servers должны корректно обрабатывать каждый случай. Следующий пример реализует tool add_expense_interactive, который собирает новую расходную запись через четыре последовательных шага elicitation: сумма, описание, категория и финальное подтверждение перед записью в DynamoDB. На каждом шаге ожидаемый ввод определяется отдельной Pydantic model, которую FastMCP преобразует в JSON Schema, отправляемую в request elicitation/create.
Server
Tool add_expense_interactive проводит user через четыре последовательных вопроса перед записью в Amazon DynamoDB. На каждом шаге ожидаемый ввод задается отдельной Pydantic model, потому что schema в form mode должна быть плоским объектом. Теоретически можно собрать все четыре поля в одной model с четырьмя свойствами, но разделение здесь дает user один сфокусированный вопрос за раз — именно такой интерактивный паттерн и задумана обеспечивать elicitation.
agents/mcp_client_features.py
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 # one of: food, transport, bills, entertainment, other
class ConfirmInput(BaseModel):
confirm: str # Yes or No
@mcp.tool()
async def add_expense_interactive(user_alias: str, ctx: Context) -> str:
"""Interactively add a new expense using elicitation.
Args:
user_alias: User identifier
"""
# Step 1: Ask for the amount
result = await ctx.elicit('How much did you spend?', AmountInput)
if not isinstance(result, AcceptedElicitation):
return 'Expense entry cancelled.'
amount = result.data.amount
# Step 2: Ask for a description
result = await ctx.elicit('What was it for?', DescriptionInput)
if not isinstance(result, AcceptedElicitation):
return 'Expense entry cancelled.'
description = result.data.description
# Step 3: Select a category
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
# Step 4: Confirm before saving
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
)
Каждый вызов await ctx.elicit() приостанавливает tool и отправляет request elicitation/create через активную session. Проверка isinstance(result, AcceptedElicitation) одинаково обрабатывает decline и cancel на каждом шаге.
Client
Регистрация elicitation_handler в fastmcp.Client одновременно подключает handler и сообщает client о поддержке elicitation server во время initialization.
import asyncio
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
# Предзагруженные ответы имитируют, как user отвечает на каждый вопрос по порядку
_responses = iter([
{'amount': 45.50},
{'description': 'Lunch at the office'},
{'category': 'food'},
{'confirm': 'Yes'},
])
async def elicit_handler(message, response_type, params, context):
# В production: отрисовать форму и вернуть ввод user
response = next(_responses)
print(f' Server asks: {message}')
print(f' Responding: {response}\n')
return response
transport = StreamableHttpTransport(url=mcp_url, headers=headers)
async with Client(transport, elicitation_handler=elicit_handler) as client:
await asyncio.sleep(2) # allow session initialization
result = await client.call_tool('add_expense_interactive', {'user_alias': 'me'})
print(result.content[0].text)
При запуске на развернутом server:
Server asks: How much did you spend?
Responding: {'amount': 45.5}
Server asks: What was it for?
Responding: {'description': 'Lunch at the office'}
Server asks: Select a category (food, transport, bills, entertainment, other):
Responding: {'category': 'food'}
Server asks: Confirm: add expense of $45.50 for Lunch at the office (category: food)? Reply Yes or No
Responding: {'confirm': 'Yes'}
Expense of $45.50 added for me
Полный рабочий пример, включая настройку DynamoDB и развертывание в AgentCore, доступен в GitHub sample repository.
Используйте elicitation, когда вашему tool нужна информация, зависящая от предыдущих результатов, ее лучше собирать интерактивно, чем заранее, или она различается у разных users так, что ее невозможно параметризовать заранее. Tool для бронирования поездки, который сначала ищет направления, а затем просит user выбрать одно из них, — естественный пример. Финансовый workflow, который подтверждает сумму транзакции перед отправкой, — еще один. Elicitation не подходит для чувствительных данных, таких как пароли или API keys; для них используйте URL mode или безопасный out-of-band channel.
Sampling: LLM generation, инициируемая server
Sampling — это механизм, с помощью которого MCP server запрашивает у client LLM completion. Server отправляет request sampling/createMessage, содержащий список conversation messages, system prompt и необязательные model preferences. Client пересылает request своей подключенной language model (с учетом подтверждения user) и возвращает сгенерированный ответ. Server получает структурированный результат, включающий сгенерированный текст, использованную model и reason остановки.
Эта возможность переворачивает обычный поток: вместо того чтобы client запрашивал у server tool results, server запрашивает у client model output. Преимущество в том, что server не нужны API keys или прямая интеграция с model. Client сохраняет полный контроль над тем, какая model используется, а спецификация MCP предусматривает шаг human-in-the-loop, когда users могут просматривать и утверждать sampling requests перед отправкой.
Servers могут выражать предпочтения к model с помощью приоритетов capabilities (costPriority, speedPriority, intelligencePriority) и необязательных model hints. Это рекомендации, окончательный выбор делает client на основе того, к каким models он имеет доступ.
Server
Tool analyze_spending извлекает transactions из DynamoDB, строит prompt на основе структурированных данных и передает анализ LLM client через ctx.sample().
agents/mcp_client_features.py (добавленный tool, тот же файл, что и elicitation)
@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.
Args:
user_alias: User identifier
"""
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}'
Tool вызывает await ctx.sample() и приостанавливается. Server отправляет request sampling/createMessage client через открытый session. Когда client возвращает ответ LLM, выполнение продолжается.
Client
sampling_handler получает prompt от server и пересылает его в language model. В этом примере это Claude Haiku на Amazon. Регистрация handler — это также способ сообщить server о поддержке sampling во время initialization.
import json
import asyncio
import boto3
from mcp.types import CreateMessageResult, TextContent
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
MODEL_ID = 'us.anthropic.claude-haiku-4-5-20251001-v1:0'
bedrock = boto3.client('bedrock-runtime', region_name=region)
def _invoke_bedrock(prompt: str, max_tokens: int) -> str:
body = json.dumps({
'anthropic_version': 'bedrock-2023-05-31',
'max_tokens': max_tokens,
'messages': [{'role': 'user', 'content': prompt}]
})
resp = bedrock.invoke_model(modelId=MODEL_ID, body=body)
return json.loads(resp['body'].read())['content'][0]['text']
async def sampling_handler(messages, params, ctx):
"""Called by fastmcp.Client when the server issues ctx.sample()."""
prompt = messages if isinstance(messages, str) else ' '.join(
m.content.text for m in messages if hasattr(m.content, 'text')
)
max_tokens = params.maxTokens if params and hasattr(params, 'maxTokens') and params.maxTokens else 300
text = await asyncio.to_thread(_invoke_bedrock, prompt, max_tokens)
return CreateMessageResult(
role='assistant',
content=TextContent(type='text', text=text),
model=MODEL_ID,
stopReason='endTurn'
)
transport = StreamableHttpTransport(url=mcp_url, headers=headers)
async with Client(transport, sampling_handler=sampling_handler) as client:
result = await client.call_tool('analyze_spending', {'user_alias': 'me'})
print(result.content[0].text)
При запуске для user с четырьмя заранее добавленными расходами:
Spending Analysis for me:
Total Spending: $266.79
Breakdown:
- Food: $130.80 (49%)
- Bills: $120.00 (45%)
- Entertainment: $15.99 (6%)
3 Actionable Recommendations:
1. Meal prep at home — cook groceries into multiple meals to reduce restaurant
spending and lower food costs by 20-30%.
2. Review entertainment subscriptions — audit all subscriptions and cancel
unused services or share family plans.
3. Reduce energy costs — use programmable thermostats, LED bulbs, and unplug
devices to lower electricity bills by 10-15% monthly.
Используйте sampling, когда ваш tool должен генерировать natural-language output, в котором выигрывает от возможностей language model. Хороший пример — tool, который собрал travel preferences пользователя и хочет сгенерировать персонализированный narrative маршрута поездки. Sampling не подходит для детерминированных операций вроде database queries, вычислений или API calls с четко определенным output. Для таких задач мы рекомендуем использовать tool logic.
Progress notifications: обратная связь о выполнении в реальном времени
Progress notifications — это events, которые server отправляет во время длительных операций, чтобы держать client и user в курсе того, какая часть работы уже выполнена. await ctx.report_progress(progress, total) отправляет message notifications/progress и сразу возвращает управление. Server не ждет ответа — это fire-and-forget в обе стороны. Client получает notification асинхронно и может отрисовать progress bar, вывести строку статуса или предотвратить ощущение, что соединение зависло. Паттерн заключается в том, чтобы вызывать report_progress на каждом логическом шаге многоэтапной операции, увеличивая progress до total.
Server
Tool generate_report строит ежемесячный финансовый report в пять шагов, отправляя progress notification в начале каждого из них.
agents/mcp_progress_server.py
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.
Args:
user_alias: User identifier
"""
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
)
Каждый await ctx.report_progress() — это fire-and-forget: notification отправляется, и выполнение немедленно переходит к следующему шагу.
Client
progress_handler получает progress, total и необязательный message каждый раз, когда server отправляет notification. Регистрация handler — это способ сообщить client о поддержке progress во время initialization.
import logging
logging.getLogger('mcp.client.streamable_http').setLevel(logging.ERROR)
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
async def progress_handler(progress: float, total: float | None, message: str | None):
pct = int((progress / total) * 100) if total else 0
filled = pct // 5
bar = '#' * filled + '-' * (20 - filled)
print(f'\r Progress: [{bar}] {pct}% ({int(progress)}/{int(total or 0)})',
end='', flush=True)
if total and progress >= total:
print(' Done!')
transport = StreamableHttpTransport(url=mcp_url, headers=headers)
async with Client(transport, progress_handler=progress_handler) as client:
result = await client.call_tool('generate_report', {'user_alias': 'me'})
print(result.content[0].text)
По мере прохождения server через пять этапов client рисует индикатор на месте:
Progress: [####----------------] 20% (1/5)
Progress: [########------------] 40% (2/5)
Progress: [############--------] 60% (3/5)
Progress: [################----] 80% (4/5)
Progress: [####################] 100% (5/5) Done!
Используйте progress notifications для любого tool call, который выполняется дольше нескольких секунд и состоит из дискретных, измеримых шагов. Операции вроде поиска по нескольким data sources, последовательности API calls, обработки batch records или многошагового booking workflow — хорошие кандидаты. Tool, который завершается менее чем за секунду, обычно не нуждается в progress reporting; накладные расходы на отправку events для быстрых операций не оправданы.
Заключение
В этом посте вы познакомились со stateful MCP client capabilities в Amazon Bedrock AgentCore Runtime. Мы показали разницу между stateless и stateful MCP deployments, разобрали elicitation, sampling и progress notifications на примерах кода и продемонстрировали, как развернуть stateful MCP server в AgentCore Runtime. С этими возможностями вы можете строить MCP servers, которые ведут structured conversations с пользователями, используют LLM client для генерации контента и дают real-time visibility в длительные операции, при этом все работает на managed, isolated infrastructure, powered by AgentCore Runtime. Предлагаем изучить следующие ресурсы, чтобы начать работу:
- GitHub sample code.
- Документация Amazon Bedrock AgentCore Runtime
- Документация по stateful MCP features
- MCP specification 2025-11-25
- Предыдущий пост: Hosting MCP servers on AgentCore Runtime
Материал — перевод статьи с английского.
Оригинал: Introducing stateful MCP client capabilities on Amazon Bedrock AgentCore Runtime