从概念验证到生产部署，AgentCore轻松破解客服Agent落地难题

from strands import Agentfrom strands.models import BedrockModel
@tooldef get_return_policy(product_category: str) -> str:    """Get return policy information for a specific product category."""    # Returns structured policy info: windows, conditions, processes, refunds    # check github for full code    return{"return_window": "10 days", "conditions": ""}    @tool  def get_product_info(product_type: str) -> str:    """Get detailed technical specifications and information for electronics products."""    # Returns warranty, specs, features, compatibility details    # check github for full code    return{"product": "ThinkPad X1 Carbon", "info": "ThinkPad X1 Carbon info"}    @tooldef web_search(keywords: str, region: str = "us-en", max_results: int = 5) -> str:    """Search the web for updated troubleshooting information."""    # Provides access to current technical solutions and guides    # check github for full code    return "results from websearch"    # Initialize the Bedrock modelmodel = BedrockModel(    model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",    temperature=0.3)
# Create the customer support agentagent = Agent(    model=model,    tools=[        get_product_info,         get_return_policy,         web_search    ],    system_prompt="""You are a helpful customer support assistant for an electronics company.    Use the appropriate tools to provide accurate information and always offer additional help.""")

# Return policy inquiryresponse = agent("What's the return policy for my ThinkPad X1 Carbon?")# Agent correctly uses get_return_policy with "laptops" category
# Technical troubleshooting  response = agent("My iPhone 14 heats up, how do I fix it?")# Agent uses web_search to find current troubleshooting solutions

pip install boto3 bedrock-agentcore bedrock-agentcore-starter-toolkit

from bedrock_agentcore.memory import MemoryClientfrom bedrock_agentcore.memory.constants import StrategyType
memory_client = MemoryClient(region_name=region)
strategies = [{        StrategyType.USER_PREFERENCE.value: {"name": "CustomerPreferences","description": "Captures customer preferences and behavior","namespaces": ["support/customer/{actorId}/preferences"],}},{        StrategyType.SEMANTIC.value: {"name": "CustomerSupportSemantic",             "description": "Stores facts from conversations","namespaces": ["support/customer/{actorId}/semantic"],}},]# Create memory resource with both strategiesresponse = memory_client.create_memory_and_wait(    name="CustomerSupportMemory",    description="Customer support agent memory",    strategies=strategies,    event_expiry_days=90,)

class CustomerSupportMemoryHooks(HookProvider):    def retrieve_customer_context(self, event: MessageAddedEvent):        """Inject customer context before processing queries"""        user_query = event.agent.messages[-1]["content"][0]["text"]                # Retrieve relevant memories from both strategies        all_context = []        for context_type, namespace in self.namespaces.items():            memories = self.client.retrieve_memories(                memory_id=self.memory_id,                namespace=namespace.format(actorId=self.actor_id),                query=user_query,                top_k=3,            )            # Format and add to context            for memory in memories:                if memory.get("content", {}).get("text"):                    all_context.append(f"[{context_type.upper()}] {memory['content']['text']}")                # Inject context into the user query        if all_context:            context_text = "\n".join(all_context)            original_text = event.agent.messages[-1]["content"][0]["text"]            event.agent.messages[-1]["content"][0]["text"] = f"Customer Context:\n{context_text}\n\n{original_text}"
    def save_support_interaction(self, event: AfterInvocationEvent):        """Save interactions after agent responses"""        # Get last customer query and agent response check github for implementation        customer_query = "This is a sample query"        agent_response = "LLM gave a sample response"                # Extract customer query and agent response        # Save to memory forfuture retrieval        self.client.create_event(            memory_id=self.memory_id,            actor_id=self.actor_id,            session_id=self.session_id,            messages=[(customer_query, "USER"), (agent_response, "ASSISTANT")]        )

# Create memory hooks forthis customer sessionmemory_hooks = CustomerSupportMemoryHooks(    memory_id=memory_id,     client=memory_client,     actor_id=customer_id,     session_id=session_id)
# Create agent with memory capabilitiesagent = Agent(    model=model,
    tools=[get_product_info, get_return_policy, web_search],    system_prompt=SYSTEM_PROMPT)

# Test personalized recommendationsresponse = agent("Which headphones would you recommend?")# Agent remembers: "prefers low latency for competitive FPS games"# Response includes gaming-focused recommendations
# Test preference recallresponse = agent("What is my preferred laptop brand?")  # Agent remembers: "prefers ThinkPad models"and"needs Linux compatibility"# Response acknowledges ThinkPad preference and suggests compatible models

