Redshift MCP Server

Instructions

Get sample rows from a table.

Args:
    table_name: Name of the table
    limit: Number of rows to return (default: 5)
    schema: Schema name (default: "public")

Returns:
    JSON sample data

Implementation Reference

• redshift_mcp_server.py:117-131 (handler)

  The handler function decorated with @mcp.tool(), defining the input schema via type hints and docstring, and implementing the logic to fetch sample data by calling redshift_query with a generated SQL query.

  @mcp.tool()
  def redshift_get_sample_data(table_name: str, limit: int = 5, schema: str = "public") -> str:
      """
      Get sample rows from a table.
      
      Args:
          table_name: Name of the table
          limit: Number of rows to return (default: 5)
          schema: Schema name (default: "public")
      
      Returns:
          JSON sample data
      """
      sql = f"SELECT * FROM {schema}.{table_name} LIMIT {limit}"
      return redshift_query(sql)

• redshift_mcp_server.py:59-76 (helper)

  Supporting helper tool 'redshift_query' that executes arbitrary SQL using pandas and returns JSON, called by redshift_get_sample_data.

  @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)

  Utility function to establish database connection, used indirectly by redshift_query.

  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