Redshift MCP Server

Instructions

Implementation Reference

• src/redshift_mcp_server/server.py:173-221 (handler)

  The @server.call_tool() decorator registers this as the tool handler. It dispatches based on the tool name. For 'execute_sql', it retrieves the 'sql' parameter from args, validates it, connects to Redshift using config from env vars, executes the SQL, fetches results if query returns columns, formats as CSV-like text, and returns as TextContent. Handles errors and closes connection.

  @server.call_tool()
  async def call_tool(name: str, args: dict) -> list[TextContent]:
      """Execute SQL"""
      config=get_redshift_config()
      sql = ''
  
      if name == "execute_sql":
          sql = args.get("sql")
          if not sql:
              raise ValueError("sql parameter is required when calling execute_sql tool")
      elif name == "analyze_table":
          schema = args.get("schema")
          table = args.get("table")
          if not all([schema, table]):
              raise ValueError("'schema' and 'table' parameters are required when calling analyze_table tool")
          sql = f"ANALYZE {schema}.{table}"
      elif name == "get_execution_plan":
          sql = args.get("sql")
          if not sql:
              raise ValueError("sql parameter is required when calling get_query_plan tool")
          sql = f"EXPLAIN {sql}"
  
      try:
          conn = redshift_connector.connect(
              host=config['host'],
              port=int(config['port']),
              user=config['user'],
              password=config['password'],
              database=config['database'],
          )
          conn.autocommit = True
  
          with conn.cursor() as cursor:
              cursor.execute(sql)
              if name == "analyze_table":
                  return [TextContent(type="text", text=f"Successfully analyzed table {schema}.{table}")]
  
              if cursor.description is None:
                  return [TextContent(type="text", text=f"Successfully execute sql {sql}")]
  
              columns = [desc[0] for desc in cursor.description]
              rows = cursor.fetchall()
              result = [",".join(map(str, row)) for row in rows]
              return [TextContent(type="text", text="\n".join([",".join(columns)] +  result ))]
      except Exception as e:
          return [TextContent(type="text", text=f"Error executing query: {str(e)}")]
      finally:
          conn.close()

• src/redshift_mcp_server/server.py:128-137 (schema)

  Pydantic-like JSON schema for 'execute_sql' tool input, defining an object with a required 'sql' string property.

  inputSchema={
      "type": "object",
      "properties": {
          "sql": {
              "type": "string",
              "description": "The SQL to Execute"
          }
      },
      "required": ["sql"]
  }

• src/redshift_mcp_server/server.py:125-138 (registration)

  The Tool object registration for 'execute_sql' returned by @server.list_tools(), including name, description, and input schema.

  Tool(
      name="execute_sql",
      description="Execute a SQL Query on the Redshift cluster",
      inputSchema={
          "type": "object",
          "properties": {
              "sql": {
                  "type": "string",
                  "description": "The SQL to Execute"
              }
          },
          "required": ["sql"]
      }
  ),