# 有状态 MCP 服务器功能
<a name="mcp-stateful-features"></a>

模型上下文协议 (MCP) 为 AI 应用程序提供了一种与外部数据和功能交互的标准化方式。本指南演示如何构建全面的 MCP 服务器，展示所有主要协议功能，以及如何在本地和部署到 Amazon Bedrock 时对其进行测试。 AgentCore

有关协议的完整详细信息，请参阅 [MCP 规范](https://modelcontextprotocol.io/specification/2025-11-25)。

## MCP 功能概述
<a name="mcp-features-overview"></a>

MCP 服务器可以通过多种功能类型向客户端公开功能。本指南演示了以下功能：

 **资源**   
资源将服务器中的数据和内容暴露给 MCP 客户端。使用资源共享配置、参考数据或客户端或 AI 模型可以读取的任何上下文信息。资源由 URI 标识（例如，`travel://destinations`）。

 **提示**   
提示是可重复使用的模板，用于为 AI 模型生成结构化消息。使用提示来标准化常见的互动，例如生成装箱单或学习目的地的当地短语。

 **工具**   
工具是 AI 模型可以调用以执行操作或检索信息的函数。工具范围从简单的数据查找到结合其他 MCP 功能的复杂多步骤工作流程。

 **引发**   
Elicitation 允许在工具执行期间由服务器启动的用户输入请求。当您的工具需要以交互方式收集信息（例如通过多回合对话收集旅行偏好）时，请使用 elicitation。

 **采样**   
采样允许服务器向客户端请求 LLM-generated 内容。当您的工具需要生成 AI-powered 文本（例如基于用户偏好的个性化旅行推荐）时，请使用采样。

 **进度通知**   
进度通知可让客户随时了解长期运行的操作。在搜索航班或处理预订等任务期间，使用进度报告提供实时反馈。

**注意**  
引发、采样和进度通知等功能需要有状态的 MCP 会话。通过在运行服务器`stateless_http=False`时进行设置来启用状态模式。

 **会话管理** 

在状态模式下，服务器在初始化调用期间返回`Mcp-Session-Id`标头。客户端必须在后续请求中包含此会话 ID 才能维护会话上下文。如果服务器终止或会话过期，请求可能会返回 404 错误，并且客户端必须重新初始化才能获得新的会话 ID。有关更多详细信息，请参阅 MCP 规范中的[会话管理](https://modelcontextprotocol.io/specification/2025-11-25/basic/transports#session-management)。

## 创建具有所有功能的 MCP 服务器
<a name="mcp-create-stateful-server"></a>

 **设置项目** 

1. 创建具有所需依赖关系的`requirements.txt`文件：

   ```
   fastmcp>=2.10.0
   mcp
   ```

1. 安装依赖项：

   ```
   pip install -r requirements.txt
   ```

使用以下代码创建名为 `travel_server.py` 的文件。该旅行预订代理在真实的工作流程中演示了所有 MCP 功能：

### travel\_server.py-完整的 MCP 服务器
<a name="mcp-stateful-server-code"></a>

```
"""
Travel Booking Agent - Stateful MCP Server
Demonstrates all MCP features in a real-world travel booking workflow:
- Elicitation: Collect trip preferences interactively
- Progress: Show search progress for flights and hotels
- Sampling: AI-generated personalized recommendations
- Resources: Expose destination data and pricing
- Prompts: Templates for packing lists and local phrases
"""
import asyncio
import json
from fastmcp import FastMCP, Context
from enum import Enum

mcp = FastMCP("Travel-Booking-Agent")

# ============================================================
# DATA
# ============================================================

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"],
              "phrases": ["Bonjour", "Merci", "S'il vous plait"]},
    "tokyo": {"name": "Tokyo, Japan", "flight": 900, "hotel": 150,
              "highlights": ["Shibuya", "Senso-ji Temple", "Mt. Fuji day trip"],
              "phrases": ["Konnichiwa", "Arigato", "Sumimasen"]},
    "new york": {"name": "New York, USA", "flight": 350, "hotel": 250,
                 "highlights": ["Central Park", "Broadway", "Statue of Liberty"],
                 "phrases": ["Hey!", "Thanks", "Excuse me"]},
    "bali": {"name": "Bali, Indonesia", "flight": 800, "hotel": 100,
             "highlights": ["Ubud Rice Terraces", "Tanah Lot", "Beach clubs"],
             "phrases": ["Selamat pagi", "Terima kasih", "Sama-sama"]}
}

# ============================================================
# RESOURCES - Expose data to MCP clients
# ============================================================

@mcp.resource("travel://destinations")
def list_destinations() -> str:
    """All available destinations with pricing."""
    return json.dumps({k: {"name": v["name"], "flight": v["flight"], "hotel": v["hotel"]}
                       for k, v in DESTINATIONS.items()}, indent=2)

@mcp.resource("travel://destination/{city}")
def get_destination(city: str) -> str:
    """Detailed info for a specific destination."""
    dest = DESTINATIONS.get(city.lower())
    return json.dumps(dest, indent=2) if dest else f"Unknown: {city}"

# ============================================================
# PROMPTS - Reusable templates for AI generation
# ============================================================

@mcp.prompt()
def packing_list(destination: str, days: int, trip_type: str) -> str:
    """Generate packing list prompt."""
    return f"Create a {days}-day packing list for a {trip_type} trip to {destination}. Be practical and concise."

@mcp.prompt()
def local_phrases(destination: str) -> str:
    """Generate local phrases prompt."""
    dest = DESTINATIONS.get(destination.lower(), {})
    phrases = dest.get("phrases", [])
    return f"Teach me essential phrases for {destination}. Start with: {', '.join(phrases)}"

# ============================================================
# MAIN TOOL - Complete booking with all MCP features
# ============================================================

@mcp.tool()
async def plan_trip(ctx: Context) -> str:
    """
    Plan a complete trip using all MCP features:
    1. Elicitation - Collect preferences
    2. Progress - Search flights and hotels
    3. Sampling - AI recommendations
    """

    # -------- PHASE 1: ELICITATION --------
    # Collect trip details through multi-turn conversation

    dest_result = await ctx.elicit(
        message="Where would you like to go?\nOptions: Paris, Tokyo, New York, Bali",
        response_type=str
    )
    if dest_result.action != "accept":
        return "Trip planning cancelled."

    dest_key = dest_result.data.lower().strip()
    dest = DESTINATIONS.get(dest_key, DESTINATIONS["paris"])

    type_result = await ctx.elicit(
        message="What type of trip?\n1. business\n2. leisure\n3. family",
        response_type=TripType
    )
    if type_result.action != "accept":
        return "Trip planning cancelled."
    trip_type = type_result.data

    days_result = await ctx.elicit(
        message="How many days? (3-14)",
        response_type=int
    )
    if days_result.action != "accept":
        return "Trip planning cancelled."
    days = max(3, min(14, days_result.data))

    travelers_result = await ctx.elicit(
        message="Number of travelers?",
        response_type=int
    )
    if travelers_result.action != "accept":
        return "Trip planning cancelled."
    travelers = travelers_result.data

    # -------- PHASE 2: PROGRESS NOTIFICATIONS --------
    # Search for flights and hotels with progress updates

    total_steps = 5

    await ctx.report_progress(progress=1, total=total_steps)  # Searching flights
    await asyncio.sleep(0.4)

    await ctx.report_progress(progress=2, total=total_steps)  # Comparing airlines
    await asyncio.sleep(0.4)

    await ctx.report_progress(progress=3, total=total_steps)  # Searching hotels
    await asyncio.sleep(0.4)

    await ctx.report_progress(progress=4, total=total_steps)  # Checking availability
    await asyncio.sleep(0.4)

    await ctx.report_progress(progress=5, total=total_steps)  # Finalizing
    await asyncio.sleep(0.2)

    # Calculate costs
    flight_cost = dest["flight"] * travelers
    hotel_cost = dest["hotel"] * days * ((travelers + 1) // 2)  # Rooms needed
    total_cost = flight_cost + hotel_cost

    # -------- PHASE 3: SAMPLING --------
    # Get AI-generated personalized recommendations

    ai_tips = f"Enjoy {dest['name']}!"
    try:
        response = await ctx.sample(
            messages=f"Give 3 brief tips for a {trip_type} trip to {dest['name']} for {travelers} travelers, {days} days. Max 60 words.",
            max_tokens=150
        )
        if hasattr(response, 'text') and response.text:
            ai_tips = response.text
    except Exception:
        ai_tips = f"Visit {dest['highlights'][0]}, try local food, learn basic phrases!"

    # -------- FINAL CONFIRMATION --------

    confirm = await ctx.elicit(
        message=f"""
========== TRIP SUMMARY ==========
Destination: {dest['name']}
Trip Type: {trip_type}
Duration: {days} days
Travelers: {travelers}

COSTS:
  Flights: ${flight_cost}
  Hotels: ${hotel_cost} ({(travelers + 1) // 2} room(s) x {days} nights)
  TOTAL: ${total_cost}

Confirm booking? (Yes/No)""",
        response_type=["Yes", "No"]
    )

    if confirm.action != "accept" or confirm.data == "No":
        return "Booking cancelled. Your search results are saved for 24 hours."

    # -------- FINAL RESULT --------

    highlights_str = '\n'.join(f'  * {h}' for h in dest['highlights'])
    phrases_str = '\n'.join(f'  * {p}' for p in dest['phrases'])

    return f"""
{'=' * 50}
  BOOKING CONFIRMED!
{'=' * 50}

Booking Reference: TRV-{ctx.session_id[:8].upper()}

TRIP DETAILS:
  {dest['name']}
  {days} days | {travelers} traveler(s)
  Trip type: {trip_type}

FLIGHTS: ${flight_cost}
  Outbound: Day 1, Morning departure
  Return: Day {days}, Evening departure

ACCOMMODATION: ${hotel_cost}
  {(travelers + 1) // 2} room(s) for {days} nights

TOTAL PAID: ${total_cost}

HIGHLIGHTS TO EXPLORE:
{highlights_str}

USEFUL PHRASES:
{phrases_str}

AI RECOMMENDATIONS:
{ai_tips}

{'=' * 50}
Thank you for booking with Travel Agent!
"""

if __name__ == "__main__":
    print("=" * 60)
    print("  Travel Booking Agent - Stateful MCP Server")
    print("=" * 60)
    print("\n MCP FEATURES DEMONSTRATED:")
    print("   * Elicitation - Multi-turn trip preference collection")
    print("   * Progress    - Real-time search progress updates")
    print("   * Sampling    - AI-powered travel recommendations")
    print("   * Resources   - Destination data and pricing")
    print("   * Prompts     - Packing list and phrase templates")
    print("\n TOOLS:")
    print("   plan_trip - Complete booking flow with all features")
    print("\n RESOURCES:")
    print("   travel://destinations - All destinations")
    print("   travel://destination/{city} - City details")
    print("\n PROMPTS:")
    print("   packing_list - Generate packing suggestions")
    print("   local_phrases - Learn useful phrases")
    print("\n" + "=" * 60)
    print(f" Server: http://0.0.0.0:8000/mcp")
    print("=" * 60)

    mcp.run(
        transport="streamable-http",
        host="0.0.0.0",
        port=8000,
        stateless_http=False
    )
```

## 本地测试
<a name="mcp-test-stateful-local"></a>

 **启动服务器** 
+ 运行 MCP 服务器：

  ```
  python travel_server.py
  ```

  您应该会看到显示服务器在端口 8000 上运行的输出。

使用以下代码创建名为 `test_client.py` 的文件。此客户端测试所有 MCP 功能，包括资源、提示和主工具：

### test\_client.py-完整测试客户端
<a name="mcp-stateful-client-code"></a>

```
"""
Travel Booking Agent - Test Client
Tests all MCP features: Elicitation, Sampling, Progress, Resources, Prompts
"""
import asyncio
import os
import sys
from fastmcp import Client
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.client.elicitation import ElicitResult
from mcp.types import CreateMessageResult, TextContent

async def elicit_handler(message: str, response_type, params, ctx):
    """Handle elicitation - interactive input."""
    print(f"\n>>> Server asks: {message}")

    if isinstance(response_type, list):
        for i, opt in enumerate(response_type, 1):
            print(f"    {i}. {opt}")
        choice = input("    Your choice (number): ").strip()
        response = response_type[int(choice) - 1]
    else:
        hint = " (number)" if response_type == int else ""
        response = input(f"    Your answer{hint}: ").strip()
        if response_type == int:
            response = int(response)

    print(f"<<< Responding: {response}")
    return ElicitResult(action="accept", content={"value": response})

async def sampling_handler(messages, params, ctx):
    """Handle sampling - provide LLM response."""
    print(f"\n>>> AI Sampling Request")
    prompt = messages if isinstance(messages, str) else str(messages)
    print(f"    Prompt: {prompt[:80]}...")

    user_input = input("    Enter AI response (or Enter for auto): ").strip()
    if not user_input:
        user_input = "1. Book popular attractions early. 2. Try local street food. 3. Learn basic greetings!"

    print(f"<<< AI Response: {user_input}")
    return CreateMessageResult(
        role="assistant",
        content=TextContent(type="text", text=user_input),
        model="test-model",
        stopReason="endTurn"
    )

async def progress_handler(progress: float, total: float | None, message: str | None):
    """Handle progress notifications."""
    pct = int((progress / total) * 100) if total else 0
    bar = "#" * (pct // 5) + "-" * (20 - pct // 5)
    print(f"\r    Progress: [{bar}] {pct}% ({int(progress)}/{int(total or 0)})", end="", flush=True)
    if progress == total:
        print(" Done!")

async def main():
    local_test = os.getenv('LOCAL_TEST', 'true').lower() == 'true'

    if local_test:
        url = sys.argv[1] if len(sys.argv) > 1 else "http://localhost:8000/mcp"
        token = None
    else:
        agent_arn = os.getenv('AGENT_ARN')
        if not agent_arn:
            print("ERROR: Missing AGENT_ARN environment variable")
            sys.exit(1)

        encoded_arn = agent_arn.replace(':', '%3A').replace('/', '%2F')
        endpoint = os.getenv('MCP_ENDPOINT', 'https://bedrock-agentcore.us-west-2.amazonaws.com')
        url = f"{endpoint}/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
        token = os.getenv('BEARER_TOKEN')

        if not token:
            print("ERROR: Missing BEARER_TOKEN for remote testing")
            sys.exit(1)

        print(f"  Agent ARN: {agent_arn}")
        print(f"  Endpoint: {endpoint}")

    print("=" * 60)
    print("  Travel Agent - MCP Feature Test Client")
    print("=" * 60)

    headers = {}
    if token:
        headers["Authorization"] = f"Bearer {token}"
        print(f"  Using auth token (len={len(token)})")

    transport = StreamableHttpTransport(url=url, headers=headers)
    client = Client(
        transport,
        elicitation_handler=elicit_handler,
        sampling_handler=sampling_handler,
        progress_handler=progress_handler
    )

    try:
        await client.__aenter__()

        # Test Resources
        print("\n[1] Testing RESOURCES...")
        resources = await client.list_resources()
        print(f"    Found {len(resources)} resource(s)")

        # Test Prompts
        print("\n[2] Testing PROMPTS...")
        prompts = await client.list_prompts()
        print(f"    Found {len(prompts)} prompt(s)")

        # Test Main Tool (Elicitation + Progress + Sampling)
        print("\n[3] Testing PLAN_TRIP tool...")
        print("    (This tests Elicitation, Progress, and Sampling)\n")

        result = await client.call_tool("plan_trip", {})

        print("\n" + "=" * 60)
        print("RESULT:")
        print("=" * 60)
        print(result.content[0].text)

        print("=" * 60)
        print("  ALL TESTS COMPLETED!")
        print("=" * 60)

    except Exception as e:
        print(f"\nERROR: {e}")
        return False
    finally:
        await client.__aexit__(None, None, None)

    return True

if __name__ == "__main__":
    success = asyncio.run(main())
    sys.exit(0 if success else 1)
```

 **运行本地测试** 

1. 当服务器在一个终端中运行时，打开一个新终端并运行测试客户端：

   ```
   python test_client.py
   ```

1. 客户端测试资源和提示，然后运行该`plan_trip`工具，在完整的工作流程中演示启示、进度通知和采样。

## 部署到 Amazon Bedrock AgentCore
<a name="mcp-deploy-stateful"></a>

 **配置和部署** 

1. 如果您尚未安装 AgentCore CLI，请安装：

   ```
   npm install -g @aws/agentcore
   ```

1. 创建要部署的项目：

   ```
   agentcore create --name TravelAgentDemo --protocol MCP
   ```

1. 部署代理：

   ```
   agentcore deploy
   ```

   部署完成后，请记下输出中提供的代理 ARN。

## 测试已部署的代理
<a name="mcp-test-stateful-deployed"></a>

 **测试已部署的代理** 

1. 设置所需的环境变量：

   ```
   export AGENT_ARN='arn:aws:bedrock-agentcore:us-west-2:YOUR_ACCOUNT:runtime/YOUR_AGENT_NAME'
   export BEARER_TOKEN='your_bearer_token'
   ```

   用您的实际代理 ARN 和不记名令牌替换占位符。

1. 在远程模式下运行测试客户端：

   ```
   LOCAL_TEST=false python test_client.py
   ```

1. 客户端将测试资源、提示和`plan_trip`工具。按照交互式提示完成预订，其中演示了已部署代理的招标、进度通知和采样。

### 预期输出
<a name="mcp-expected-output"></a>

```
  Agent ARN: arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/TravelAgentDemo
  Endpoint: https://bedrock-agentcore.us-west-2.amazonaws.com
============================================================
  Travel Agent - MCP Feature Test Client
============================================================
  Using auth token (len=1034)

[1] Testing RESOURCES...
    Found 1 resource(s)

[2] Testing PROMPTS...
    Found 2 prompt(s)

[3] Testing PLAN_TRIP tool...
    (This tests Elicitation, Progress, and Sampling)

>>> Server asks: Where would you like to go?
Options: Paris, Tokyo, New York, Bali
    Your answer: Paris
<<< Responding: Paris

>>> Server asks: What type of trip?
1. business
2. leisure
3. family
    Your answer: leisure
<<< Responding: leisure

>>> Server asks: How many days? (3-14)
    Your answer (number): 5
<<< Responding: 5

>>> Server asks: Number of travelers?
    Your answer (number): 2
<<< Responding: 2

    Progress: [####################] 100% (5/5) Done!

>>> AI Sampling Request
    Prompt: Give 3 brief tips for a leisure trip to Paris, France for 2 travelers...
    Enter AI response (or Enter for auto):
<<< AI Response: 1. Book popular attractions early. 2. Try local street food. 3. Learn basic greetings!

>>> Server asks:
========== TRIP SUMMARY ==========
Destination: Paris, France
Trip Type: leisure
Duration: 5 days
Travelers: 2

COSTS:
  Flights: $900
  Hotels: $900 (1 room(s) x 5 nights)
  TOTAL: $1800

Confirm booking? (Yes/No)
    1. Yes
    2. No
    Your choice (number): 1
<<< Responding: Yes

============================================================
RESULT:
============================================================

==================================================
  BOOKING CONFIRMED!
==================================================

Booking Reference: TRV-A1B2C3D4

TRIP DETAILS:
  Paris, France
  5 days | 2 traveler(s)
  Trip type: leisure

FLIGHTS: $900
  Outbound: Day 1, Morning departure
  Return: Day 5, Evening departure

ACCOMMODATION: $900
  1 room(s) for 5 nights

TOTAL PAID: $1800

HIGHLIGHTS TO EXPLORE:
  * Eiffel Tower
  * Louvre
  * Notre-Dame

USEFUL PHRASES:
  * Bonjour
  * Merci
  * S'il vous plait

AI RECOMMENDATIONS:
1. Book popular attractions early. 2. Try local street food. 3. Learn basic greetings!

==================================================
Thank you for booking with Travel Agent!

============================================================
  ALL TESTS COMPLETED!
============================================================
```

**提示**  
你也可以使用 MCP Inspector 来测试 MCP 服务器，这是一款用于测试 MCP 服务器的可视化工具。有关本地测试说明，请参阅[使用 MCP 检查器进行本地测试](runtime-mcp.md#runtime-mcp-appendix-b)。有关远程测试的说明，请参阅[使用 MCP 检查器进行远程测试](runtime-mcp.md#runtime-mcp-appendix-c)。