Redshift MCP Server

Instructions

Implementation Reference

• src/redshift_mcp_server/server.py:189-193 (handler)

  Specific dispatch handler for the 'get_execution_plan' tool within the call_tool function. It extracts the 'sql' argument, validates its presence (note: minor typo in error message), and modifies the SQL by prefixing 'EXPLAIN ' to obtain the execution plan.

  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}"

• src/redshift_mcp_server/server.py:160-169 (schema)

  Input schema definition for the tool, specifying an object with a required 'sql' property of type string.

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

• src/redshift_mcp_server/server.py:157-170 (registration)

  Registration of the 'get_execution_plan' tool in the @server.list_tools() callback, defining its metadata and input schema.

  Tool(
      name="get_execution_plan",
      description="Get actual execution plan with runtime statistics for a SQL query",
      inputSchema={
          "type": "object",
          "properties": {
              "sql": {
                  "type": "string",
                  "description": "The SQL query to analyze"
              }
          },
          "required": ["sql"]
      }
  )

• src/redshift_mcp_server/server.py:195-220 (helper)

  Shared execution helper used by all tools (including get_execution_plan): establishes Redshift connection using config, executes the prepared SQL, handles special cases, fetches and formats query results (columns + rows as CSV) into TextContent for MCP response, with error handling and connection cleanup.

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