Amazon Bedrock научил MCP‑агентов вести диалог и показывать прогресс. Что это меняет для разработчиков — VogueTech

Что нового

Amazon добавил в AgentCore Runtime для Bedrock полноценную поддержку stateful MCP‑клиента. Это не просто ещё один флаг в конфиге, а расширение протокола взаимодействия между агентом и внешними инструментами.

Ключевые изменения:

Переход от stateless к stateful‑режиму для MCP‑серверов
- Раньше каждый HTTP‑запрос к MCP‑серверу жил сам по себе, без общего контекста.
- Теперь AgentCore поднимает выделенный microVM на каждую сессию пользователя.
- Время жизни сессии — до 8 часов, таймаут бездействия — 15 минут (настраивается через idleRuntimeSessionTimeout).
- Идентификатор сессии передаётся в заголовке Mcp-Session-Id: сервер выдаёт его при инициализации, клиент добавляет его во все последующие запросы.
Три новые клиентские возможности по спецификации MCP:
- Elicitation — сервер останавливает выполнение инструмента и запрашивает у пользователя данные через клиента.
  Пример: уточнить сумму траты или категорию, подтвердить действие.
- Sampling — сервер просит клиента вызвать LLM и сгенерировать текст.
  Сервер не хранит ключи к моделям, всё управление моделями остаётся на стороне клиента.
- Progress notification — сервер может стримить прогресс долгой операции через ctx.report_progress(progress, total).
Минимальная конфигурация для включения stateful‑режима:

mcp.run(
    transport="streamable-http",
    host="0.0.0.0",
    port=8000,
    stateless_http=False  # Enable stateful mode
)

После этого, если MCP‑клиент при инициализации объявляет поддержку elicitation, sampling и progress, сервер может использовать эти возможности без дополнительной настройки протокола.

Изоляция и безопасность:
- Каждый пользовательский сеанс работает в отдельном microVM.
- Изолированы CPU, память и файловая система между сессиями.
- При завершении сессии или рестарте сервера запросы со старым Mcp-Session-Id получают 404, клиент обязан переинициализировать соединение.

Как это работает

Stateful против stateless MCP на AgentCore

Раньше AgentCore поддерживал только stateless MCP:
каждый HTTP‑запрос — отдельный вызов инструмента без памяти о прошлом.

Теперь при запуске с stateless_http=False AgentCore:

При первом запросе инициализации:
- создаёт dedicated microVM для сессии,
- возвращает клиенту Mcp-Session-Id.
Для всех последующих запросов с этим Mcp-Session-Id:
- маршрутизирует их в тот же microVM,
- поддерживает единый контекст выполнения и открытые асинхронные операции.
Клиент при инициализации объявляет, какие client capabilities он поддерживает: elicitation, sampling, progress notifications.
Сервер обязан использовать только те возможности, которые клиент явно указал.

1. Elicitation: сервер сам спрашивает пользователя

Что происходит под капотом:

Сервер в ходе выполнения инструмента вызывает await ctx.elicit(message, SchemaOrModel).
AgentCore отправляет в MCP‑клиент JSON‑RPC запрос elicitation/create c:
- человекочитаемым сообщением,
- requestedSchema — JSON Schema ожидаемого ответа.
Клиент строит интерфейс (форма, промпт), пользователь вводит данные или отказывается.
Клиент возвращает один из вариантов:
- accept — данные введены,
- decline — пользователь явно отказался,
- cancel — пользователь закрыл запрос без выбора.
Сервер получает результат и продолжает выполнение или завершает инструмент.

Спецификация MCP поддерживает два режима:

Form mode — данные проходят через MCP‑клиента в виде структуры (JSON). Удобно для сумм, категорий, настроек.
URL mode — клиент открывает внешний URL (OAuth, платежи, ввод чувствительных данных), сами данные не проходят через MCP.

Пример: пошаговое добавление расхода в DynamoDB

Серверный код (упрощённый, но полный по логике) использует FastMCP и Pydantic‑модели для пошагового диалога:

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()

приостанавливает выполнение инструмента,
отправляет elicitation/create через активную сессию,
возобновляет выполнение после ответа клиента.

Клиент объявляет поддержку elicitation, регистрируя обработчик:

import asyncio
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport

# Pre-loaded responses simulate the user answering each question in sequence
_responses = iter([
    {'amount': 45.50},
    {'description': 'Lunch at the office'},
    {'category': 'food'},
    {'confirm': 'Yes'},
])

async def elicit_handler(message, response_type, params, context):
    # In production: render a form and return the user's input
    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 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

2. Sampling: сервер просит LLM‑ответ у клиента

