快时尚电商行业智能体设计思路与应用实践（四）借助Amazon AgentCore 实现智能客服 SOP 的可执行 MCP 工具化实践

from mcp.server.fastmcp import FastMCP
from starlette.responses import JSONResponse
import json
import os
import logging
from typing import Dict, List, Optional

# Setup logging
logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Initialize MCP server
mcp = FastMCP(host="0.0.0.0", stateless_http=True)

# In-memory SOP database (in a real scenario, this would be in a database)
# We're using a simple dictionary for demo purposes
sop_database = {
    "billing": {
        "refund_policy": "Customers are eligible for refunds within 30 days of purchase with valid receipt. Process through the billing system with approval for amounts over $50.",
        "payment_methods": "We accept Visa, MasterCard, American Express, and PayPal. For corporate accounts, we also accept wire transfers with NET-30 terms.",
        "subscription_cancellation": "To cancel a subscription, verify customer identity, process in billing portal, and send confirmation email. Offer retention options for high-value accounts."
    },
    "technical_support": {
        "password_reset": "Verify customer identity using security questions. Generate temporary password and send to registered email. Follow up after 24 hours if needed.",
        "account_lockout": "Check lockout reason in security logs. For suspicious activity, escalate to security team. For failed login attempts, verify identity and unlock with 24-hour monitoring.",
        "system_outage": "Acknowledge the issue, check status dashboard, inform about estimated recovery time. Create incident ticket if not already created. Update customers every 30 minutes."
    },
    "product": {
        "warranty_information": "Standard warranty is 12 months from purchase date. Premium products have 24-month coverage. Warranty covers manufacturing defects but not physical damage.",
        "returns_process": "Returns accepted within 14 days with original packaging. Issue RMA number, provide return shipping label, process refund within 5 business days of receiving returned item.",
        "product_compatibility": "Check product compatibility matrix in knowledge base. For custom setups, consult with technical specialist before providing definitive answer."
    }
}

@mcp.tool()
def get_sop_for_question(question: str) -> Dict:
    """
    Retrieves the appropriate Standard Operating Procedure (SOP) based on customer's question.
    
    Args:
        question (str): The customer's question or issue description
        
    Returns:
        Dict: A dictionary containing the SOP information
    """
    logger.info(f"Received question: {question}")
    
    # Convert question to lowercase for case-insensitive matching
    question_lower = question.lower()
    
    # Define keywords for each category
    keywords = {
        "billing": ["bill", "payment", "refund", "charge", "subscription", "cancel", "price"],
        "technical_support": ["password", "reset", "locked", "outage", "down", "error", "bug", "login"],
        "product": ["warranty", "return", "compatibility", "feature", "specification", "damaged"]
    }
    
    # Find matching categories based on keywords
    matching_categories = {}
    for category, words in keywords.items():
        score = sum(1 for word in words if word in question_lower)
        if score > 0:
            matching_categories[category] = score
    
    # If no matches found, return generic response
    if not matching_categories:
        logger.info("No matching SOP found, returning generic response")
        return {
            "category": "general",
            "sop_content": "I don't have a specific SOP for this query. Please gather the customer's specific issue details and transfer to the appropriate department."
        }
    
    # Get the category with highest match score
    best_category = max(matching_categories.items(), key=lambda x: x[1])[0]
    logger.info(f"Best matching category: {best_category}")
    
    # Find the best matching SOP in the category
    sops_in_category = sop_database[best_category]
    best_sop = None
    best_score = 0
    
    for sop_key, sop_content in sops_in_category.items():
        # Calculate simple relevance score by counting keyword occurrences
        score = sum(1 for word in sop_key.split("_") if word in question_lower)
        if score > best_score:
            best_score = score
            best_sop = sop_key
    
    # If we couldn't determine a specific SOP, return all SOPs for the category
    if not best_sop or best_score == 0:
        logger.info(f"No specific SOP found in category, returning all SOPs for {best_category}")
        return {
            "category": best_category,
            "sop_list": list(sops_in_category.keys()),
            "sop_content": "Multiple SOPs may apply to this query. Please select the most relevant one."
        }
    
    # Return the best matching SOP
    logger.info(f"Best matching SOP: {best_sop}")
    return {
        "category": best_category,
        "sop_key": best_sop,
        "sop_content": sops_in_category[best_sop]
    }


