快时尚电商行业智能体设计思路与应用实践（六）借助 Amazon Bedrock AgentCore MCP Server，Amazon Bedrock，Strands Agents，Kiro 实现智能体极速研发

# Windows (使用PowerShell)
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# 或使用pip安装
pip install uv

1.3 配置MCP服务器

{
  "mcpServers": {
    "awslabs.amazon-bedrock-agentcore-mcp-server": {
      "command": "uvx",
      "args": ["awslabs.amazon-bedrock-agentcore-mcp-server@latest"],
      "env": {
        "FASTMCP_LOG_LEVEL": "ERROR"
      },
      "disabled": false,
      "autoApprove": []
    }
  }
}

# 在IDE中通过AI助手调用MCP Server
query = "strands agents integration example"
# MCP Server立即返回相关文档和代码示例

# 查询Memory配置指南
manage_agentcore_memory()

# 获取部署指导
manage_agentcore_runtime()

mkdir intelligent-customer-service
cd intelligent-customer-service

# 创建虚拟环境
python -m venv .venv

# 激活虚拟环境
# Windows
.venv\Scripts\activate
# macOS/Linux
source .venv/bin/activate

# 安装必要依赖
pip install "bedrock-agentcore-starter-toolkit>=0.1.21" strands-agents boto3

我们将基于以下组件搭建示例系统：
Amazon Bedrock AgentCore：负责智能体的运行时管理、状态维护与可观测性。
Amazon Bedrock 模型平台：提供高性能的基础模型能力。
Strands Agents：用于定义和编排智能体的对话流与行为逻辑。
示例场景为一个 简化的购物客服对话系统，包含常见的业务流程（SOP），如下单、退货、物流查询、商品咨询等。为展示清晰的逻辑链路，这些 SOP 将以硬编码方式实现。
智能客服应具备以下能力：
多轮对话理解能力：能够根据用户连续输入判断意图变化。
基于 SOP 的流程调度：根据识别出的意图调用对应的业务流程节点。
智能体链路展示：在执行过程中体现 AgentCore 的执行链路、内存管理与上下文控制。
Strands 流程建模：通过 Strands Agents 将 SOP 与模型推理结合，实现可视化且可维护的对话流程。

"""
购物客服标准操作程序 (SOP)
"""

class ShoppingSOP:
    def __init__(self):
        self.sop_data = {
            "订单查询": {
                "keywords": ["订单", "查询", "状态", "物流", "快递"],
                "response_template": "请提供您的订单号，我来帮您查询订单状态。订单号通常是以'ORD'开头的12位数字。",
                "follow_up": "如果您找不到订单号，可以提供下单时使用的手机号码或邮箱地址。"
            },
            "退换货": {
                "keywords": ["退货", "换货", "退款", "不满意", "质量问题"],
                "response_template": "我理解您的困扰。请告诉我具体的问题：1) 商品质量问题 2) 尺寸不合适 3) 收到错误商品 4) 其他原因",
                "follow_up": "退换货需要在收货后7天内申请，商品需保持原包装完好。"
            },
            "商品咨询": {
                "keywords": ["商品", "产品", "价格", "规格", "参数", "功能"],
                "response_template": "我很乐意为您介绍商品信息。请告诉我您感兴趣的具体商品名称或类别。",
                "follow_up": "您也可以提供商品链接或商品编号，我可以为您提供更详细的信息。"
            },
            "支付问题": {
                "keywords": ["支付", "付款", "银行卡", "支付宝", "微信", "优惠券"],
                "response_template": "关于支付问题，我来帮您解决。常见支付方式包括：支付宝、微信支付、银行卡。请描述您遇到的具体问题。",
                "follow_up": "如果支付失败，请检查网络连接和账户余额，或尝试更换支付方式。"
            },
            "物流配送": {
                "keywords": ["配送", "快递", "物流", "送货", "收货地址"],
                "response_template": "关于配送服务：1) 市内当日达 2) 次日达 3) 标准配送(2-3天) 4) 偏远地区配送(3-7天)",
                "follow_up": "您可以在下单时选择配送方式，部分商品支持指定配送时间。"
            },
            "会员服务": {
                "keywords": ["会员", "积分", "等级", "权益", "折扣"],
                "response_template": "我们的会员体系分为：普通会员、银卡会员、金卡会员、钻石会员。不同等级享有不同权益。",
                "follow_up": "会员积分可用于兑换商品或抵扣现金，积分有效期为2年。"
            }
        }

    def match_intent(self, user_message):
        """根据用户消息匹配意图"""
        user_message_lower = user_message.lower()

        for intent, data in self.sop_data.items():
            for keyword in data["keywords"]:
                if keyword in user_message_lower:
                    return intent, data

        return "通用咨询", {
            "response_template": "感谢您的咨询！我是您的专属客服助手，可以帮您解决订单查询、退换货、商品咨询、支付问题、物流配送和会员服务等问题。请告诉我您需要什么帮助？",
            "follow_up": "您也可以直接描述遇到的具体问题，我会尽力为您解答。"
        }

    def get_response(self, intent, sop_data, context=""):
        """生成回复"""
        response = sop_data["response_template"]
        if "follow_up" in sop_data and context:
            response += f"\n\n补充信息：{sop_data['follow_up']}"
        return response

