list_hg_tables_in_a_schema
Retrieve all tables, views, and foreign tables from a specified Hologres database schema to analyze database structure and metadata.
Instructions
List all tables in a specific schema in the current Hologres database, including their types (table, view, foreign table, partitioned table).
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| schema | Yes | Schema name to list tables from in Hologres database |
Implementation Reference
- 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"] } ),
- 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"] }
- 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)}"