Sample MCP For Long Runing Tasks With Amazon Bedrock Agentcore

使用Strands Agent集成在Amazon Bedrock AgentCore上构建长期运行的MCP服务器

此仓库演示了在模型上下文协议（MCP）服务器中处理长时间运行任务的四种不同模式，以及链代理/代理核心实现和测试框架。

回购结构

mcp-longresponse-pattern/
├── mcp/                    # MCP server implementations
│   ├── mcp_server_pattern1.py  # Progress reporting pattern
│   ├── mcp_server_pattern2.py  # Background timer pattern
│   ├── mcp_server_pattern3.py  # Async task management pattern
│   ├── mcp_server_pattern4.py  # Memory-integrated async pattern
│   ├── Dockerfile             # MCP server containerization
│   └── requirements.txt       # MCP server dependencies
├── agent/                  # Agent implementations
│   ├── agent_local.py         # Local agent with streaming
│   ├── agent.py              # Basic agent implementation
│   ├── Dockerfile             # Agent containerization
│   └── requirements.txt       # Agent dependencies
├── local_test/            # Test clients
│   ├── local_mcp_test.py     # Direct MCP client test
│   └── local_agentcore_test.py # AgentCore HTTP test
├── utils/                 # Utility modules
│   ├── agentcore_client.py    # AgentCore client utilities
│   ├── agentcore_memory.py    # Memory service utilities
│   ├── build_tools.py         # Build automation tools
│   ├── codebuild.py          # AWS CodeBuild integration
│   └── ecr.py               # AWS ECR utilities
├── config.json            # Configuration settings
├── deploy.py             # Deployment automation
├── deploy_memory.py      # Memory service deployment
├── local_mcp_test.sh     # MCP server test script
├── local_agentcore_mcp_test.sh # Agent+MCP integration test
└── requirements.txt      # Root project dependencies

上下文消息传递方法

pattern1

Context Messaging利用MCP的内置上下文对象将周期性信号从服务器发送到MCP客户端，从而在较长的操作过程中有效地保持连接。将其视为发送“心跳”消息，以防止连接超时。

模式1：定期进度消息(mcp_server_pattern1.py)

用例: 当任务进度可以很容易地量化时很有用，例如包含迭代器。

主要特点:

• 用途 ctx.report_progress(progress, total, message) 用于实时进度跟踪
• 通过类型注释自动注入上下文
• 最适合具有可预测进度里程碑的任务

实施:

@mcp.tool()
async def model_training(model_name: str, epochs: int, ctx: Context) -> str:

    for i in range(epochs):
        progress = (i + 1) / epochs
        await asyncio.sleep(5)  # Simulate work
        await ctx.report_progress(
            progress=progress,
            total=1.0,
            message=f"Step {i + 1}/{epochs}",
        )
    
    return f"{model_name} training completed"

模式2：上下文消息传递（后台定时器）(mcp_server_pattern2.py)

用例: 当任务进度无法轻松量化时，例如处理不可预测的数据集或进行外部API调用时，这很有用。

主要特点:

• 使用并发定时器任务进行定期更新 ctx.info()
• 后台监控的主要任务执行
• 任务之间基于事件的协调
• 适用于持续时间不可预测的任务

上下文用法:

• ctx.info()：初始任务开始通知

实施:

@mcp.tool()
async def model_training(model_name: str, epochs: int, ctx: Context) -> str:
    await ctx.info(f"Starting {model_name} Training:")
    done_event = asyncio.Event()
    start_time = time.time()

    async def timer():
        while not done_event.is_set():
            elapsed = time.time() - start_time
            await ctx.info(f"Processing ......: {elapsed:.1f} seconds elapsed")
            await asyncio.sleep(5)

    timer_task = asyncio.create_task(timer())
    
    # Main task execution
    for i in range(epochs):
        await asyncio.sleep(5)  # Simulate work
    
    done_event.set()
    await timer_task
    return f"{model_name} training completed"

异步任务管理

模式3：记忆中(mcp_server_pattern3.py)

pattern3

用例: 真正长时间运行的操作可能需要几个小时，或者用户需要稍后断开连接和重新连接的情况

主要特点:

• 任务启动立即返回任务ID
• 用于状态检查和结果检索的单独工具
• 具有进度跟踪功能的内存任务存储
• 非常适合长时间运行的后台进程

提供的工具:

• model_training()：启动任务，返回任务ID
• check_task_status()：监控任务进度
• get_task_results()：检索已完成的任务结果

实施:

tasks: Dict[str, Dict[str, Any]] = {}

@mcp.tool()
def model_training(model_name: str, epochs: int = 10) -> str:
    task_id = str(uuid.uuid4())
    tasks[task_id] = {"status": "started", "progress": 0.0, "task_type": "model_training"}
    asyncio.create_task(_execute_model_training(task_id, model_name, epochs))
    return f"Model Training task initiated with task ID: {task_id}"