"""
基于Amazon Bedrock AgentCore和Strands的智能客服系统
"""
import os
import json
from strands import Agent
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from bedrock_agentcore.memory.integrations.strands.config import (
    AgentCoreMemoryConfig, 
    RetrievalConfig
)
from bedrock_agentcore.memory.integrations.strands.session_manager import (
    AgentCoreMemorySessionManager
)
from shopping_sop import ShoppingSOP

# 创建AgentCore应用实例
app = BedrockAgentCoreApp()

# 配置参数
MEMORY_ID = os.getenv("BEDROCK_AGENTCORE_MEMORY_ID")
REGION = os.getenv("AWS_REGION", "us-west-2")
MODEL_ID = "global.anthropic.claude-sonnet-4-5-20250929-v1:0"

# 初始化SOP
sop = ShoppingSOP()

# 日志记录
app.logger.info(f"智能客服系统初始化完成 - Region: {REGION}, Model: {MODEL_ID}")


@app.entrypoint
def invoke(payload, context):
    """
    智能客服主入口函数
    
    Args:
        payload: 请求负载，包含用户消息和元数据
        context: 运行时上下文
        
    Returns:
        dict: 包含回复内容和元数据的响应
    """
    
    # 提取请求参数
    user_message = payload.get("prompt", "")
    actor_id = payload.get("user_id", "customer_default")
    session_id = getattr(context, 'session_id', None) or payload.get("session_id", "default")
    
    app.logger.info(f"收到客服请求 - User: {actor_id}, Session: {session_id}, Message: {user_message}")
    
    # 配置Memory管理（如果启用）
    session_manager = None
    if MEMORY_ID:
        try:
            memory_config = AgentCoreMemoryConfig(
                memory_id=MEMORY_ID,
                session_id=session_id,
                actor_id=actor_id,
                retrieval_config={
                    # 客户偏好信息
                    f"/customers/{actor_id}/preferences": RetrievalConfig(
                        top_k=3, 
                        relevance_score=0.5
                    ),
                    # 历史对话记录
                    f"/customers/{actor_id}/history": RetrievalConfig(
                        top_k=5, 
                        relevance_score=0.6
                    ),
                    # 订单信息
                    f"/customers/{actor_id}/orders": RetrievalConfig(
                        top_k=3, 
                        relevance_score=0.7
                    )
                }
            )
            session_manager = AgentCoreMemorySessionManager(memory_config, REGION)
            app.logger.info(f"Memory管理已启用 - Memory ID: {MEMORY_ID}")
        except Exception as e:
            app.logger.warning(f"Memory管理初始化失败: {e}")
    
    # SOP意图识别
    intent, sop_data = sop.match_intent(user_message)
    sop_response = sop.get_response(intent, sop_data, include_follow_up=True)
    
    app.logger.info(f"意图识别结果: {intent}")
    
    # 构建系统提示词
    system_prompt = f"""你是一个专业、友好的购物客服助手，名字叫小智。

【你的职责】
1. 根据标准操作程序(SOP)为客户提供准确、专业的服务
2. 保持友好、耐心、热情的服务态度
3. 记住客户的对话历史和偏好，提供个性化服务
4. 在必要时主动询问更多信息以更好地帮助客户
5. 对于复杂问题，引导客户提供必要的信息

【当前对话上下文】
- 识别的客户意图：{intent}
- SOP标准回复参考：
{sop_response}

【回复要求】
1. 基于SOP标准回复，结合客户的具体问题生成个性化回复
2. 使用温暖、友好、专业的语调
3. 回复要清晰、具体、可操作
4. 适当使用emoji增加亲和力（但不要过度）
5. 如果客户问题不够明确，礼貌地询问更多细节
6. 结尾可以询问是否还有其他需要帮助的地方

【注意事项】
- 不要生硬地复制SOP内容，要自然地融入对话
- 保持回复简洁，避免信息过载
- 对于无法解决的问题，建议联系人工客服
- 始终保持积极、解决问题的态度

请根据以上要求，为客户提供优质的服务体验。"""

    # 创建Strands代理
    try:
        agent = Agent(
            model=MODEL_ID,
            session_manager=session_manager,
            system_prompt=system_prompt,
            tools=[]  # 可以在这里添加工具，如订单查询API、库存查询API等
        )
        
        app.logger.info("Strands代理创建成功")
        
        # 调用代理处理用户消息
        result = agent(user_message)
        
        # 提取回复内容
        response_text = extract_response_text(result)
        
        app.logger.info(f"代理回复生成成功 - 长度: {len(response_text)}")
        
        # 构建响应
        return {
            "response": response_text,
            "intent": intent,
            "session_id": session_id,
            "user_id": actor_id,
            "status": "success",
            "metadata": {
                "model": MODEL_ID,
                "memory_enabled": MEMORY_ID is not None,
                "sop_matched": intent != "通用咨询"
            }
        }
        
    except Exception as e:
        app.logger.error(f"客服代理处理错误: {str(e)}", exc_info=True)
        
        # 返回友好的错误消息
        return {
            "response": "非常抱歉，我遇到了一些技术问题。😔\n\n请您稍后再试，或者拨打客服热线 400-XXX-XXXX 联系人工客服。\n\n给您带来不便，敬请谅解！",
            "intent": "error",
            "session_id": session_id,
            "user_id": actor_id,
            "status": "error",
            "error": str(e)
        }


