get_table_schema
Retrieve comprehensive schema details of a specified table, including column names, data types, and comments, using the Doris MCP Server. Specify the table name and optionally the database name for targeted metadata extraction.
Instructions
[Function Description]: Get detailed structure information of the specified table (columns, types, comments, etc.).
[Parameter Content]:
table_name (string) [Required] - Name of the table to query
db_name (string) [Optional] - Target database name, defaults to the current database
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| db_name | No | ||
| table_name | Yes |
Implementation Reference
- doris_mcp_server/tools/tools_manager.py:114-136 (registration)Registers the 'get_table_schema' tool with the MCP server using the @mcp.tool decorator. Defines input parameters: table_name (required), db_name and catalog_name (optional). The function proxies to self.call_tool for execution.
@mcp.tool( "get_table_schema", description="""[Function Description]: Get detailed structure information of the specified table (columns, types, comments, etc.). [Parameter Content]: - table_name (string) [Required] - Name of the table to query - db_name (string) [Optional] - Target database name, defaults to the current database - catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog """, ) async def get_table_schema_tool( table_name: str, db_name: str = None, catalog_name: str = None ) -> str: """Get table schema information""" return await self.call_tool("get_table_schema", { "table_name": table_name, "db_name": db_name, "catalog_name": catalog_name }) - Defines the input schema validation for the 'get_table_schema' tool in the list_tools method, specifying JSON schema with table_name as required string property.
name="get_table_schema", description="""[Function Description]: Get detailed structure information of the specified table (columns, types, comments, etc.). [Parameter Content]: - table_name (string) [Required] - Name of the table to query - db_name (string) [Optional] - Target database name, defaults to the current database - catalog_name (string) [Optional] - Target catalog name for federation queries, defaults to current catalog """, inputSchema={ "type": "object", "properties": { "table_name": {"type": "string", "description": "Table name"}, "db_name": {"type": "string", "description": "Database name"}, "catalog_name": {"type": "string", "description": "Catalog name"}, }, "required": ["table_name"], }, - Dispatcher handler in ToolsManager.call_tool that routes 'get_table_schema' tool invocations by extracting arguments and delegating to MetadataExtractor.get_table_schema_for_mcp.
async def _get_table_schema_tool(self, arguments: Dict[str, Any]) -> Dict[str, Any]: """Get table schema tool routing""" table_name = arguments.get("table_name") db_name = arguments.get("db_name") catalog_name = arguments.get("catalog_name") # Delegate to metadata extractor for processing return await self.metadata_extractor.get_table_schema_for_mcp( table_name, db_name, catalog_name ) - Core handler implementation for 'get_table_schema' tool in MetadataExtractor. Validates input, calls get_table_schema_async to fetch schema via DESCRIBE query, formats MCP response with success/error handling.
async def get_table_schema_for_mcp( self, table_name: str, db_name: str = None, catalog_name: str = None ) -> Dict[str, Any]: """Get detailed schema information for specified table (columns, types, comments, etc.) - MCP interface""" logger.info(f"Getting table schema: Table: {table_name}, DB: {db_name}, Catalog: {catalog_name}") if not table_name: return self._format_response(success=False, error="Missing table_name parameter") try: schema = await self.get_table_schema_async(table_name=table_name, db_name=db_name, catalog_name=catalog_name) if not schema: return self._format_response( success=False, error="Table does not exist or has no columns", message=f"Unable to get schema for table {catalog_name or 'default'}.{db_name or self.db_name}.{table_name}" ) return self._format_response(success=True, result=schema) except Exception as e: logger.error(f"Failed to get table schema: {str(e)}", exc_info=True) return self._format_response(success=False, error=str(e), message="Error occurred while getting table schema") - Helper function that executes DESCRIBE query (with catalog.db.table support) to retrieve raw table schema and formats it into structured column information list.
async def get_table_schema_async(self, table_name: str, db_name: str = None, catalog_name: str = None) -> List[Dict[str, Any]]: """Asynchronously get table schema information""" try: # Use async query method effective_catalog = catalog_name or self.catalog_name # Build query statement if effective_catalog and effective_catalog != "internal": query = f"DESCRIBE `{effective_catalog}`.`{db_name or self.db_name}`.`{table_name}`" else: query = f"DESCRIBE `{db_name or self.db_name}`.`{table_name}`" # Execute async query result = await self._execute_query_async(query, db_name) if not result: return [] # Process results schema = [] for row in result: if isinstance(row, dict): schema.append({ 'column_name': row.get('Field', ''), 'data_type': row.get('Type', ''), 'is_nullable': row.get('Null', 'NO') == 'YES', 'default_value': row.get('Default', None), 'comment': row.get('Comment', ''), 'key': row.get('Key', ''), 'extra': row.get('Extra', '') }) return schema except Exception as e: logger.error(f"Failed to get table schema: {e}") return []