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

src/redshift_utils_mcp/tools/handlers/get_table_definition.py:19-98 (handler)

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

src/redshift_utils_mcp/tools/handlers/common.py:266-279 (schema)

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

redshift-utils-mcp