redshift-utils-mcp

Instructions

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.

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