test: add integration tests for AWS MCP

Author: detti456

examples/mcp-client/langchain/main.py

@@ -26,12 +26,6 @@
    - MCP_REGION: AWS region where the MCP server is hosted (e.g., "us-west-2")
 3. Run: `uv run main.py`
 
-Example .env file:
-==================
-MCP_SERVER_URL=https://example.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp
-MCP_SERVER_AWS_SERVICE=bedrock-agentcore
-MCP_SERVER_REGION=us-west-2
-
 Example .env file:
 ==================
 MCP_SERVER_URL=https://example.gateway.bedrock-agentcore.us-west-2.amazonaws.com/mcp

mcp_proxy_for_aws/middleware/initialize_middleware.py

@@ -9,7 +9,7 @@
 
 
 class InitializeMiddleware(Middleware):
-    """Intecept MCP initialize request and initialize the proxy client."""
+    """Intercept MCP initialize request and initialize the proxy client."""
 
     def __init__(self, client_factory: AWSMCPProxyClientFactory) -> None:
         """Create a middleware with client factory."""

tests/integ/conftest.py

@@ -90,3 +90,15 @@ def _build_endpoint_environment_remote_configuration():
         endpoint=remote_endpoint_url,
         region_name=region_name,
     )
+
+
+@pytest_asyncio.fixture(loop_scope="module", scope="module")
+async def aws_mcp_client():
+    """Create MCP Client for AWS MCP Server."""
+    client = build_mcp_client(
+        endpoint="https://aws-mcp.us-east-1.api.aws/mcp",
+        region_name="us-east-1",
+    )
+
+    async with client:
+        yield client

tests/integ/mcp/simple_mcp_client.py

@@ -27,7 +27,7 @@ def build_mcp_client(
             **_build_mcp_config(endpoint=endpoint, region_name=region_name, metadata=metadata)
         ),
         elicitation_handler=_basic_elicitation_handler,
-        timeout=10.0,  # seconds
+        timeout=20.0,  # seconds
     )
 
 

tests/integ/test_aws_mcp_server.py

@@ -0,0 +1,138 @@
+"""Integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp."""
+
+import fastmcp
+import logging
+import pytest
+from fastmcp.client.client import CallToolResult
+
+from tests.integ.test_proxy_simple_mcp_server import get_text_content
+import json
+
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.asyncio(loop_scope="module")
+async def test_aws_mcp_ping(aws_mcp_client: fastmcp.Client):
+    """Test ping to AWS MCP Server."""
+    await aws_mcp_client.ping()
+
+@pytest.mark.asyncio(loop_scope="module")
+async def test_aws_mcp_list_tools(aws_mcp_client: fastmcp.Client):
+    """Test list tools from AWS MCP Server."""
+    tools = await aws_mcp_client.list_tools()
+
+    assert len(tools) > 0, f"AWS MCP Server should have tools (got {len(tools)})"
+
+
+def verify_response_content(response: CallToolResult):
+    """Verify that a tool call response is successful and contains text content.
+
+    Args:
+        response: The CallToolResult from an MCP tool call
+
+    Returns:
+        str: The extracted text content from the response
+
+    Raises:
+        AssertionError: If response indicates an error or has empty content
+    """
+    assert (
+        response.is_error is False
+    ), f"is_error returned true. Returned response body: {response}."
+    assert len(response.content) > 0, f"Empty result list in response. Response: {response}"
+
+    response_text = get_text_content(response)
+    assert len(response_text) > 0, f"Empty response text. Response: {response}"
+
+    return response_text
+
+def verify_json_response(response: CallToolResult):
+    """Verify that a tool call response is successful and contains valid JSON data.
+
+    Args:
+        response: The CallToolResult from an MCP tool call
+
+    Raises:
+        AssertionError: If response indicates an error, has empty content,
+                       contains invalid JSON, or JSON data is empty
+    """
+    response_text = verify_response_content(response)
+
+    # Verify response_text is valid JSON
+    try:
+        response_data = json.loads(response_text)
+    except json.JSONDecodeError:
+        raise AssertionError(f"Response text is not valid JSON. Response text: {response_text}")
+
+    assert len(response_data) > 0, f"Empty JSON content in response. Response: {response}"
+
+
+@pytest.mark.parametrize(
+    "tool_name,params",
+    [
+        ("aws___list_regions", {}),
+        ("aws___suggest_aws_commands", {"query": "how to list my lambda functions"}),
+        ("aws___search_documentation", {"search_phrase": "S3 bucket versioning"}),
+        (
+            "aws___recommend",
+            {"url": "bad_url"},
+        ),
+        (
+            "aws___read_documentation",
+            {"url": "https://docs.aws.amazon.com/lambda/latest/dg/lambda-invocation.html"},
+        ),
+        (
+            "aws___get_regional_availability",
+            {"resource_type": "cfn", "region": "us-east-1"},
+        ),
+        ("aws___call_aws", {"cli_command": "aws s3 ls", "max_results": 50}),
+    ],
+    ids=[
+        "list_regions",
+        "suggest_aws_commands",
+        "search_documentation",
+        "recommend",
+        "read_documentation",
+        "get_regional_availability",
+        "call_aws",
+    ],
+)
+@pytest.mark.asyncio(loop_scope="module")
+async def test_aws_mcp_tools(aws_mcp_client: fastmcp.Client, tool_name: str, params: dict):
+    """Test AWS MCP tools with minimal valid params."""
+    response = await aws_mcp_client.call_tool(tool_name, params)
+    verify_json_response(response)
+
+
+@pytest.mark.asyncio(loop_scope="module")
+async def test_aws_mcp_tools_retrieve_agent_sop(aws_mcp_client: fastmcp.Client):
+    """Test aws___retrieve_agent_sop by retrieving the list of available SOPs."""
+
+    # Step 1: Call retrieve_agent_sop with empty params to get list of available SOPs
+    list_sops_response = await aws_mcp_client.call_tool("aws___retrieve_agent_sop")
+
+    list_sops_response_text = verify_response_content(list_sops_response)
+
+    # Parse SOP names from text (format: "* sop_name : description")
+    sop_names = []
+    for line in list_sops_response_text.split("\n"):
+        line = line.strip()
+        if line.startswith("*") and ":" in line:
+            # Extract the SOP name between '*' and ':'
+            sop_name = line.split("*", 1)[1].split(":", 1)[0].strip()
+            if sop_name:
+                sop_names.append(sop_name)
+
+    assert (
+        len(sop_names) > 0
+    ), f"No SOPs found in response. Response: {list_sops_response_text[:200]}..."
+    logger.info(f"Found {len(sop_names)} SOPs: {sop_names}")
+
+    # Step 2: Test retrieving the first SOP
+    test_script = sop_names[0]
+    logger.info(f"Testing with SOP: {test_script}")
+
+    response = await aws_mcp_client.call_tool("aws___retrieve_agent_sop", {"sop_name": test_script})
+
+    verify_response_content(response)