模式4：外部内存（代理核+链代理实现）(mcp_server_pattern4.py)

pattern 4

用例：具有持久内存集成的长时间运行任务

Pattern 4 Demo

主要特点:

• 将异步任务管理与Amazon Bedrock AgentCore内存服务相结合
• 从中提取会话上下文 Mcp-Session-Id AgentCore运行时提供的标头
• 使用MemoryClient将任务结果存储在持久云内存中
• 通过基于云的内存持久性实现跨会话任务连续性

AgentCore运行时集成:

• 当部署在AgentCore运行时，平台会自动包括 Mcp-Session-Id 头球
• Mcp-Session-Id 标题格式： session_id@@@memory_id@@@actor_id （以 @@@). Mcp-Session-Id 标头用于将代理会话ID、内存ID和参与者ID传递给MCP服务器
• ctx.request_context.request.headers.get("mcp-session-id")：提取组合会话信息
• 保持与同一Amazon Bedrock AgentCore运行时会话的连接连续性

内存集成:

• 用途 MemoryClient 与Amazon Bedrock AgentCore存储服务连接
• 任务完成后自动将任务结果保存到代理的内存中
• 内存持续存在于单个会话之外，可以在不同的交互中访问
• 结果与会话上下文一起作为事件存储在AgentCore内存中

AgentCore内存集成 (agent/agent.py):

• 用途 AgentCoreMemoryConfig 使用session_id、memory_id和actor_id配置内存连接
• 用途 AgentCoreMemorySessionManager 处理与Strands代理的持久内存集成
• 自动管理会话历史加载和消息持久性
• 通过内存持久性保持会话之间的对话连续性

与模式3的主要区别:

• 与基于云的AgentCore内存服务集成以实现持久性
• 使用提供的平台 Mcp-Session-Id 会话管理标头
• 完成后自动将结果保存到持久内存中
• 不需要状态/结果检查工具（具有内存持久性的即发即弃）

MCP服务器实现:

@mcp.tool()
def model_training(model_name: str, epochs: int, ctx: Context) -> str:
    # Extract session context from Mcp-Session-Id header
    mcp_session_id = ctx.request_context.request.headers.get("mcp-session-id", "")
    temp_id_list = mcp_session_id.split("@@@")
    session_id = temp_id_list[0]
    memory_id = temp_id_list[1]
    actor_id = temp_id_list[2]

    asyncio.create_task(_execute_model_training(model_name, epochs, session_id, actor_id, memory_id))
    return f"Model {model_name} Training task has been initiated. Results will be saved to memory upon completion."

代理内存集成实现:

# Configure AgentCore memory
agentcore_memory_config = AgentCoreMemoryConfig(
    memory_id=memory_id,
    session_id=session_id,
    actor_id=actor_id
)

# Create session manager for automatic memory handling
session_manager = AgentCoreMemorySessionManager(
    agentcore_memory_config=agentcore_memory_config
)

# Create agent with session manager
agent = Agent(
    tools=tools,
    callback_handler=call_back_handler,
    session_manager=session_manager
)

测试和部署

本地MCP测试(local_mcp_test.sh)

目的：使用Strands代理集成测试MCP服务器

工作流程:

1. 模式选择：选择MCP服务器模式的交互式菜单（1-4）
2. MCP服务器启动：在端口8000本地启动选定的MCP服务器
3. 代理集成：跑步 local_mcp_test.py 创建连接到MCP服务器的Strands代理
4. 交互式测试：用于模式验证的直接代理MCP通信
5. 清理：自动终止MCP服务器进程

用法:

./local_mcp_test.sh
# Select pattern (1-4)
# Test Strands agent + MCP server integration

本地代理核心模拟(local_agentcore_mcp_test.sh)

目的：在本地模拟AgentCore运行时环境

工作流程:

1. 模式选择：选择MCP服务器模式（1-4）
2. 双服务器设置:

- 端口8000上的MCP服务器 - agent_local.py 在端口8080上创建AgentCore运行时应用程序

1. 运行时模拟：Strands代理在AgentCore应用程序中连接到MCP服务器
2. HTTP测试: local_agentcore_test.py 通过HTTP API与代理交互
3. 环境清理：终止两个服务器进程

用法:

./local_agentcore_mcp_test.sh
# Select pattern (1-4)
# Test AgentCore runtime simulation

生产部署(deploy.py)

目的：部署到AWS AgentCore运行时环境

部署流程:

1. 构建和部署：部署MCP服务器和代理(agent.py)到AgentCore运行时
2. AWS集成：使用默认的AWS环境配置
3. 生产测试: run.py 在云环境中与部署的代理交互

清理过程:

• 处理已部署组件的资源清理和拆卸

用法:

python deploy.py  # Deploy to AgentCore runtime
python run.py     # Interact with deployed agent

配置

内存配置(config.json)

{
  "job_name": "default-job",
  "region": "us-east-1"
}