# Original Lambda function(simplified)def web_search(keywords: str, region: str = "us-en", max_results: int = 5) -> str:    # web_search functionality        def lambda_handler(event, context):    ifget_tool_name(event)== "web_search":        query = get_named_parameter(event=event, name="query")                search_result = web_search(keywords)        return {"statusCode": 200, "body": search_result}

{        "name": "web_search",        "description": "Search the web for updated information using DuckDuckGo",        "inputSchema": {            "type": "object",            "properties": {                "keywords": {                    "type": "string",                    "description": "The search query keywords"                },                "region": {                    "type": "string",                    "description": "The search region (e.g., us-en, uk-en, ru-ru)"                },                "max_results": {                    "type": "integer",                    "description": "The maximum number of results to return"                }            },            "required": [                "keywords"            ]        }    }

# Create gateway with JWT-based authenticationauth_config = {    "customJWTAuthorizer": {        "allowedClients": [cognito_client_id],        "discoveryUrl": cognito_discovery_url    }}
gateway_response = gateway_client.create_gateway(    name="customersupport-gw",    roleArn=gateway_iam_role,    protocolType="MCP",    authorizerType="CUSTOM_JWT",    authorizerConfiguration=auth_config,    description="Customer Support AgentCore Gateway")

lambda_target_config = {    "mcp": {        "lambda": {            "lambdaArn": lambda_function_arn,            "toolSchema": {"inlinePayload": api_spec},        }    }}
gateway_client.create_gateway_target(    gatewayIdentifier=gateway_id,    name="LambdaTools",    targetConfiguration=lambda_target_config,    credentialProviderConfigurations=[{        "credentialProviderType": "GATEWAY_IAM_ROLE"    }])

# Get OAuth token for gateway accessgateway_access_token = get_token(    client_id=cognito_client_id,    client_secret=cognito_client_secret,    scope=auth_scope,    url=token_url)
# Create authenticated MCP clientmcp_client = MCPClient(    lambda: streamablehttp_client(        gateway_url,        headers={"Authorization": f"Bearer {gateway_access_token['access_token']}"}    ))
# Combine local and MCP toolstools = [    get_product_info,     # Local tool (customer support specific)    get_return_policy,    # Local tool (customer support specific)] + mcp_client.list_tools_sync()  # Centralized tools from gateway
agent = Agent(    model=model,    tools=tools,    hooks=[memory_hooks],    system_prompt=SYSTEM_PROMPT)

# Test web search using centralized tool  response = agent("How can I fix Lenovo ThinkPad with a blue screen?")# Agent uses web_search from AgentCore Gateway

# Your existing agent code remains unchangedmodel = BedrockModel(model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0")memory_hooks = CustomerSupportMemoryHooks(memory_id, memory_client, actor_id, session_id)agent = Agent(    model=model,    tools=[get_return_policy, get_product_info],    system_prompt=SYSTEM_PROMPT,    hooks=[memory_hooks])
def invoke(payload):    user_input = payload.get("prompt", "")    response = agent(user_input)    return response.message["content"][0]["text"]
if __name__ == "__main__":

from bedrock_agentcore_starter_toolkit import Runtime
# Configure secure deployment with Cognito authenticationagentcore_runtime = Runtime()
response = agentcore_runtime.configure(    entrypoint="lab_helpers/lab4_runtime.py",    execution_role=execution_role_arn,    auto_create_ecr=True,    requirements_file="requirements.txt",    region=region,    agent_name="customer_support_agent",    authorizer_configuration={"customJWTAuthorizer": {"allowedClients": [cognito_client_id],"discoveryUrl": cognito_discovery_url,}})# Deploy to productionlaunch_result = agentcore_runtime.launch()

# Customer 1 conversationresponse1 = agentcore_runtime.invoke({"prompt": "My iPhone Bluetooth isn't working. What should I do?"},    bearer_token=auth_token,    session_id="session-customer-1")
# Customer 1 follow-up (maintains context)response2 = agentcore_runtime.invoke({"prompt": "I've turned Bluetooth on and off but it still doesn't work"},    bearer_token=auth_token,    session_id="session-customer-1"  # Same session, context preserved)
# Customer 2 conversation (completely separate)response3 = agentcore_runtime.invoke({"prompt": "Still not working. What is going on?"},    bearer_token=auth_token,    session_id="session-customer-2"  # Different session, no context)