@mcp.tool()
def get_specific_sop(category: str, sop_key: str) -> Dict:
    """
    Retrieves a specific SOP by category and key.
    
    Args:
        category (str): The SOP category (e.g., "billing", "technical_support", "product")
        sop_key (str): The specific SOP key within the category
        
    Returns:
        Dict: A dictionary containing the SOP information
    """
    logger.info(f"Retrieving specific SOP: {category}/{sop_key}")
    
    if category not in sop_database:
        logger.warning(f"Category {category} not found")
        return {
            "error": "Category not found",
            "available_categories": list(sop_database.keys())
        }
    
    if sop_key not in sop_database[category]:
        logger.warning(f"SOP key {sop_key} not found in category {category}")
        return {
            "error": "SOP key not found",
            "available_sops": list(sop_database[category].keys())
        }
    
    return {
        "category": category,
        "sop_key": sop_key,
        "sop_content": sop_database[category][sop_key]
    }


@mcp.tool()
def list_sop_categories() -> List[str]:
    """
    Lists all available SOP categories.
    
    Returns:
        List[str]: A list of available SOP categories
    """
    logger.info("Listing all SOP categories")
    return list(sop_database.keys())


@mcp.tool()
def add_sop_entry(category: str, sop_key: str, sop_content: str) -> Dict:
    """
    Adds a new SOP entry to the database.
    
    Args:
        category (str): The SOP category
        sop_key (str): The specific SOP key
        sop_content (str): The SOP content/instructions
        
    Returns:
        Dict: A dictionary with the result of the operation
    """
    logger.info(f"Adding new SOP entry: {category}/{sop_key}")
    
    # Create category if it doesn't exist
    if category not in sop_database:
        sop_database[category] = {}
    
    # Add or update the SOP
    sop_database[category][sop_key] = sop_content
    
    return {
        "status": "success",
        "message": f"Added SOP entry: {category}/{sop_key}",
        "category": category,
        "sop_key": sop_key
    }


if __name__ == "__main__":
    logger.info("Starting Customer Service SOP MCP server on port 8000")
    mcp.run(transport="streamable-http")

"""
StreamableHTTP Client Transport with AWS SigV4 Signing

This module extends the MCP StreamableHTTPTransport to add AWS SigV4 request signing
for authentication with MCP servers that authenticate using AWS IAM.
"""

from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from datetime import timedelta
from typing import Generator

import httpx
from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from botocore.credentials import Credentials
from mcp.client.streamable_http import (
    GetSessionIdCallback,
    StreamableHTTPTransport,
    streamablehttp_client,
)
from mcp.shared._httpx_utils import McpHttpClientFactory, create_mcp_http_client
from mcp.shared.message import SessionMessage


class SigV4HTTPXAuth(httpx.Auth):
    """HTTPX Auth class that signs requests with AWS SigV4."""

    def __init__(
        self,
        credentials: Credentials,
        service: str,
        region: str,
    ):
        self.credentials = credentials
        self.service = service
        self.region = region
        self.signer = SigV4Auth(credentials, service, region)

    def auth_flow(
        self, request: httpx.Request
    ) -> Generator[httpx.Request, httpx.Response, None]:
        """Signs the request with SigV4 and adds the signature to the request headers."""

        # Create an AWS request
        headers = dict(request.headers)
        # Header 'connection' = 'keep-alive' is not used in calculating the request
        # signature on the server-side, and results in a signature mismatch if included
        headers.pop("connection", None)  # Remove if present, ignore if not

        aws_request = AWSRequest(
            method=request.method,
            url=str(request.url),
            data=request.content,
            headers=headers,
        )

        # Sign the request with SigV4
        self.signer.add_auth(aws_request)

        # Add the signature header to the original request
        request.headers.update(dict(aws_request.headers))

        yield request


