AIP-308: Add user/password authentication support for Redshift MCP

- Added RedshiftAuthConfig model for authentication configuration

- Added support for user/password authentication in execute_statement function

- Updated tool definitions to include authentication parameters

- Updated documentation to reflect new authentication options

Co-authored-by: Daniel Foguelman <>

Author: openhands-agent

src/redshift-mcp-server/README.md

@@ -9,6 +9,7 @@ This MCP server provides tools to discover, explore, and query Amazon Redshift c
 - **Cluster Discovery**: Automatically discover both provisioned Redshift clusters and serverless workgroups
 - **Metadata Exploration**: Browse databases, schemas, tables, and columns
 - **Safe Query Execution**: Execute SQL queries in a READ ONLY mode (a safe READ WRITE support is planned to be implemnted in the future versions)
+- **Multiple Authentication Methods**: Support for both IAM authentication and username/password authentication
 - **Multi-Cluster Support**: Work with multiple clusters and workgroups simultaneously
 
 ## Prerequisites
@@ -27,6 +28,19 @@ This MCP server provides tools to discover, explore, and query Amazon Redshift c
    - Region specified in your AWS profile configuration
 3. **Permissions**: Ensure your AWS credentials have the required permissions (see [Permissions](#permissions) section)
 
+### Authentication Methods
+
+1. **IAM Authentication** (Default): Uses AWS IAM credentials to authenticate with Redshift
+   - Requires appropriate IAM permissions for Redshift Data API
+   - No additional configuration needed beyond AWS credentials
+
+2. **Username/Password Authentication**: Uses database credentials to authenticate
+   - Set the following environment variables:
+     - `REDSHIFT_AUTH_TYPE=user_password`
+     - `REDSHIFT_USERNAME=<your-username>`
+     - `REDSHIFT_PASSWORD=<your-password>`
+   - Or provide these parameters directly to the MCP tools
+
 ## Installation
 
 | Cursor | VS Code |

src/redshift-mcp-server/awslabs/redshift_mcp_server/consts.py

@@ -14,6 +14,11 @@
 
 """Redshift MCP Server constants."""
 
+# Authentication
+AUTH_TYPE_IAM = "iam"
+AUTH_TYPE_USER_PASSWORD = "user_password"
+DEFAULT_AUTH_TYPE = AUTH_TYPE_IAM
+
 # System
 CLIENT_CONNECT_TIMEOUT = 60
 CLIENT_READ_TIMEOUT = 600
@@ -59,6 +64,7 @@
 
 - We are use the Redshift API and Redshift Data API.
 - Leverage IAM authentication when possible instead of secrets (database passwords).
+- When IAM authentication is not available, user/password authentication can be used as an alternative.
 """
 
 # SQL queries

src/redshift-mcp-server/awslabs/redshift_mcp_server/models.py

@@ -38,6 +38,23 @@ class RedshiftCluster(BaseModel):
     tags: Optional[Dict[str, str]] = Field(
         default_factory=dict, description='Tags associated with the cluster'
     )
+    auth_type: Optional[str] = Field(
+        None, description='Authentication type (iam or user_password)'
+    )
+
+
+class RedshiftAuthConfig(BaseModel):
+    """Authentication configuration for Redshift connections."""
+
+    auth_type: str = Field(
+        'iam', description='Authentication type (iam or user_password)'
+    )
+    username: Optional[str] = Field(
+        None, description='Username for user/password authentication'
+    )
+    password: Optional[str] = Field(
+        None, description='Password for user/password authentication'
+    )
 
 
 class RedshiftDatabase(BaseModel):

src/redshift-mcp-server/awslabs/redshift_mcp_server/redshift.py

@@ -20,6 +20,9 @@
 import regex
 from awslabs.redshift_mcp_server import __version__
 from awslabs.redshift_mcp_server.consts import (
+    AUTH_TYPE_IAM,
+    AUTH_TYPE_USER_PASSWORD,
+    DEFAULT_AUTH_TYPE,
     CLIENT_CONNECT_TIMEOUT,
     CLIENT_READ_TIMEOUT,
     CLIENT_RETRIES,
@@ -40,11 +43,15 @@ class RedshiftClientManager:
     """Manages AWS clients for Redshift operations."""
 
     def __init__(
-        self, config: Config, aws_region: str | None = None, aws_profile: str | None = None
+        self, config: Config, aws_region: str | None = None, aws_profile: str | None = None,
+        auth_type: str | None = None, username: str | None = None, password: str | None = None
     ):
         """Initialize the client manager."""
         self.aws_region = aws_region
         self.aws_profile = aws_profile
+        self.auth_type = auth_type
+        self.username = username
+        self.password = password
         self._redshift_client = None
         self._redshift_serverless_client = None
         self._redshift_data_client = None
@@ -189,18 +196,34 @@ async def execute_statement(
     logger.debug(f'Protected and versioned SQL: {" ".join(sqls)}')
 
     # Execute the query using Data API
+    auth_type = os.environ.get("REDSHIFT_AUTH_TYPE", "iam")
+    username = os.environ.get("REDSHIFT_USERNAME")
+    password = os.environ.get("REDSHIFT_PASSWORD")
+    
+    # Prepare common parameters
+    params = {
+        "Database": database_name,
+        "Sqls": sqls
+    }
+    
+    # Add authentication parameters based on auth type
+    if auth_type == "user_password" and username and password:
+        logger.debug(f"Using user/password authentication with username: {username}")
+        params["DbUser"] = username
+        params["SecretArn"] = None  # Not using a secret
+    
+    # Add cluster identifier based on cluster type
     if cluster_info['type'] == 'provisioned':
         logger.debug(f'Using ClusterIdentifier for provisioned cluster: {cluster_identifier}')
-        response = data_client.batch_execute_statement(
-            ClusterIdentifier=cluster_identifier, Database=database_name, Sqls=sqls
-        )
+        params["ClusterIdentifier"] = cluster_identifier
     elif cluster_info['type'] == 'serverless':
         logger.debug(f'Using WorkgroupName for serverless workgroup: {cluster_identifier}')
-        response = data_client.batch_execute_statement(
-            WorkgroupName=cluster_identifier, Database=database_name, Sqls=sqls
-        )
+        params["WorkgroupName"] = cluster_identifier
     else:
         raise Exception(f'Unknown cluster type: {cluster_info["type"]}')
+    
+    # Execute the statement
+    response = data_client.batch_execute_statement(**params)
 
     query_id = response['Id']
     logger.debug(f'Started query execution: {query_id}')
@@ -619,4 +642,7 @@ async def execute_query(cluster_identifier: str, database_name: str, sql: str) -
     ),
     aws_region=os.environ.get('AWS_REGION'),
     aws_profile=os.environ.get('AWS_PROFILE'),
+    auth_type=os.environ.get('REDSHIFT_AUTH_TYPE'),
+    username=os.environ.get('REDSHIFT_USERNAME'),
+    password=os.environ.get('REDSHIFT_PASSWORD'),
 )

src/redshift-mcp-server/awslabs/redshift_mcp_server/server.py

@@ -17,12 +17,16 @@
 import os
 import sys
 from awslabs.redshift_mcp_server.consts import (
+    AUTH_TYPE_IAM,
+    AUTH_TYPE_USER_PASSWORD,
     CLIENT_BEST_PRACTICES,
+    DEFAULT_AUTH_TYPE,
     DEFAULT_LOG_LEVEL,
     REDSHIFT_BEST_PRACTICES,
 )
 from awslabs.redshift_mcp_server.models import (
     QueryResult,
+    RedshiftAuthConfig,
     RedshiftCluster,
     RedshiftColumn,
     RedshiftDatabase,
@@ -57,6 +61,8 @@
 
 This MCP server provides comprehensive access to Amazon Redshift clusters and serverless workgroups.
 
+It supports both IAM authentication and username/password authentication for connecting to Redshift clusters.
+
 ## Available Tools
 
 ### list_clusters
@@ -536,9 +542,23 @@ async def execute_query_tool(
     sql: str = Field(
         ..., description='The SQL statement to execute. Should be a single SQL statement.'
     ),
+    auth_type: str = Field(
+        DEFAULT_AUTH_TYPE,
+        description='Authentication type to use (iam or user_password). Default is iam.',
+    ),
+    username: str = Field(
+        None,
+        description='Username for user/password authentication. Required if auth_type is user_password.',
+    ),
+    password: str = Field(
+        None,
+        description='Password for user/password authentication. Required if auth_type is user_password.',
+    ),
 ) -> QueryResult:
     """Execute a SQL query against a Redshift cluster or serverless workgroup.
 
+    This tool supports both IAM authentication and username/password authentication.
+
     This tool uses the Redshift Data API to execute SQL queries and return results.
     It supports both provisioned clusters and serverless workgroups, and handles
     various data types in the result set.
@@ -549,6 +569,7 @@ async def execute_query_tool(
     - The cluster must be available and accessible.
     - Required IAM permissions: redshift-data:ExecuteStatement, redshift-data:DescribeStatement, redshift-data:GetStatementResult.
     - The user must have appropriate permissions to execute queries in the specified database.
+    - For user/password authentication, provide the username and password parameters and set auth_type to "user_password".
 
     ## Parameters
 
@@ -557,6 +578,9 @@ async def execute_query_tool(
     - database_name: The database name to execute the query against.
                     IMPORTANT: Use a valid database name from the list_databases tool.
     - sql: The SQL statement to execute. Should be a single SQL statement.
+    - auth_type: Authentication type to use (iam or user_password). Default is iam.
+    - username: Username for user/password authentication. Required if auth_type is user_password.
+    - password: Password for user/password authentication. Required if auth_type is user_password.
 
     ## Response Structure