def extract_response_text(result):
    """
    从代理结果中提取回复文本
    
    Args:
        result: Strands代理的返回结果
        
    Returns:
        str: 提取的回复文本
    """
    try:
        if hasattr(result, 'message') and result.message:
            content = result.message.get('content', [])
            if content and isinstance(content, list) and len(content) > 0:
                return content[0].get('text', str(result))
        return str(result)
    except Exception as e:
        app.logger.warning(f"提取回复文本失败: {e}")
        return str(result)


# 本地测试入口
if __name__ == "__main__":
    print("=" * 70)
    print("智能客服系统 - 本地测试模式")
    print("=" * 70)
    print("\n启动AgentCore本地服务器...")
    print("访问 http://localhost:8080 进行测试\n")
    print("测试命令示例:")
    print('curl -X POST http://localhost:8080/invocations \\')
    print('  -H "Content-Type: application/json" \\')
    print('  -d \'{"prompt": "我想查询订单", "user_id": "test_user"}\'')
    print("\n" + "=" * 70 + "\n")
    
    # 启动本地服务器
    app.run()

strands-agents
bedrock-agentcore-starter-toolkit>=0.1.21
boto3

# 配置AgentCore项目
agentcore configure -e customer_service_agent.py

# 在配置过程中：
# 1. 执行角色：按Enter自动创建新角色（或选择已有角色）
# 2. ECR仓库：按Enter自动创建
# 3. 需求文件：确认requirements.txt
# 4. OAuth配置：输入'no'
# 5. 请求头白名单：输入'no'  
# 6. Memory配置：输入'yes'启用长期Memory（可选，本地测试可以选'no'）

# 在项目目录下启动开发服务器
agentcore dev

Starting development server with hot reloading
Agent: customer_service_agent
Module: customer_service_agent:app
Server will be available at: http://localhost:8080/invocations
Test your agent with: agentcore invoke --dev "Hello" in a new terminal window
This terminal window will be used to run the dev server
Press Ctrl+C to stop the server

INFO:     Will watch for changes in these directories: ['/path/to/project']
INFO:     Uvicorn running on http://0.0.0.0:8080 (Press CTRL+C to quit)
INFO:     Started reloader process [12345] using WatchFiles
INFO:     Started server process [12346]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "你好，我想查询我的订单状态",
    "user_id": "customer_001"
  }'

