Hologres MCP Server

Instructions

Implementation Reference

• src/hologres_mcp_server/server.py:625-653 (handler)

  Handler logic for the 'list_hg_tables_in_a_schema' tool: extracts schema from arguments, validates it, and constructs a complex SQL query to list tables in the schema, annotating types (view, foreign table, partitioned table) while excluding system schemas and non-top-level partitioned tables.

  elif name == "list_hg_tables_in_a_schema":
      schema = arguments.get("schema")
      if not schema:
          raise ValueError("Schema name is required")
      query = f"""
          SELECT
              tab.table_name,
              CASE WHEN tab.table_type = 'VIEW' THEN ' (view)'
                  WHEN tab.table_type = 'FOREIGN' THEN ' (foreign table)'
                  WHEN p.partrelid IS NOT NULL THEN ' (partitioned table)'
                  ELSE ''
              END AS table_type_info
          FROM
              information_schema.tables AS tab
          LEFT JOIN pg_class AS cls ON tab.table_name = cls.relname
          LEFT JOIN pg_namespace AS ns ON tab.table_schema = ns.nspname
          LEFT JOIN pg_inherits AS inh ON cls.oid = inh.inhrelid
          LEFT JOIN pg_partitioned_table AS p ON cls.oid = p.partrelid
          WHERE
              tab.table_schema NOT IN ('pg_catalog', 'information_schema', 'hologres', 'hologres_statistic', 'hologres_streaming_mv')
              AND tab.table_schema = '{schema}'
              AND (inh.inhrelid IS NULL OR NOT EXISTS (
                  SELECT 1
                  FROM pg_inherits
                  WHERE inh.inhrelid = pg_inherits.inhrelid
              ))
          ORDER BY
              tab.table_name;
      """

• src/hologres_mcp_server/server.py:514-526 (registration)

  Registers the tool in the list_tools() decorator response with name, description, and input schema.

      name="list_hg_tables_in_a_schema",
      description="List all tables in a specific schema in the current Hologres database, including their types (table, view, foreign table, partitioned table).",
      inputSchema={
          "type": "object",
          "properties": {
              "schema": {
                  "type": "string",
                  "description": "Schema name to list tables from in Hologres database"
              }
          },
          "required": ["schema"]
      }
  ),

• src/hologres_mcp_server/server.py:517-525 (schema)

  Pydantic-like input schema definition for the tool, requiring a single 'schema' string parameter.

      "type": "object",
      "properties": {
          "schema": {
              "type": "string",
              "description": "Schema name to list tables from in Hologres database"
          }
      },
      "required": ["schema"]
  }

• src/hologres_mcp_server/utils.py:48-79 (helper)

  Generic helper function that executes the SQL query prepared by tool handlers, handles connection retries, sets serverless mode if needed, formats SELECT results as CSV, and returns success/error messages.

  def handle_call_tool(tool_name, query, serverless = False):
      """Handle callTool method."""
      config = get_db_config()
      try:
          with connect_with_retry() as conn:
              with conn.cursor() as cursor:
  
                  # 特殊处理 serverless computing 查询
                  if serverless:
                      cursor.execute("set hg_computing_resource='serverless'")
                  
                  # Execute the query
                  cursor.execute(query)
                  
                  # 特殊处理 ANALYZE 命令
                  if tool_name == "gather_hg_table_statistics":
                      return f"Successfully {query}"
                  
                  # 处理其他有返回结果的查询
                  if cursor.description:  # SELECT query
                      columns = [desc[0] for desc in cursor.description]
                      rows = cursor.fetchall()
                      result = [",".join(map(str, row)) for row in rows]
                      return "\n".join([",".join(columns)] + result)
                  elif tool_name == "execute_dml_sql":  # Non-SELECT query
                      row_count = cursor.rowcount
                      return f"Query executed successfully. {row_count} rows affected."
                  else:
                      return "Query executed successfully"
      except Exception as e:
          return f"Error executing query: {str(e)}"