feat(elasticache-mcp-server): connection_type for serverless public endpoints (#4867)

* feat(elasticache-mcp-server): connection_type for serverless public endpoints

Amazon ElastiCache Serverless for Valkey 9.0+ supports a 'public' connection
type: an internet-reachable endpoint with IAM authentication over TLS 1.3 and
no VPC, subnets, or security groups.

- create-serverless-cache: new optional connection_type ('vpc' default or
  'public'), validated by the request model and forwarded as ConnectionType
- Jump-host tools (create/connect/get-ssh-tunnel-command) return a clear error
  for caches with a public endpoint and point to the Valkey MCP server
- boto3 floor raised to 1.43.105, the first release with ConnectionType
- README tool list and prerequisites, CHANGELOG, unit tests

* fix(elasticache-mcp-server): address review feedback

- README: document the elasticache:ConnectionType condition key for denying
  public cache creation instead of restating the boto3 floor
- Jump-host guard: point at the Valkey MCP server connection guide instead of
  a single env var that is not sufficient on its own
- create docstring: major_engine_version '9' or later

---------

Co-authored-by: mwaisaida-13 <aidams@amazon.com>

Author: mwaisaida-13

src/elasticache-mcp-server/CHANGELOG.md

@@ -9,4 +9,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `connection_type` parameter on `create-serverless-cache` to create ElastiCache Serverless caches with a public endpoint (`'public'`) or a VPC endpoint (`'vpc'`, default).
+- Jump-host tools now return a clear error for caches with a public endpoint, since those are reached directly over the internet with IAM authentication.
 - Initial project setup

src/elasticache-mcp-server/README.md

@@ -5,13 +5,15 @@ The official MCP Server for interacting with AWS ElastiCache control plane. In o
 ## Available MCP Tools
 
 ### Serverless Cache Operations
-- `create-serverless-cache` - Create a new ElastiCache serverless cache
+- `create-serverless-cache` - Create a new ElastiCache serverless cache. Pass `connection_type='public'` to create a Valkey 9+ cache with a public endpoint that is reachable over the internet without a VPC (IAM authentication over TLS 1.3, IAM user group required); the default `'vpc'` creates a VPC endpoint
 - `delete-serverless-cache` - Delete a serverless cache
 - `describe-serverless-caches` - Get information about serverless caches
 - `modify-serverless-cache` - Modify settings of a serverless cache
-- `connect-jump-host-serverless-cache` - Configure an EC2 instance as a jump host for serverless cache access
-- `create-jump-host-serverless-cache` - Create an EC2 jump host to access a serverless cache via SSH tunnel
-- `get-ssh-tunnel-command-serverless-cache` - Generate SSH tunnel command for serverless cache access
+- `connect-jump-host-serverless-cache` - Configure an EC2 instance as a jump host for access to a serverless cache with a VPC endpoint
+- `create-jump-host-serverless-cache` - Create an EC2 jump host to access a serverless cache with a VPC endpoint via SSH tunnel
+- `get-ssh-tunnel-command-serverless-cache` - Generate SSH tunnel command for a serverless cache with a VPC endpoint
+
+The jump-host tools apply only to caches with a VPC endpoint. For a cache with a public endpoint they return an error, because the cache is reached directly over the internet with IAM authentication.
 
 ### Replication Group Operations
 - `create-replication-group` - Create an Amazon ElastiCache replication group with specified configuration
@@ -73,6 +75,7 @@ All tools support an optional `region_name` parameter to specify which AWS regio
 2. Install Python using `uv python install 3.10`
 3. Set up AWS credentials with access to AWS services
    - Consider setting up Read-only permission if you don't want the LLM to modify any resources
+   - To keep the LLM from creating caches with a public endpoint, deny `elasticache:CreateServerlessCache` when the `elasticache:ConnectionType` condition key is `public`. See [Using condition keys](https://docs.aws.amazon.com/AmazonElastiCache/latest/dg/IAM.ConditionKeys.html#IAM.ExamplePolicies).
 
 ## Installation
 

src/elasticache-mcp-server/awslabs/elasticache_mcp_server/tools/serverless/connect.py

@@ -22,6 +22,29 @@
 from typing import Any, Dict, Optional, Tuple, Union
 
 
+def _ensure_vpc_cache(serverless_cache: Dict[str, Any], serverless_cache_name: str) -> None:
+    """Reject jump-host operations for serverless caches with ConnectionType 'public'.
+
+    Public-endpoint caches have no VPC endpoint, subnets, or security groups, so jump hosts
+    and SSH tunnels do not apply. A missing ConnectionType field is treated as 'vpc'.
+
+    Args:
+        serverless_cache (Dict[str, Any]): A ServerlessCache entry from describe_serverless_caches
+        serverless_cache_name (str): Name of the cache, used in the error message
+
+    Raises:
+        ValueError: If the cache has a public endpoint
+    """
+    if serverless_cache.get('ConnectionType') == 'public':
+        endpoint = serverless_cache.get('Endpoint', {}).get('Address', '<endpoint>')
+        raise ValueError(
+            f'Serverless cache {serverless_cache_name} has a public endpoint '
+            f'(ConnectionType=public), so a jump host or SSH tunnel is not needed and cannot '
+            f'be configured. Connect directly to {endpoint} over TLS 1.3 using IAM '
+            'authentication; see the ElastiCache connection guide of the Valkey MCP server.'
+        )
+
+
 async def _configure_security_groups(
     serverless_cache_name: str,
     instance_id: str,
@@ -51,6 +74,7 @@ async def _configure_security_groups(
     serverless_cache = elasticache_client.describe_serverless_caches(
         ServerlessCacheName=serverless_cache_name
     )['ServerlessCaches'][0]
+    _ensure_vpc_cache(serverless_cache, serverless_cache_name)
 
     # Get cache security groups
     cache_security_groups = serverless_cache['SecurityGroupIds']
@@ -148,6 +172,8 @@ async def connect_jump_host_serverless(
 ) -> Dict[str, Any]:
     """Configures an existing EC2 instance as a jump host to access an ElastiCache serverless cache.
 
+    Not applicable to serverless caches with a public endpoint; returns an error for them.
+
     Args:
         serverless_cache_name (str): Name of the ElastiCache serverless cache to connect to
         instance_id (str): ID of the EC2 instance to use as jump host
@@ -192,6 +218,8 @@ async def get_ssh_tunnel_command_serverless(
 ) -> Dict[str, Union[str, int]]:
     """Generates an SSH tunnel command to connect to an ElastiCache serverless cache through an EC2 jump host.
 
+    Not applicable to serverless caches with a public endpoint; returns an error for them.
+
     Args:
         serverless_cache_name (str): Name of the ElastiCache serverless cache to connect to
         instance_id (str): ID of the EC2 instance to use as jump host
@@ -235,6 +263,7 @@ async def get_ssh_tunnel_command_serverless(
         serverless_cache = elasticache_client.describe_serverless_caches(
             ServerlessCacheName=serverless_cache_name
         )['ServerlessCaches'][0]
+        _ensure_vpc_cache(serverless_cache, serverless_cache_name)
 
         # Get cache endpoint and port
         cache_endpoint = serverless_cache['Endpoint']['Address']
@@ -281,6 +310,8 @@ async def create_jump_host_serverless(
 ) -> Dict[str, Any]:
     """Creates an EC2 jump host instance to access an ElastiCache serverless cache via SSH tunnel.
 
+    Not applicable to serverless caches with a public endpoint; returns an error for them.
+
     Args:
         serverless_cache_name (str): Name of the ElastiCache serverless cache to connect to
         key_name (str): Name of the EC2 key pair to use for SSH access
@@ -324,6 +355,7 @@ async def create_jump_host_serverless(
         serverless_cache = elasticache_client.describe_serverless_caches(
             ServerlessCacheName=serverless_cache_name
         )['ServerlessCaches'][0]
+        _ensure_vpc_cache(serverless_cache, serverless_cache_name)
 
         # Get cache security groups
         cache_security_groups = serverless_cache['SecurityGroupIds']

src/elasticache-mcp-server/awslabs/elasticache_mcp_server/tools/serverless/create.py

@@ -70,6 +70,8 @@ async def create_serverless_cache(request: CreateServerlessCacheRequest) -> Dict
         user_group_id (Optional[str]): ID of the user group to associate with the cache.
         snapshot_retention_limit (Optional[int]): Number of days for which ElastiCache retains automatic snapshots.
         daily_snapshot_time (Optional[str]): Time range (in UTC) when daily snapshots are taken (e.g., '04:00-05:00').
+        connection_type (Optional[str]): 'vpc' (default) or 'public'. For 'public', pass engine='valkey',
+            major_engine_version='9' or later, and user_group_id; omit subnet_ids and security_group_ids.
 
     Returns:
         Dict containing information about the created serverless cache.
@@ -113,6 +115,7 @@ async def create_serverless_cache(request: CreateServerlessCacheRequest) -> Dict
         ('MajorEngineVersion', request.major_engine_version),
         ('UserGroupId', request.user_group_id),
         ('DailySnapshotTime', request.daily_snapshot_time),
+        ('ConnectionType', request.connection_type),
     ]:
         if value:
             create_request[param_name] = str(value)

src/elasticache-mcp-server/awslabs/elasticache_mcp_server/tools/serverless/models.py

@@ -103,6 +103,18 @@ class CreateServerlessCacheRequest(BaseModel):
         None,
         description="Time range (in UTC) when daily snapshots are taken (e.g., '04:00-05:00')",
     )
+    connection_type: Optional[str] = Field(
+        None,
+        description=(
+            'The connection type for the serverless cache. Must be either vpc | public. Use vpc '
+            'to access the cache through a VPC endpoint, or public to access the cache over the '
+            'internet. If not specified, defaults to vpc. This value cannot be changed after the '
+            'serverless cache is created. Setting this to public requires Valkey 9 or above, a '
+            'user group in which every user uses IAM authentication (e.g. '
+            'default.iam-user-group), and clients that connect with IAM authentication over '
+            'TLS 1.3.'
+        ),
+    )
 
     @field_validator('daily_snapshot_time')
     def validate_snapshot_time(cls, v):
@@ -114,6 +126,13 @@ def validate_snapshot_time(cls, v):
                 raise ValueError('Invalid time range format. Must be in format HH:MM-HH:MM')
         return v
 
+    @field_validator('connection_type')
+    def validate_connection_type(cls, v):
+        """Validate that connection_type is 'vpc' or 'public'."""
+        if v is not None and v not in ('vpc', 'public'):
+            raise ValueError("connection_type must be 'vpc' or 'public'")
+        return v
+
     model_config = ConfigDict(validate_by_name=True, arbitrary_types_allowed=True)
 
 

src/elasticache-mcp-server/pyproject.toml

@@ -8,7 +8,7 @@ dependencies = [
     "loguru>=0.7.0",
     "mcp[cli]>=2.0.0,<3.0.0",
     "pydantic>=2.10.6",
-    "boto3>=1.34.0",
+    "boto3>=1.43.105",
 ]
 license = {text = "Apache-2.0"}
 license-files = ["LICENSE", "NOTICE" ]

src/elasticache-mcp-server/tests/tools/serverless/test_connect.py

@@ -794,3 +794,145 @@ async def test_create_jump_host_serverless_default_vpc_map_public_ip():
         assert result['SubnetId'] == 'subnet-1234'
         assert result['SecurityGroupId'] == 'sg-1234'
         assert result['ServerlessCacheName'] == 'cache-1'
+
+
+PUBLIC_CACHE = {
+    'ServerlessCacheName': 'public-cache',
+    'ConnectionType': 'public',
+    'Engine': 'valkey',
+    'MajorEngineVersion': '9',
+    'Endpoint': {
+        'Address': 'public-cache-x2e9hv.public.serverless.use1.cache.amazonaws.com',
+        'Port': 6379,
+    },
+}
+
+
+@pytest.mark.asyncio
+async def test_configure_security_groups_rejects_public_cache():
+    """A public-endpoint cache is rejected before any security group work."""
+    mock_ec2 = MagicMock()
+    mock_elasticache = MagicMock()
+    mock_elasticache.describe_serverless_caches.return_value = {'ServerlessCaches': [PUBLIC_CACHE]}
+
+    with pytest.raises(ValueError) as exc_info:
+        await _configure_security_groups('public-cache', 'i-1234', mock_ec2, mock_elasticache)
+
+    assert 'public endpoint' in str(exc_info.value)
+    assert PUBLIC_CACHE['Endpoint']['Address'] in str(exc_info.value)
+    mock_ec2.describe_subnets.assert_not_called()
+    mock_ec2.authorize_security_group_ingress.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_connect_jump_host_serverless_rejects_public_cache():
+    """connect-jump-host returns an error for a public-endpoint cache."""
+    mock_ec2 = MagicMock()
+    mock_elasticache = MagicMock()
+    mock_elasticache.describe_serverless_caches.return_value = {'ServerlessCaches': [PUBLIC_CACHE]}
+
+    with (
+        patch(
+            'awslabs.elasticache_mcp_server.common.connection.EC2ConnectionManager.get_connection',
+            return_value=mock_ec2,
+        ),
+        patch(
+            'awslabs.elasticache_mcp_server.common.connection.ElastiCacheConnectionManager.get_connection',
+            return_value=mock_elasticache,
+        ),
+    ):
+        result = await connect_jump_host_serverless('public-cache', 'i-1234')
+
+    assert 'error' in result
+    assert 'public endpoint' in result['error']
+    mock_ec2.authorize_security_group_ingress.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_get_ssh_tunnel_command_serverless_rejects_public_cache():
+    """get-ssh-tunnel-command returns an error for a public-endpoint cache."""
+    mock_ec2 = MagicMock()
+    mock_elasticache = MagicMock()
+    mock_ec2.describe_instances.return_value = {
+        'Reservations': [
+            {
+                'Instances': [
+                    {
+                        'KeyName': 'my-key',
+                        'PublicDnsName': 'ec2-1-2-3-4.compute-1.amazonaws.com',
+                        'Platform': '',
+                    }
+                ]
+            }
+        ]
+    }
+    mock_elasticache.describe_serverless_caches.return_value = {'ServerlessCaches': [PUBLIC_CACHE]}
+
+    with (
+        patch(
+            'awslabs.elasticache_mcp_server.common.connection.EC2ConnectionManager.get_connection',
+            return_value=mock_ec2,
+        ),
+        patch(
+            'awslabs.elasticache_mcp_server.common.connection.ElastiCacheConnectionManager.get_connection',
+            return_value=mock_elasticache,
+        ),
+    ):
+        result = await get_ssh_tunnel_command_serverless('public-cache', 'i-1234')
+
+    assert 'error' in result
+    assert 'public endpoint' in result['error']
+    assert 'command' not in result
+
+
+@pytest.mark.asyncio
+async def test_create_jump_host_serverless_rejects_public_cache_before_launch():
+    """create-jump-host returns an error for a public-endpoint cache and launches nothing."""
+    mock_ec2 = MagicMock()
+    mock_elasticache = MagicMock()
+    mock_ec2.describe_key_pairs.return_value = {'KeyPairs': [{'KeyName': 'my-key'}]}
+    mock_elasticache.describe_serverless_caches.return_value = {'ServerlessCaches': [PUBLIC_CACHE]}
+
+    with (
+        patch(
+            'awslabs.elasticache_mcp_server.common.connection.EC2ConnectionManager.get_connection',
+            return_value=mock_ec2,
+        ),
+        patch(
+            'awslabs.elasticache_mcp_server.common.connection.ElastiCacheConnectionManager.get_connection',
+            return_value=mock_elasticache,
+        ),
+    ):
+        result = await create_jump_host_serverless('public-cache', 'my-key')
+
+    assert 'error' in result
+    assert 'public endpoint' in result['error']
+    mock_ec2.run_instances.assert_not_called()
+    mock_ec2.describe_subnets.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_configure_security_groups_vpc_cache_without_connection_type_field():
+    """Caches created before ConnectionType existed are treated as VPC caches."""
+    mock_ec2 = MagicMock()
+    mock_elasticache = MagicMock()
+    mock_elasticache.describe_serverless_caches.return_value = {
+        'ServerlessCaches': [
+            {'SecurityGroupIds': ['sg-cache'], 'SubnetIds': ['subnet-1234'], 'Engine': 'valkey'}
+        ]
+    }
+    mock_ec2.describe_subnets.return_value = {'Subnets': [{'VpcId': 'vpc-1234'}]}
+    mock_ec2.describe_instances.return_value = {
+        'Reservations': [
+            {'Instances': [{'VpcId': 'vpc-1234', 'SecurityGroups': [{'GroupId': 'sg-instance'}]}]}
+        ]
+    }
+    mock_ec2.describe_security_groups.return_value = {'SecurityGroups': [{'IpPermissions': []}]}
+
+    success, vpc_id, port = await _configure_security_groups(
+        'cache-1', 'i-1234', mock_ec2, mock_elasticache
+    )
+
+    assert success is True
+    assert vpc_id == 'vpc-1234'
+    assert port == 6379