{
  "response": "您好！我是小智，很高兴为您服务。请提供您的订单号，我来帮您查询订单状态。订单号通常是以'ORD'开头的12位数字。\n\n如果您找不到订单号，也可以提供下单时使用的手机号码或邮箱地址，我会帮您查找。",
  "intent": "订单查询",
  "session_id": "default",
  "status": "success"
}

curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "我买的衣服尺寸不合适，想要退货",
    "user_id": "customer_001"
  }'

{
  "response": "我理解您的困扰，很抱歉给您带来不便。关于退货，我需要了解一些具体情况：\n\n1) 商品质量问题\n2) 尺寸不合适\n3) 收到错误商品\n4) 其他原因\n\n您的情况是尺寸不合适，对吗？请提供您的订单号，我会立即为您处理退货申请。\n\n温馨提示：退换货需要在收货后7天内申请，商品需保持原包装完好。",
  "intent": "退换货",
  "session_id": "default",
  "status": "success"
}

# 第一轮对话
curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "你好，我是新用户",
    "user_id": "customer_002",
    "session_id": "session_123"
  }'

# 第二轮对话（使用相同的session_id）
curl -X POST http://localhost:8080/invocations \
  -H "Content-Type: application/json" \
  -d '{
    "prompt": "我想了解你们的会员服务",
    "user_id": "customer_002",
    "session_id": "session_123"
  }'

# 基础测试
agentcore invoke --dev "我想查询订单状态"

# 指定用户ID
agentcore invoke --dev '{"prompt": "我想退货", "user_id": "customer_001"}'

# 多轮对话测试
agentcore invoke --dev "你好" --session-id session_456
agentcore invoke --dev "我想了解会员服务" --session-id session_456

# 部署到AgentCore Runtime
agentcore deploy

Building Docker image...
Pushing image to ECR...
Deploying to AgentCore Runtime...
✅ Deployment successful!

Agent ID: agent-abc123def456
Invoke URL: https://runtime.bedrock-agentcore.us-west-2.amazonaws.com/agents/agent-abc123def456/invoke

# 第一轮对话
SESSION_ID=$(python -c "import uuid; print(str(uuid.uuid4()))")
agentcore invoke '{"prompt": "你好，我是新用户", "user_id": "customer_002", "session_id": "'$SESSION_ID'"}' --session-id $SESSION_ID

# 第二轮对话（同一会话）
agentcore invoke '{"prompt": "我想了解你们的会员服务", "user_id": "customer_002", "session_id": "'$SESSION_ID'"}' --session-id $SESSION_ID

# 第三轮对话（同一会话）
agentcore invoke '{"prompt": "如何成为金卡会员？", "user_id": "customer_002", "session_id": "'$SESSION_ID'"}' --session-id $SESSION_ID

# 第一步：第一个会话 - 咨询会员服务
SESSION_ID=$(python -c "import uuid; print(str(uuid.uuid4()))")
agentcore invoke '{"prompt": "你好，我是新用户，想了解会员服务", "user_id": "customer_002"}' --session-id $SESSION_ID

# 第二步：等待30秒
echo "等待30秒让Memory提取长期记忆..."
sleep 30

# 第三步：新会话 - 引用之前的对话
NEW_SESSION_ID=$(python -c "import uuid; print(str(uuid.uuid4()))")
agentcore invoke '{"prompt": "我之前咨询过会员服务，现在想知道更多详情", "user_id": "customer_002"}' --session-id $NEW_SESSION_ID

retrieval_config={
    f"/customers/{actor_id}/preferences": RetrievalConfig(
        top_k=3,              # 检索前3条相关记忆
        relevance_score=0.5   # 相关性阈值（0-1）
    ),
    f"/customers/{actor_id}/history": RetrievalConfig(
        top_k=5,              # 检索前5条历史对话
        relevance_score=0.6
    ),
    f"/customers/{actor_id}/orders": RetrievalConfig(
        top_k=3,
        relevance_score=0.7   # 订单信息要求更高相关性
    )
}

# 检查部署状态
agentcore status

# 查看日志
aws logs tail /aws/bedrock-agentcore/runtimes/AGENT_ID-DEFAULT --follow

https://console.aws.amazon.com/cloudwatch/home?region=us-west-2#gen-ai-observability/agent-core