feat(dynamodb-mcp-server): Adding direct connection support to include RDS / self-hosted MySQL DB in the Source DB Analyzer tool

Author: kishorln

.secrets.baseline

@@ -215,18 +215,17 @@
         "filename": "src/dynamodb-mcp-server/README.md",
         "hashed_secret": "37b5ecd16fe6c599c85077c7992427df62b2ab71",
         "is_verified": false,
-        "line_number": 200,
+        "line_number": 223,
         "is_secret": false
       }
     ],
     "src/dynamodb-mcp-server/awslabs/dynamodb_mcp_server/database_analyzers.py": [
       {
         "type": "Secret Keyword",
         "filename": "src/dynamodb-mcp-server/awslabs/dynamodb_mcp_server/database_analyzers.py",
-        "hashed_secret": "38a2bae6275b4d868c488758d213827833cd8570",
+        "hashed_secret": "90181b7e929dd123c6fb2bfb6917e005e9ed6992",
         "is_verified": false,
-        "line_number": 107,
-        "is_secret": false
+        "line_number": 155
       }
     ],
     "src/dynamodb-mcp-server/tests/test_dynamodb_server.py": [
@@ -906,5 +905,5 @@
       }
     ]
   },
-  "generated_at": "2025-12-01T14:26:01Z"
+  "generated_at": "2025-12-05T13:59:20Z"
 }

src/dynamodb-mcp-server/README.md

@@ -14,7 +14,7 @@ The DynamoDB MCP server provides four tools for data modeling and validation:
 
   **Example invocation:** "Validate my DynamoDB data model"
 
-- `source_db_analyzer` - Analyzes existing MySQL/Aurora databases to extract schema structure, access patterns from Performance Schema, and generates timestamped analysis files for use with dynamodb_data_modeling. Requires AWS RDS Data API and credentials in Secrets Manager.
+- `source_db_analyzer` - Analyzes existing MySQL/Aurora databases to extract schema structure, access patterns from Performance Schema, and generates timestamped analysis files for use with dynamodb_data_modeling. Supports both AWS RDS Data API (Aurora MySQL) and direct connections (RDS/self-hosted MySQL).
 
   **Example invocation:** "Analyze my MySQL database and help me design a DynamoDB data model"
 
@@ -159,11 +159,24 @@ The tool automates the traditional manual validation process:
 
 The `source_db_analyzer` tool analyzes existing MySQL/Aurora databases to extract schema and access patterns for DynamoDB modeling. This is useful when migrating from relational databases.
 
+The tool supports two connection methods:
+- **AWS RDS Data API** (Aurora MySQL): Serverless connection using cluster ARN
+- **Direct Connection** (RDS/self-hosted MySQL): Traditional connection using hostname/port
+
 #### Prerequisites for MySQL Integration
 
+**For AWS RDS Data API (Aurora MySQL):**
 1. Aurora MySQL Cluster with credentials stored in AWS Secrets Manager
 2. Enable RDS Data API for your Aurora MySQL Cluster
-3. Enable Performance Schema for access pattern analysis (optional but recommended):
+3. AWS credentials with permissions to access RDS Data API and AWS Secrets Manager
+
+**For Direct Connection (RDS/self-hosted MySQL):**
+1. MySQL server (RDS, Aurora, or self-hosted) accessible from your environment
+2. Database credentials stored in AWS Secrets Manager
+3. AWS credentials with permissions to access AWS Secrets Manager
+
+**For both connection methods:**
+4. Enable Performance Schema for access pattern analysis (optional but recommended):
    - Set `performance_schema` parameter to 1 in your DB parameter group
    - Reboot the DB instance after changes
    - Verify with: `SHOW GLOBAL VARIABLES LIKE '%performance_schema'`
@@ -172,18 +185,28 @@ The `source_db_analyzer` tool analyzes existing MySQL/Aurora databases to extrac
      - `performance_schema_max_digest_length` - Maximum byte length per statement digest (default: 1024)
    - Without Performance Schema, analysis is based on information schema only
 
