长运行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

@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"

@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"

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}"

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

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

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

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

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