tests/integ/test_aws_mcp_server_negative.py

@@ -0,0 +1,132 @@
+"""Negative integration tests for AWS MCP Server at https://aws-mcp.us-east-1.api.aws/mcp."""
+
+import fastmcp
+import logging
+import pytest
+import boto3
+from fastmcp.client import StdioTransport
+from datetime import datetime, timedelta
+
+logger = logging.getLogger(__name__)
+
+
+@pytest.mark.asyncio(loop_scope="module")
+async def test_nonexistent_tool(aws_mcp_client: fastmcp.Client):
+    """Test that calling a nonexistent tool raises an exception.
+
+    This test will:
+    - PASS when calling a nonexistent tool raises an exception
+    - FAIL if the nonexistent tool somehow succeeds
+    """
+    exception_raised = False
+    exception_message = None
+
+    try:
+        response = await aws_mcp_client.call_tool("aws___nonexistent_tool_12345", {})
+        logger.info(f"Unexpected success, response: {response}")
+    except Exception as e:
+        exception_raised = True
+        exception_message = str(e)
+        logger.info(f"Exception raised as expected: {type(e).__name__}: {exception_message}")
+
+    # Assert that an exception was raised
+    assert exception_raised, (
+        f"Expected exception when calling nonexistent tool 'aws___nonexistent_tool_12345', "
+        f"but call succeeded unexpectedly."
+    )
+
+    # Verify the exception mentions the tool not being found
+    error_message_lower = exception_message.lower()
+    tool_error_patterns = [
+        "not found",
+        "unknown",
+        "does not exist",
+        "invalid tool",
+        "tool",
+        "unknown tool",
+    ]
+
+    assert any(pattern in error_message_lower for pattern in tool_error_patterns), (
+        f"Exception was raised but doesn't appear to be about a missing tool. "
+        f"Expected one of {tool_error_patterns}, but got: {exception_message[:200]}"
+    )
+
+    logger.info(f"Test passed: Nonexistent tool correctly raised exception")
+
+
+@pytest.mark.asyncio(loop_scope="module")
+async def test_expired_credentials():
+    """Test that expired credentials are properly rejected.
+
+    This test uses real AWS credentials but modifies the session token to simulate
+    an expired token, which should result in an 'expired token' error message.
+
+    This test will:
+    - PASS when expired credentials are rejected with appropriate error
+    - FAIL if the modified credentials somehow work
+    """
+
+    # Get real credentials from boto3
+    session = boto3.Session()
+    creds = session.get_credentials()
+
+    # Use real access key and secret, but modify the token to simulate expiration by changing a few characters
+    expired_token = creds.token[:-20] + "EXPIRED_TOKEN_12345"
+
+    expired_client = fastmcp.Client(
+        StdioTransport(
+            command="mcp-proxy-for-aws",
+            args=[
+                "https://aws-mcp.us-east-1.api.aws/mcp",
+                "--log-level",
+                "DEBUG",
+                "--region",
+                "us-east-1",
+            ],
+            env={
+                "AWS_REGION": "us-east-1",
+                "AWS_ACCESS_KEY_ID": creds.access_key,
+                "AWS_SECRET_ACCESS_KEY": creds.secret_key,
+                "AWS_SESSION_TOKEN": expired_token,
+            },
+        ),
+        timeout=30.0,
+    )
+
+    exception_raised = False
+    exception_message = None
+
+    try:
+        async with expired_client:
+            response = await expired_client.call_tool("aws___list_regions")
+            logger.info(f"Tool call completed without exception: is_error={response.is_error}")
+    except Exception as e:
+        exception_raised = True
+        exception_message = str(e)
+        logger.info(f"Exception raised as expected: {type(e).__name__}: {exception_message}")
+
+    # Assert that an exception was raised (credentials are invalid)
+    assert exception_raised, (
+        f"Expected authentication exception with invalid credentials, " f"but tool call succeeded."
+    )
+
+    # Verify the exception is related to authentication/credentials
+    error_message_lower = exception_message.lower()
+    auth_error_patterns = [
+        "credential",
+        "authentication",
+        "authorization",
+        "access denied",
+        "unauthorized",
+        "invalid",
+        "expired",
+        "signature",
+        "401",
+    ]
+
+    assert any(pattern in error_message_lower for pattern in auth_error_patterns), (
+        f"Exception was raised but doesn't appear to be authentication-related. "
+        f"Expected one of {auth_error_patterns}, but got: {exception_message[:200]}"
+    )
+
+    logger.info(f"Test passed: Invalid credentials correctly rejected")