class StreamableHTTPTransportWithSigV4(StreamableHTTPTransport):
    """
    Streamable HTTP client transport with AWS SigV4 signing support.

    This transport enables communication with MCP servers that authenticate using AWS IAM,
    such as servers behind a Lambda function URL or API Gateway.
    """

    def __init__(
        self,
        url: str,
        credentials: Credentials,
        service: str,
        region: str,
        headers: dict[str, str] | None = None,
        timeout: float | timedelta = 30,
        sse_read_timeout: float | timedelta = 60 * 5,
    ) -> None:
        """Initialize the StreamableHTTP transport with SigV4 signing.

        Args:
            url: The endpoint URL.
            credentials: AWS credentials for signing.
            service: AWS service name (e.g., 'lambda').
            region: AWS region (e.g., 'us-east-1').
            headers: Optional headers to include in requests.
            timeout: HTTP timeout for regular operations.
            sse_read_timeout: Timeout for SSE read operations.
        """
        # Initialize parent class with SigV4 auth handler
        super().__init__(
            url=url,
            headers=headers,
            timeout=timeout,
            sse_read_timeout=sse_read_timeout,
            auth=SigV4HTTPXAuth(credentials, service, region),
        )

        self.credentials = credentials
        self.service = service
        self.region = region


@asynccontextmanager
async def streamablehttp_client_with_sigv4(
    url: str,
    credentials: Credentials,
    service: str,
    region: str,
    headers: dict[str, str] | None = None,
    timeout: float | timedelta = 30,
    sse_read_timeout: float | timedelta = 60 * 5,
    terminate_on_close: bool = True,
    httpx_client_factory: McpHttpClientFactory = create_mcp_http_client,
) -> AsyncGenerator[
    tuple[
        MemoryObjectReceiveStream[SessionMessage | Exception],
        MemoryObjectSendStream[SessionMessage],
        GetSessionIdCallback,
    ],
    None,
]:
    """
    Client transport for Streamable HTTP with SigV4 auth.

    This transport enables communication with MCP servers that authenticate using AWS IAM,
    such as servers behind a Lambda function URL or API Gateway.

    Yields:
        Tuple containing:
            - read_stream: Stream for reading messages from the server
            - write_stream: Stream for sending messages to the server
            - get_session_id_callback: Function to retrieve the current session ID
    """

    async with streamablehttp_client(
        url=url,
        headers=headers,
        timeout=timeout,
        sse_read_timeout=sse_read_timeout,
        terminate_on_close=terminate_on_close,
        httpx_client_factory=httpx_client_factory,
        auth=SigV4HTTPXAuth(credentials, service, region),
    ) as result:
        yield result

mcp>=1.10.0
boto3
bedrock-agentcore<=0.1.5
bedrock-agentcore-starter-toolkit==0.1.14
invoke_customer_service_sop_tools.py
import asyncio
import sys
import os
import logging
import boto3
from boto3.session import Session
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
from streamable_http_sigv4 import streamablehttp_client_with_sigv4

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


def create_streamable_http_transport_sigv4(
    mcp_url: str, service_name: str, region: str
):
    """
    Create a streamable HTTP transport with AWS SigV4 authentication.

    This function creates an MCP client transport that uses AWS Signature Version 4 (SigV4)
    to authenticate requests. This is necessary because standard MCP clients don't natively
    support AWS IAM authentication, and this bridges that gap.

    Args:
        mcp_url (str): The URL of the MCP gateway endpoint
        service_name (str): The AWS service name for SigV4 signing (typically "bedrock-agentcore")
        region (str): The AWS region where the gateway is deployed

    Returns:
        StreamableHTTPTransportWithSigV4: A transport instance configured for SigV4 auth
    """
    # Get AWS credentials from the current boto3 session
    # These credentials will be used to sign requests with SigV4
    session = boto3.Session()
    credentials = session.get_credentials()

    # Create and return the custom transport with SigV4 signing capability
    return streamablehttp_client_with_sigv4(
        url=mcp_url,
        credentials=credentials,
        service=service_name,
        region=region,
    )