-4. AWS credentials with permissions to access RDS Data API and AWS Secrets Manager
-
 #### MySQL Environment Variables
 
 Add these environment variables to enable MySQL integration:
 
+**For AWS RDS Data API (Aurora MySQL):**
 - `MYSQL_CLUSTER_ARN`: Aurora MySQL cluster Resource ARN
 - `MYSQL_SECRET_ARN`: ARN of secret containing database credentials
 - `MYSQL_DATABASE`: Database name to analyze
 - `AWS_REGION`: AWS region of the Aurora MySQL cluster
+
+**For Direct Connection (RDS/self-hosted MySQL):**
+- `MYSQL_HOSTNAME`: MySQL server hostname or endpoint
+- `MYSQL_PORT`: MySQL server port (optional, default: 3306)
+- `MYSQL_SECRET_ARN`: ARN of secret containing database credentials
+- `MYSQL_DATABASE`: Database name to analyze
+- `AWS_REGION`: AWS region where Secrets Manager is located
+
+**Common options:**
 - `MYSQL_MAX_QUERY_RESULTS`: Maximum rows in analysis output files (optional, default: 500)
 
+**Note:** Explicit tool parameters take precedence over environment variables. Only one connection method (cluster ARN or hostname) should be specified.
+
 #### MCP Configuration with MySQL
 
 ```json

src/dynamodb-mcp-server/awslabs/dynamodb_mcp_server/database_analyzers.py

@@ -21,13 +21,17 @@
     get_schema_queries,
 )
 from awslabs.dynamodb_mcp_server.markdown_formatter import MarkdownFormatter
-from awslabs.mysql_mcp_server.server import DBConnection, DummyCtx
+from awslabs.mysql_mcp_server.connection.asyncmy_pool_connection import AsyncmyPoolConnection
+from awslabs.mysql_mcp_server.connection.rds_data_api_connection import RDSDataAPIConnection
+from awslabs.mysql_mcp_server.server import DummyCtx
 from awslabs.mysql_mcp_server.server import run_query as mysql_query
 from datetime import datetime
 from loguru import logger
 from typing import Any, Dict, List, Tuple
 
 
+DEFAULT_MYSQL_PORT = 3306
+DEFAULT_READONLY = True
 DEFAULT_ANALYSIS_DAYS = 30
 DEFAULT_MAX_QUERY_RESULTS = 500
 SECONDS_PER_DAY = 86400
@@ -63,11 +67,42 @@ def build_connection_params(source_db_type: str, **kwargs) -> Dict[str, Any]:
                 )
             output_dir = user_provided_dir
 
+            # Validate port parameter
+            port_value = kwargs.get('port') or os.getenv('MYSQL_PORT', str(DEFAULT_MYSQL_PORT))
+            port = int(port_value) if str(port_value).isdigit() else DEFAULT_MYSQL_PORT
+
+            # Determine connection method - explicit parameters override environment variables
+            explicit_cluster_arn = kwargs.get('aws_cluster_arn')
+            explicit_hostname = kwargs.get('hostname')
+            if explicit_cluster_arn:
+                cluster_arn = explicit_cluster_arn
+                hostname = None
+            elif explicit_hostname:
+                cluster_arn = None
+                hostname = explicit_hostname
+            else:
+                # Fall back to environment variables
+                env_cluster_arn = os.getenv('MYSQL_CLUSTER_ARN')
+                env_hostname = os.getenv('MYSQL_HOSTNAME')
+
+                # Apply same exclusion to env vars as above
+                if env_cluster_arn:
+                    cluster_arn = env_cluster_arn
+                    hostname = None
+                elif env_hostname:
+                    cluster_arn = None
+                    hostname = env_hostname
+                else:
+                    cluster_arn = None
+                    hostname = None
+
             return {
-                'cluster_arn': kwargs.get('aws_cluster_arn') or os.getenv('MYSQL_CLUSTER_ARN'),
+                'cluster_arn': cluster_arn,
                 'secret_arn': kwargs.get('aws_secret_arn') or os.getenv('MYSQL_SECRET_ARN'),
                 'database': kwargs.get('database_name') or os.getenv('MYSQL_DATABASE'),
                 'region': kwargs.get('aws_region') or os.getenv('AWS_REGION'),
+                'hostname': hostname,
+                'port': port,
                 'max_results': kwargs.get('max_query_results')
                 or int(os.getenv('MYSQL_MAX_QUERY_RESULTS', str(DEFAULT_MAX_QUERY_RESULTS))),
                 'pattern_analysis_days': kwargs.get(
@@ -91,23 +126,37 @@ def validate_connection_params(
             Tuple of (missing_params, param_descriptions)
         """
         if source_db_type == 'mysql':
-            required_params = ['cluster_arn', 'secret_arn', 'database', 'region']
-            missing_params = [
-                param
-                for param in required_params
-                if not connection_params.get(param)
-                or (
-                    isinstance(connection_params[param], str)
-                    and connection_params[param].strip() == ''
+            missing_params = []
+            param_descriptions = {}
+            cluster_arn = connection_params.get('cluster_arn')
+            hostname = connection_params.get('hostname')
+
+            # Check for either RDS Data API or direct connection parameters
+            has_rds_data_api = bool(isinstance(cluster_arn, str) and cluster_arn.strip())
+            has_direct_connection = bool(isinstance(hostname, str) and hostname.strip())
+
+            # Check that we have a connection method
+            if not has_rds_data_api and not has_direct_connection:
+                missing_params.append('cluster_arn OR hostname')
+                param_descriptions['cluster_arn OR hostname'] = (
+                    'Required: Either aws_cluster_arn (Aurora MySQL cluster ARN for RDS Data API) OR hostname (MySQL server hostname for direct connection)'
                 )
-            ]
 
-            param_descriptions = {
-                'cluster_arn': 'AWS cluster ARN',
-                'secret_arn': 'AWS secret ARN',
-                'database': 'Database name',
-                'region': 'AWS region',
-            }
+            # Check common required parameters
+            common_required_params = ['secret_arn', 'database', 'region']
+            for param in common_required_params:
+                if not connection_params.get(param) or (
+                    isinstance(connection_params[param], str)
+                    and connection_params[param].strip() == ''
+                ):
+                    missing_params.append(param)
+            param_descriptions.update(
+                {
+                    'secret_arn': 'Secrets Manager secret ARN containing DB credentials',
+                    'database': 'Database name to analyze',
+                    'region': 'AWS region where your database instance and Secrets Manager are located',
+                }
+            )
             return missing_params, param_descriptions
         return [], {}
 
@@ -232,15 +281,17 @@ class MySQLAnalyzer(DatabaseAnalyzer):
     def is_performance_schema_enabled(result):
         """Check if MySQL performance schema is enabled from query result."""
         if result and len(result) > 0:
-            performance_schema_value = str(
-                result[0].get('', '0')
-            )  # Key is empty string by mysql package design, so checking only value here
+            # MySQL MCP server uses col['label'] for column names, creating {"@@performance_schema": "1"}
+            # Reference: https://github.com/awslabs/mcp/pull/1361
+            performance_schema_value = str(result[0].get('@@performance_schema', '0'))
             return performance_schema_value == '1'
         return False
 
     def __init__(self, connection_params):
         """Initialize MySQL analyzer with connection parameters."""
-        self.cluster_arn = connection_params['cluster_arn']
+        self.cluster_arn = connection_params.get('cluster_arn')
+        self.hostname = connection_params.get('hostname')
+        self.port = connection_params.get('port', 3306)
         self.secret_arn = connection_params['secret_arn']
         self.database = connection_params['database']
         self.region = connection_params['region']
@@ -250,10 +301,27 @@ def __init__(self, connection_params):
     async def _run_query(self, sql, query_parameters=None):
         """Internal method to run SQL queries against MySQL database."""
         try:
-            # Create a new connection with current parameters
-            db_connection = DBConnection(
-                self.cluster_arn, self.secret_arn, self.database, self.region, True
-            )
+            # Create appropriate connection type based on available parameters
+            if self.cluster_arn:
+                # Use RDS Data API connection
+                db_connection = RDSDataAPIConnection(
+                    cluster_arn=self.cluster_arn,
+                    secret_arn=self.secret_arn,
+                    database=self.database,
+                    region=self.region,
+                    readonly=DEFAULT_READONLY,
+                )
+            else:
+                # Use direct asyncmy connection
+                db_connection = AsyncmyPoolConnection(
+                    hostname=self.hostname,
+                    port=self.port,
+                    database=self.database,
+                    readonly=DEFAULT_READONLY,
+                    secret_arn=self.secret_arn,
+                    region=self.region,
+                )
+
             # Pass connection parameter directly to mysql_query
             result = await mysql_query(sql, DummyCtx(), db_connection, query_parameters)
             return result

src/dynamodb-mcp-server/awslabs/dynamodb_mcp_server/server.py

@@ -124,7 +124,16 @@ async def source_db_analyzer(
         ge=1,
     ),
     aws_cluster_arn: Optional[str] = Field(
-        default=None, description='AWS cluster ARN (overrides MYSQL_CLUSTER_ARN env var)'
+        default=None,
+        description='AWS cluster ARN for RDS Data API connection (overrides MYSQL_CLUSTER_ARN env var)',
+    ),
+    hostname: Optional[str] = Field(
+        default=None,
+        description='Database hostname for direct connection (overrides MYSQL_HOSTNAME env var)',
+    ),
+    port: Optional[int] = Field(
+        default=None,
+        description='Database port for direct connection (overrides MYSQL_PORT env var, default: 3306)',
     ),
     aws_secret_arn: Optional[str] = Field(
         default=None, description='AWS secret ARN (overrides MYSQL_SECRET_ARN env var)'
@@ -150,18 +159,26 @@ async def source_db_analyzer(
     - Use these analysis files with the dynamodb_data_modeling tool to design your DynamoDB schema
 
     Connection Requirements (MySQL/Aurora):
-    - AWS RDS Data API enabled on your Aurora MySQL cluster
+    Two connection methods are supported:
+    1. RDS Data API (Aurora MySQL): Requires aws_cluster_arn
+    2. Direct Connection (Aurora/RDS/self-hosted MySQL): Requires hostname and port
+
+    Note: Do not provide both CLUSTER_ARN and HOSTNAME- the tool will automatically use the first available option
+
+    Both methods require:
     - Database credentials stored in AWS Secrets Manager
-    - Appropriate IAM permissions to access RDS Data API and Secrets Manager
+    - Appropriate IAM permissions to access Secrets Manager
     - For complete analysis: MySQL Performance Schema must be enabled (set performance_schema=ON)
     - Without Performance Schema: Schema-only analysis is performed (no query pattern data)
 
     Environment Variables (Optional):
     You can set these instead of passing parameters:
     - MYSQL_DATABASE: Database name to analyze
-    - MYSQL_CLUSTER_ARN: Aurora cluster ARN
+    - MYSQL_CLUSTER_ARN: Aurora cluster ARN (for RDS Data API)
+    - MYSQL_HOSTNAME: Database hostname (for direct connection)
+    - MYSQL_PORT: Database port (for direct connection, default: 3306)
     - MYSQL_SECRET_ARN: Secrets Manager secret ARN containing DB credentials
-    - AWS_REGION: AWS region where your database is located
+    - AWS_REGION: AWS region where your database instance and Secrets Manager are located
     - MYSQL_MAX_QUERY_RESULTS: Maximum rows per query (default: 500)
 
     Typical Usage:
@@ -185,6 +202,8 @@ async def source_db_analyzer(
         pattern_analysis_days=pattern_analysis_days,
         max_query_results=max_query_results,
         aws_cluster_arn=aws_cluster_arn,
+        hostname=hostname,
+        port=port,
         aws_secret_arn=aws_secret_arn,
         aws_region=aws_region,
         output_dir=output_dir,
@@ -195,10 +214,9 @@ async def source_db_analyzer(
         source_db_type, connection_params
     )
     if missing_params:
+        # Handle missing required parameters
         missing_descriptions = [param_descriptions[param] for param in missing_params]
-        return (
-            f'To analyze your {source_db_type} database, I need: {", ".join(missing_descriptions)}'
-        )
+        return f'Missing required parameters: {", ".join(missing_descriptions)}'
 
     logger.info(
         f'Starting database analysis for {source_db_type} database: {connection_params.get("database")}'

src/dynamodb-mcp-server/pyproject.toml

@@ -13,7 +13,7 @@ dependencies = [
     "typing-extensions==4.14.1",
     "strands-agents>=1.5.0",
     "dspy-ai>=2.6.27",
-    "awslabs.mysql-mcp-server==1.0.5",
+    "awslabs.mysql-mcp-server==1.0.9",
     "awslabs-aws-api-mcp-server==1.0.2",
 ]
 license = {text = "Apache-2.0"}

src/dynamodb-mcp-server/tests/test_dynamodb_server.py

@@ -73,7 +73,7 @@ async def test_source_db_analyzer_missing_parameters(tmp_path):
         output_dir=str(tmp_path),
     )
 
-    assert 'To analyze your mysql database, I need:' in result
+    assert 'Missing required parameters: Database name to analyze' in result
 
 
 @pytest.mark.asyncio
@@ -90,7 +90,10 @@ async def test_source_db_analyzer_empty_parameters(tmp_path):
         output_dir=str(tmp_path),
     )
 
-    assert 'To analyze your mysql database, I need:' in result
+    assert (
+        'Missing required parameters: Required: Either aws_cluster_arn (Aurora MySQL cluster ARN for RDS Data API) OR hostname (MySQL server hostname for direct connection)'
+        in result
+    )
 
 
 @pytest.mark.asyncio
@@ -112,7 +115,29 @@ async def test_source_db_analyzer_env_fallback(monkeypatch, tmp_path):
     )
 
     # Should still fail due to missing cluster_arn, but covers env fallback lines
-    assert 'To analyze your mysql database, I need:' in result
+    assert 'Missing required parameters:' in result
+
+
+@pytest.mark.asyncio
+async def test_source_db_analyzer_connection_method_precedence(mysql_env_setup, tmp_path):
+    """Test that explicit connection parameters take precedence over environment variables."""
+    # mysql_env_setup fixture sets MYSQL_CLUSTER_ARN, MYSQL_SECRET_ARN, AWS_REGION
+    # Pass explicit hostname parameter - this should take precedence over env cluster_arn
+    result = await source_db_analyzer(
+        source_db_type='mysql',
+        database_name='test',
+        pattern_analysis_days=30,
+        max_query_results=None,
+        aws_cluster_arn=None,  # No explicit cluster_arn
+        hostname='explicit-hostname',  # Explicit hostname should pass
+        aws_secret_arn=None,  # Will use env var
+        aws_region=None,  # Will use env var
+        output_dir=str(tmp_path),
+    )
+
+    # The test validates the precedence works: it used Asyncmy Direct connection (hostname)
+    # instead of RDS Data API (cluster_arn), even though env had MYSQL_CLUSTER_ARN
+    assert 'Database Analysis Failed' in result
 
 
 @pytest.mark.asyncio