Механика:

Сервер вызывает await ctx.sample(...).
AgentCore отправляет в MCP‑клиент запрос sampling/createMessage с:
- списком сообщений диалога или строкой‑промптом,
- системным промптом (опционально),
- параметрами генерации (например, max_tokens),
- пожеланиями по модели: costPriority, speedPriority, intelligencePriority, hints.
Клиент решает, какую LLM использовать, и вызывает её.
Клиент возвращает CreateMessageResult с текстом, именем модели и причиной остановки (stopReason).

Серверу не нужны ключи к LLM и прямые интеграции. Клиент контролирует, какие модели доступны и какие ограничения действуют.

Пример: анализ расходов через LLM на стороне клиента

Серверный инструмент analyze_spending достаёт транзакции из DynamoDB и просит клиента сгенерировать анализ:

@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\n{lines}\n\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\n{ai_analysis}'

Клиент реализует sampling_handler и прокидывает запрос в Amazon Bedrock с Claude Haiku:

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)

Пример вывода для пользователя с четырьмя расходами:

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.

3. Progress notifications: стрим прогресса долгих задач

Как устроено:

Сервер вызывает await ctx.report_progress(progress, total) на каждом шаге.
AgentCore отправляет клиенту событие notifications/progress.
Вызов не ждёт ответа — это fire‑and‑forget.
Клиент асинхронно получает обновления и может:
- рисовать progress bar,
- логировать шаги,
- показывать пользователю, что задача живёт, а не зависла.

Пример: генерация месячного отчёта по финансам

Серверный код, который делит отчёт на пять шагов и шлёт прогресс на каждом:

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\n'
        f'{"=" * 50}\n\n'
        f' {"Category":<15} {"Spent":>10} {"Budget":>8} Status\n\n'
        f'{"-" * 50}\n\n'
        + '\n'.join(lines)
        + f"\n\nTotal spent: ${total_spent:.2f}"
    )

Клиент, который умеет слушать notifications/progress, может показывать пользователю понятное движение от 1/5 до 5/5.

Что это значит для вас

Когда стоит использовать stateful MCP на AgentCore

Интерактивные, многошаговые сценарии
Если вы строите агента, который:
- уточняет детали по ходу выполнения (сумма платежа, выбор варианта, подтверждение),
- не может собрать все параметры заранее,
- должен учитывать результаты предыдущих шагов — stateful MCP с elicitation сильно упрощает архитектуру.
  Примеры:
- подбор путешествия с выбором направления после поиска,
- финансовые операции с обязательным подтверждением перед записью в базу.
Когда серверу нужен LLM, но вы не хотите светить ключи
Если вы пишете MCP‑сервер, который разворачивается у клиентов, и не хотите:
- вшивать в него ключи к GPT‑4, Claude 3 или другим моделям,
- брать на себя биллинг за запросы к моделям, используйте sampling. Сервер просто запрашивает текст, а клиент решает, какую модель вызвать и кто за неё платит.
Долгие операции, которые нельзя оставлять «в тишине»
Поиск по нескольким источникам, сложные аналитические отчёты, бэкенд‑процессы — всё, что занимает секунды и минуты.
Прогресс‑нотификации помогают:
- снизить количество повторных кликов,
- уменьшить ощущение «подвисания» интерфейса,
- сделать работу агента предсказуемее.

Где лучше не использовать

Простые, быстрые инструменты
Если ваш MCP‑сервер — это по сути обёртка над одной функцией «вход → выход» (например, запрос к API или простая БД‑операция), stateful‑режим и дополнительные capabilities могут быть избыточны. Статический HTTP‑сервер проще и дешевле в обслуживании.
Сбор чувствительных данных
Пароли, токены, платёжные реквизиты лучше не передавать через elicitation в form‑режиме.
Для этого используйте URL mode или отдельные защищённые каналы.
Детерминированные операции
Для чётких вычислений, SQL‑запросов, вызовов внешних API с жёстко определённым ответом логичнее писать обычную логику инструмента.
Sampling нужен там, где важен текст: объяснения, резюме, рекомендации.

Доступность и ограничения

Все описанные возможности завязаны на Amazon Bedrock AgentCore Runtime и Model Context Protocol (MCP).
Для работы с примером с sampling в коде используется Claude Haiku на Amazon Bedrock с идентификатором us.anthropic.claude-haiku-4-5-20251001-v1:0.
Если у вас нет доступа к Bedrock из вашего региона или из‑за ограничений по стране, эти возможности придётся поднимать через аккаунт в поддерживаемом регионе или через корпоративную инфраструктуру. Без доступа к Bedrock и соответствующим AWS‑сервисам код из примеров работать не будет.