def get_full_tools_list(client):
    """
    Retrieve the complete list of tools from an MCP client, handling pagination.

    MCP servers may return tools in paginated responses. This function handles the
    pagination automatically and returns all available tools in a single list.

    Args:
        client: An MCP client instance (from strands.tools.mcp.mcp_client.MCPClient)

    Returns:
        list: A complete list of all tools available from the MCP server
    """
    more_tools = True
    tools = []
    pagination_token = None

    # Loop until we've fetched all pages
    while more_tools:
        tmp_tools = client.list_tools_sync(pagination_token=pagination_token)

        tools.extend(tmp_tools)

        # Check if there are more pages to fetch
        if tmp_tools.pagination_token is None:
            # No more pages - we're done
            more_tools = False
        else:
            # More pages exist - prepare to fetch the next one
            more_tools = True
            pagination_token = tmp_tools.pagination_token

    return tools


async def main():
    boto_session = Session()
    region = boto_session.region_name
    print(f"Using AWS region: {region}")

    # Check if SSM parameter name is provided as command line argument
    if len(sys.argv) > 1:
        param_name = sys.argv[1]
    else:
        param_name = "/mcp_server/customer_service_sop/agent_arn"
    
    print(f"Using SSM parameter: {param_name}")
    
    ssm_client = boto3.client("ssm", region_name=region)

    agent_arn_response = ssm_client.get_parameter(
        Name=param_name
    )
    agent_arn = agent_arn_response["Parameter"]["Value"]
    print(f"Retrieved Agent ARN: {agent_arn}")

    if not agent_arn:
        print("❌ Error: Agent ARN not found")
        sys.exit(1)

    encoded_arn = agent_arn.replace(":", "%3A").replace("/", "%2F")
    mcp_url = f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"

    try:
        async with create_streamable_http_transport_sigv4(
                mcp_url=mcp_url, service_name="bedrock-agentcore", region=region
        ) as (
            read_stream,
            write_stream,
            _,
        ):
            async with ClientSession(read_stream, write_stream) as session:
                print("\n🔄 Initializing MCP session...")
                await session.initialize()
                print("✓ MCP session initialized")

                print("\n🔄 Listing available tools...")
                tool_result = await session.list_tools()

                print("\n📋 Available Customer Service SOP Tools:")
                print("=" * 50)
                for tool in tool_result.tools:
                    print(f"🔧 {tool.name}: {tool.description}")

                print("\n🧪 Testing Customer Service SOP Tools:")
                print("=" * 50)

                # Test 1: List SOP categories
                try:
                    print("\n📊 Testing list_sop_categories()...")
                    categories_result = await session.call_tool(
                        name="list_sop_categories", arguments={}
                    )
                    print(f"   Result: {categories_result.content[0].text}")
                except Exception as e:
                    print(f"   Error: {e}")
                
                # Test 2: Get SOP for a billing question
                try:
                    print("\n💳 Testing get_sop_for_question(billing question)...")
                    question = "I want to know the refund policy for my recent purchase"
                    billing_result = await session.call_tool(
                        name="get_sop_for_question", arguments={"question": question}
                    )
                    print(f"   Question: {question}")
                    print(f"   Result: {billing_result.content[0].text}")
                except Exception as e:
                    print(f"   Error: {e}")
                
                # Test 3: Get SOP for a technical support question
                try:
                    print("\n🔒 Testing get_sop_for_question(technical question)...")
                    question = "My account is locked and I can't log in"
                    tech_result = await session.call_tool(
                        name="get_sop_for_question", arguments={"question": question}
                    )
                    print(f"   Question: {question}")
                    print(f"   Result: {tech_result.content[0].text}")
                except Exception as e:
                    print(f"   Error: {e}")
                
                # Test 4: Get specific SOP by category and key
                try:
                    print("\n📝 Testing get_specific_sop()...")
                    category = "product"
                    sop_key = "warranty_information"
                    specific_result = await session.call_tool(
                        name="get_specific_sop", 
                        arguments={"category": category, "sop_key": sop_key}
                    )
                    print(f"   Category: {category}, SOP Key: {sop_key}")
                    print(f"   Result: {specific_result.content[0].text}")
                except Exception as e:
                    print(f"   Error: {e}")
                
                # Test 5: Add a new SOP entry
                try:
                    print("\n➕ Testing add_sop_entry()...")
                    new_sop_result = await session.call_tool(
                        name="add_sop_entry", 
                        arguments={
                            "category": "shipping", 
                            "sop_key": "international_shipping", 
                            "sop_content": "For international shipping inquiries, please check that the destination country is on our approved list. Standard shipping takes 7-14 business days. Express shipping (where available) takes 3-5 business days. All international orders over $100 receive free standard shipping."
                        }
                    )
                    print(f"   Result: {new_sop_result.content[0].text}")
                    
                    # Verify the newly added SOP by retrieving it
                    print("\n🔍 Verifying newly added SOP...")
                    verify_result = await session.call_tool(
                        name="get_specific_sop", 
                        arguments={"category": "shipping", "sop_key": "international_shipping"}
                    )
                    print(f"   Result: {verify_result.content[0].text}")
                except Exception as e:
                    print(f"   Error: {e}")

                print("\n✅ Customer Service SOP tool testing completed!")

    except Exception as e:
        print(f"❌ Error connecting to MCP server: {e}")
        import traceback
        print("\n🔍 Full error traceback:")
        traceback.print_exc()
        sys.exit(1)


