搞 MCP 开发的应该都碰过这个问题——server 不记状态。每次请求都是从零开始，上一轮用户说了什么完全不知道。
这在简单场景没啥，但做复杂 agent 就很烦。比如订机票，用户得一口气把目的地、日期、人数、舱位全说清楚，不能分步来。
上周亚马逊云科技给 Bedrock AgentCore Runtime 加了有状态 MCP 支持。三个新能力：Elicitation（主动追问）、Sampling（反向调 LLM）、Progress Notifications（进度通知）。我试了一天，记录一下。
怎么用
核心是 FastMCP 库，版本要 >= 2.10.0：
pip install "fastmcp>=2.10.0" mcp
Server 端
from fastmcp import FastMCP, Context
from enum import Enum
import asynciomcp = FastMCP("Travel-Booking-Agent")class TripType(str, Enum):BUSINESS = "business"LEISURE = "leisure"FAMILY = "family"DESTINATIONS = {"paris": {"name": "Paris, France", "flight": 450, "hotel": 180,"highlights": ["Eiffel Tower", "Louvre", "Notre-Dame"]},"tokyo": {"name": "Tokyo, Japan", "flight": 900, "hotel": 150,"highlights": ["Shibuya", "Senso-ji Temple"]},
}@mcp.tool()
async def plan_trip(ctx: Context) -> str:# Elicitation: 多轮追问dest_result = await ctx.elicit(message="Where would you like to go? (Paris/Tokyo)",response_type=str)if dest_result.action != "accept":return "Cancelled."dest = DESTINATIONS.get(dest_result.data.lower().strip())type_result = await ctx.elicit(message="Trip type? (business/leisure/family)",response_type=TripType)days_result = await ctx.elicit(message="How many days? (3-14)",response_type=int)days = max(3, min(14, days_result.data))# Progress: 搜索进度for i in range(1, 6):await ctx.report_progress(progress=i, total=5)await asyncio.sleep(0.4)# Sampling: 让客户端的 LLM 生成推荐try:resp = await ctx.sample(messages=f"3 tips for {dest['name']}, {days} days",max_tokens=150)tips = resp.text if hasattr(resp, 'text') and resp.text else ""except Exception:tips = f"Visit {dest['highlights'][0]}!"total = dest["flight"] * 2 + dest["hotel"] * daysreturn f"Booked: {dest['name']}, {days} days, ${total}\nTips: {tips}"# 启动时必须关闭无状态模式
mcp.run(transport="streamable-http",host="0.0.0.0",port=8000,stateless_http=False
)
客户端
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.client.elicitation import ElicitResult
from mcp.types import CreateMessageResult, TextContentasync def elicit_handler(message, response_type, params, ctx):print(f"\n>>> {message}")answer = input("回答: ").strip()if response_type == int:answer = int(answer)return ElicitResult(action="accept", content={"value": answer})async def sampling_handler(messages, params, ctx):return CreateMessageResult(role="assistant",content=TextContent(type="text", text="试试当地美食！"),model="local",stopReason="endTurn")transport = StreamableHttpTransport(url="http://localhost:8000/mcp")
client = Client(transport,elicitation_handler=elicit_handler,sampling_handler=sampling_handler)
部署到 AgentCore
pip install bedrock-agentcore-starter-toolkit
agentcore configure -e travel_server.py -p MCP -n travel_agent_demo
agentcore deploy
每个 session 在独立 microVM 里运行，session 之间隔离。
踩坑
stateless_http默认 True，不关就用不了 elicitation- Session 有超时，客户端要处理 404 重连
sampling_handler返回空文本会导致 server 端字符串拼接报错
适用场景
有状态 MCP 把多轮对话能力从 agent 框架层下沉到 tool 层面。适合多步骤表单、审批流程、交互式调试、个性化推荐等场景。
目前 14 个区域可用，含亚太区的孟买、首尔、新加坡、悉尼、东京。