handle_get_table_definition
Retrieve the DDL statement for a specific table in Redshift schema via SQL script. Returns CREATE TABLE statement on success or error details if retrieval fails.
Instructions
Retrieves the DDL (Data Definition Language) statement for a specific table.
Executes a SQL script designed to generate or retrieve the CREATE TABLE
statement for the given table.
Args:
ctx: The MCP context object.
schema_name: The schema name of the table.
table_name: The name of the table.
Returns:
A dictionary conforming to GetTableDefinitionResult structure:
- On success: {"status": "success", "ddl": "<CREATE TABLE statement>"}
- On table not found or DDL retrieval error:
{"status": "error", "error_message": "...", "error_type": "..."}
Raises:
TableNotFound: If the specified table is not found.
DataApiError: If a critical, unexpected error occurs during execution.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| schema_name | Yes | ||
| table_name | Yes |
Implementation Reference
- Core handler function decorated with @mcp.tool(). Executes 'SHOW TABLE' via Data API to retrieve table DDL, parses result, handles errors like TableNotFound.@mcp.tool() async def handle_get_table_definition( ctx: Context, schema_name: str, table_name: str ) -> str: """Retrieves the DDL (Data Definition Language) statement for a specific table. Executes a SQL script designed to generate or retrieve the CREATE TABLE statement for the given table. Args: ctx: The MCP context object. schema_name: The schema name of the table. table_name: The name of the table. Returns: A dictionary conforming to GetTableDefinitionResult structure: - On success: {"status": "success", "ddl": "<CREATE TABLE statement>"} - On table not found or DDL retrieval error: {"status": "error", "error_message": "...", "error_type": "..."} Raises: TableNotFound: If the specified table is not found. DataApiError: If a critical, unexpected error occurs during execution. """ ctx.info(f"Starting DDL retrieval for: {schema_name}.{table_name}...") sql: str = f"SHOW TABLE {schema_name}.{table_name}" ctx.debug(f"Executing command: {sql}") try: config: DataApiConfig = get_data_api_config() ddl_rows: List[Dict[str, Any]] = await execute_sql( config=config, sql=sql, params=None ) if not ddl_rows: ctx.warning( f"Table '{schema_name}.{table_name}' not found or DDL could not be retrieved." ) raise TableNotFound( f"Table '{schema_name}.{table_name}' not found or DDL could not be retrieved." ) ctx.debug(f"Parsing DDL result for {schema_name}.{table_name}") first_row = ddl_rows[0] ddl_string: Optional[str] = list(first_row.values())[0] if first_row else None if not isinstance(ddl_string, str) or not ddl_string.strip().upper().startswith( "CREATE TABLE" ): ctx.error( f"DDL column missing or null in result for {schema_name}.{table_name}" ) raise DataApiError( f"Could not extract DDL string for table '{schema_name}.{table_name}'. Query returned no DDL column." ) ctx.info(f"DDL for {schema_name}.{table_name} retrieved successfully.") return ddl_string except TableNotFound as e: raise e except ( SqlScriptNotFoundError, DataApiError, SqlExecutionError, ClientError, ) as e: ctx.error( f"SQL execution failed while retrieving DDL for {schema_name}.{table_name}: {e}", exc_info=True, ) raise except Exception as e: ctx.error(f"Unexpected error getting DDL for {schema_name}.{table_name}: {e}") raise DataApiError( f"An unexpected server error occurred while retrieving DDL: {e}" )
- Output schema definitions using TypedDict for success (status, ddl) and error (status, error_message, error_type) cases, unioned as GetTableDefinitionResult.class GetTableDefinitionSuccessResult(TypedDict): status: str ddl: str class GetTableDefinitionErrorResult(TypedDict): status: str error_message: str error_type: str GetTableDefinitionResult = Union[ GetTableDefinitionSuccessResult, GetTableDefinitionErrorResult ]
- src/redshift_utils_mcp/server.py:64-72 (registration)Imports tool handlers from tools/handlers package into the FastMCP server module, including get_table_definition (alias for handle_get_table_definition), which triggers registration via decorators.from .tools.handlers import ( # noqa: E402 check_cluster_health, diagnose_locks, diagnose_query_performance, execute_ad_hoc_query, get_table_definition, inspect_table, monitor_workload, )
- src/redshift_utils_mcp/server.py:81-91 (registration)References all imported handlers in a discarded tuple to ensure they are loaded and registered by FastMCP decorators._ = ( check_cluster_health, diagnose_locks, diagnose_query_performance, execute_ad_hoc_query, get_table_definition, inspect_table, monitor_workload, resource_handlers, prompt_handlers, )
- src/redshift_utils_mcp/tools/handlers/__init__.py:7-16 (registration)Re-exports handle_get_table_definition from its module and includes in __all__ for convenient import in server.py.from .get_table_definition import handle_get_table_definition from .inspect_table import handle_inspect_table from .monitor_workload import handle_monitor_workload __all__ = [ "handle_check_cluster_health", "handle_diagnose_locks", "handle_diagnose_query_performance", "handle_execute_ad_hoc_query", "handle_get_table_definition",