if __name__ == "__main__":
    asyncio.run(main())

!pip install -U -r requirements.txt 
from bedrock_agentcore_starter_toolkit import Runtime
from bedrock_agentcore_starter_toolkit.operations.runtime import destroy_bedrock_agentcore
from boto3.session import Session
from pathlib import Path
import os
boto_session = Session()
region = boto_session.region_name
agentcore_control_client = boto_session.client("bedrock-agentcore-control", region_name=region)
ssm_client = boto_session.client('ssm', region_name=region)
tool_name = "customer_service_sop"

print(f"Using AWS region: {region}")

required_files = ["customer_service_sop_server.py", "requirements.txt"]
for file in required_files:
    if not os.path.exists(file):
        raise FileNotFoundError(f"Required file {file} not found")
print("All required files found ✓")

agentcore_runtime = Runtime()

print("Configuring AgentCore Runtime...")
response = agentcore_runtime.configure(
    entrypoint="customer_service_sop_server.py",
    auto_create_execution_role=True,
    auto_create_ecr=True,
    requirements_file="requirements.txt",
    region=region,
    protocol="MCP",
    agent_name=tool_name,
)
print("Configuration completed ✓")

print("Launching Customer Service SOP MCP server to AgentCore Runtime...")
print("This may take several minutes...")
launch_result = agentcore_runtime.launch()
print("Launch completed ✓")
print(f"Agent ARN: {launch_result.agent_arn}")
print(f"Agent ID: {launch_result.agent_id}")

agent_arn_response = ssm_client.put_parameter(
    Name='/mcp_server/customer_service_sop/agent_arn',
    Value=launch_result.agent_arn,
    Type='String',
    Description='Agent ARN for Customer Service SOP MCP server',
    Overwrite=True
)
print("✓ Agent ARN stored in Parameter Store")

print("\nConfiguration stored successfully!")
print(f"Agent ARN: {launch_result.agent_arn}")

!python invoke_customer_service_sop_tools.py

python invoke_customer_service_sop_tools.py
Using AWS region: us-west-2
Using SSM parameter: /mcp_server/customer_service_sop/agent_arn
2025-10-28 17:05:33,819 - botocore.credentials - INFO - Found credentials in shared credentials file: ~/.aws/credentials
Retrieved Agent ARN: arn:aws:bedrock-agentcore:us-west-2:310850127430:runtime/customer_service_sop-QIAcnREytM
2025-10-28 17:05:35,223 - botocore.credentials - INFO - Found credentials in shared credentials file: ~/.aws/credentials

