Redshift MCP Server

Instructions

List all tables in a specific schema.

Args:
    schema: The schema name (default: "public")

Returns:
    JSON list of table names

Implementation Reference

• redshift_mcp_server.py:77-95 (handler)

  The core handler implementation for the 'redshift_list_tables' MCP tool. Decorated with @mcp.tool() for registration, defines input schema via type hints and docstring, constructs SQL to list tables in the given schema, and executes via 'redshift_query' helper.

  @mcp.tool()
  def redshift_list_tables(schema: str = "public") -> str:
      """
      List all tables in a specific schema.
      
      Args:
          schema: The schema name (default: "public")
      
      Returns:
          JSON list of table names
      """
      sql = f"""
      SELECT table_name 
      FROM information_schema.tables 
      WHERE table_schema = '{schema}'
      AND table_type = 'BASE TABLE'
      """
      return redshift_query(sql)

• redshift_mcp_server.py:78-87 (schema)

  Input/output schema definition for the tool via function signature (schema: str = 'public') and docstring.

  def redshift_list_tables(schema: str = "public") -> str:
      """
      List all tables in a specific schema.
      
      Args:
          schema: The schema name (default: "public")
      
      Returns:
          JSON list of table names
      """

• redshift_mcp_server.py:59-76 (helper)

  Helper tool 'redshift_query' used by 'redshift_list_tables' to execute the generated SQL and return JSON results.

  @mcp.tool()
  def redshift_query(sql: str) -> str:
      """
      Execute a SQL query on Redshift and return results as JSON.
      
      Args:
          sql: The SQL query to execute
      
      Returns:
          JSON string of the query results or error message
      """
      try:
          with get_connection() as conn:
              df = pd.read_sql(sql, conn)
              return df.to_json(orient="records", indent=2)
      except Exception as e:
          return f"Error executing query: {str(e)}"

• redshift_mcp_server.py:32-56 (helper)

  Supporting get_connection function used indirectly via redshift_query for database connectivity.

  def get_connection():
      """Create a connection to Redshift or local Postgres."""
      try:
          # If host is localhost and port is 5432, assume local Postgres for testing
          if REDSHIFT_HOST == "localhost" and REDSHIFT_PORT == 5432:
              import psycopg2
              return psycopg2.connect(
                  host=REDSHIFT_HOST,
                  port=REDSHIFT_PORT,
                  database=REDSHIFT_DATABASE,
                  user=REDSHIFT_USER,
                  password=REDSHIFT_PASSWORD
              )
          else:
              return redshift_connector.connect(
                  host=REDSHIFT_HOST,
                  port=REDSHIFT_PORT,
                  database=REDSHIFT_DATABASE,
                  user=REDSHIFT_USER,
                  password=REDSHIFT_PASSWORD
              )
      except Exception as e:
          logger.error(f"Connection error: {e}")
          raise