Место на рынке

MCP и агенты на фоне классических LLM‑интеграций

Большинство агентных фреймворков строят интеграции с внешними сервисами через:

прямые HTTP‑вызовы из кода агента,
проприетарные протоколы инструментов (tools/functions) у конкретного вендора LLM.

MCP идёт другим путём: это открытый протокол для связи LLM‑приложений с внешними инструментами и данными.
Важно, что теперь в AgentCore поддержаны обе стороны протокола:

Server capabilities — инструменты, промпты, ресурсы, которые сервер отдаёт клиенту (это было раньше),
Client capabilities — elicitation, sampling, progress, которые клиент предоставляет серверу (добавлено сейчас).

На фоне собственных протоколов инструментов у OpenAI или Anthropic это даёт:

более чёткое разделение ролей: сервер не знает о конкретной LLM, клиент не знает о деталях реализации инструмента;
независимость от конкретного поставщика моделей: MCP‑клиент может работать с Bedrock, GPT‑4o, локальными моделями — серверу всё равно;
возможность переносить MCP‑серверы между разными стеками, если платформа поддерживает MCP.

Позиция Amazon Bedrock AgentCore

Если сравнивать с классическим «агентом внутри одного LLM‑провайдера», у AgentCore с stateful MCP есть несколько чётких акцентов:

Сессионная изоляция через microVM. Многие фреймворки ограничиваются in‑process состоянием или контейнерами. Здесь каждая сессия живёт в отдельной виртуальной машине с изоляцией CPU, памяти и файловой системы.
Полная реализация двустороннего MCP‑протокола. Ранее AgentCore умел только хостить stateless MCP‑серверы, теперь сервер может инициировать запросы к клиенту.

Прямых численных сравнений по скорости с альтернативами (например, self‑hosted агенты на базе LangChain или OpenAI Agents) здесь нет, но архитектурно:

stateful MCP даёт предсказуемое время жизни сессии (до 8 часов),
изоляция microVM добавляет безопасности при работе с пользовательским кодом и данными,
переносимость MCP‑серверов упрощает мульти‑облачные и гибридные сценарии.

Если вы уже строите агентов на Amazon Bedrock, новый stateful‑режим MCP логично использовать там, где нужен диалоговый сценарий, LLM‑генерация на стороне клиента и прозрачный прогресс долгих задач.

Как запустить

Ниже — минимальные шаги, чтобы повторить примеры из материала.

1. Запуск stateful MCP‑сервера с elicitation

Установите зависимости (упрощённо):
- fastmcp
- pydantic
- boto3 (для работы с DynamoDB через FinanceDB)
Настройте окружение AWS:
- AWS_REGION или AWS_DEFAULT_REGION (например, us-east-1),
- таблицы DynamoDB для транзакций (реализовано в dynamo_utils.FinanceDB).
Запустите сервер:

python agents/mcp_client_features.py

Внутри файла уже есть запуск:

if __name__ == '__main__':
    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=8000,
        stateless_http=False
    )

2. Клиент с поддержкой elicitation

Создайте клиент, который подключается к вашему MCP‑серверу по streamable-http и регистрирует elicitation_handler (полный код выше).

Главное — передать elicitation_handler в конструктор Client:

transport = StreamableHttpTransport(url=mcp_url, headers=headers)

async with Client(transport, elicitation_handler=elicit_handler) as client:
    result = await client.call_tool('add_expense_interactive', {'user_alias': 'me'})
    print(result.content[0].text)

3. Клиент с поддержкой sampling и Bedrock

Настройте AWS‑доступ для boto3 с правами на bedrock-runtime.
Используйте код из примера с sampling_handler и MODEL_ID = 'us.anthropic.claude-haiku-4-5-20251001-v1:0'.
Запустите клиента и вызовите инструмент analyze_spending.

4. Прогресс‑нотификации

Для прогресса на сервере достаточно вызывать:

await ctx.report_progress(progress=step_number, total=total_steps)

Клиенту нужно реализовать обработку notifications/progress в своём MCP‑клиенте (конкретная реализация зависит от библиотеки, которую вы используете поверх MCP).

Stateful MCP в Amazon Bedrock AgentCore делает агентные сценарии ближе к реальным бизнес‑процессам: с паузами, уточнениями, отчётами о ходе работы и генерацией текстов там, где это действительно нужно. Для разработчиков, которые уже в экосистеме AWS, это прямой способ построить более живые и понятные пользователю агенты без жёсткой привязки к одной LLM и без лишнего управления ключами и сессиями на своей стороне.