🔄 Initializing MCP session...
2025-10-28 17:05:36,845 - httpx - INFO - HTTP Request: POST https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 200 OK"
2025-10-28 17:05:36,846 - mcp.client.streamable_http - INFO - Received session ID: 59915786-4fca-4c1c-9c76-f7190d420eb6
2025-10-28 17:05:36,847 - mcp.client.streamable_http - INFO - Negotiated protocol version: 2025-06-18
✓ MCP session initialized

🔄 Listing available tools...
2025-10-28 17:05:37,992 - httpx - INFO - HTTP Request: GET https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 404 Not Found"
2025-10-28 17:05:38,105 - httpx - INFO - HTTP Request: POST https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 202 Accepted"
2025-10-28 17:05:39,362 - httpx - INFO - HTTP Request: POST https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 200 OK"

📋 Available Customer Service SOP Tools:
🔧 get_sop_for_question:
Retrieves the appropriate Standard Operating Procedure (SOP) based on customer's question.

Args:
question (str): The customer's question or issue description

Returns:
Dict: A dictionary containing the SOP information

🔧 get_specific_sop:
Retrieves a specific SOP by category and key.

Args:
category (str): The SOP category (e.g., "billing", "technical_support", "product")
sop_key (str): The specific SOP key within the category

Returns:
Dict: A dictionary containing the SOP information

🔧 list_sop_categories:
Lists all available SOP categories.

Returns:
List[str]: A list of available SOP categories

🔧 add_sop_entry:
Adds a new SOP entry to the database.

Args:
category (str): The SOP category
sop_key (str): The specific SOP key
sop_content (str): The SOP content/instructions

Returns:
Dict: A dictionary with the result of the operation


🧪 Testing Customer Service SOP Tools:

📊 Testing list_sop_categories()...
2025-10-28 17:05:40,567 - httpx - INFO - HTTP Request: POST https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 200 OK"
Result: billing

💳 Testing get_sop_for_question(billing question)...
2025-10-28 17:05:41,857 - httpx - INFO - HTTP Request: POST https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 200 OK"
Question: I want to know the refund policy for my recent purchase
Result: {
"category": "billing",
"sop_key": "refund_policy",
"sop_content": "Customers are eligible for refunds within 30 days of purchase with valid receipt. Process through the billing system with approval for amounts over $50."
}

🔒 Testing get_sop_for_question(technical question)...
2025-10-28 17:05:43,219 - httpx - INFO - HTTP Request: POST https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 200 OK"
Question: My account is locked and I can't log in
Result: {
"category": "technical_support",
"sop_key": "account_lockout",
"sop_content": "Check lockout reason in security logs. For suspicious activity, escalate to security team. For failed login attempts, verify identity and unlock with 24-hour monitoring."
}

📝 Testing get_specific_sop()...
2025-10-28 17:05:44,559 - httpx - INFO - HTTP Request: POST https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 200 OK"
Category: product, SOP Key: warranty_information
Result: {
"category": "product",
"sop_key": "warranty_information",
"sop_content": "Standard warranty is 12 months from purchase date. Premium products have 24-month coverage. Warranty covers manufacturing defects but not physical damage."
}

➕ Testing add_sop_entry()...
2025-10-28 17:05:45,837 - httpx - INFO - HTTP Request: POST https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 200 OK"
Result: {
"status": "success",
"message": "Added SOP entry: shipping/international_shipping",
"category": "shipping",
"sop_key": "international_shipping"
}

🔍 Verifying newly added SOP...
2025-10-28 17:05:47,157 - httpx - INFO - HTTP Request: POST https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A310850127430%3Aruntime%2Fcustomer_service_sop-QIAcnREytM/invocations?qualifier=DEFAULT "HTTP/1.1 200 OK"
Result: {
"category": "shipping",
"sop_key": "international_shipping",
"sop_content": "For international shipping inquiries, please check that the destination country is on our approved list. Standard shipping takes 7-14 business days. Express shipping (where available) takes 3-5 business days. All international orders over $100 receive free standard shipping."
}

✅ Customer Service SOP tool testing completed!