MCP Vertica

Instructions

Args:
    ctx: FastMCP context for progress reporting and logging
    table_name: Name of the table to inspect
    schema: Schema name (default: public)

Returns:
    Table structure information as a string

Implementation Reference

• src/mcp_vertica/mcp.py:367-458 (handler)

  The main handler function for the 'get_table_structure' tool. It is decorated with @mcp.tool(), which serves as both the implementation and registration with the FastMCP server. The function connects to Vertica, queries v_catalog.columns for column info and v_catalog.constraint_columns for constraints, then formats and returns the table structure as a string.

  @mcp.tool()
  async def get_table_structure(
      ctx: Context,
      table_name: str,
      schema: str = "public"
  ) -> str:
      """Get the structure of a table including columns, data types, and constraints.
  
      Args:
          ctx: FastMCP context for progress reporting and logging
          table_name: Name of the table to inspect
          schema: Schema name (default: public)
  
      Returns:
          Table structure information as a string
      """
      await ctx.info(f"Getting structure for table: {schema}.{table_name}")
  
      # Get or create connection manager
      manager = await get_or_create_manager(ctx)
      if not manager:
          return "Error: Failed to initialize database connection. Check configuration."
  
      query = """
      SELECT
          column_name,
          data_type,
          character_maximum_length,
          numeric_precision,
          numeric_scale,
          is_nullable,
          column_default
      FROM v_catalog.columns
      WHERE table_schema = %s
      AND table_name = %s
      ORDER BY ordinal_position;
      """
  
      conn = None
      cursor = None
      try:
          conn = manager.get_connection()
          cursor = conn.cursor()
          cursor.execute(query, (schema, table_name))
          columns = cursor.fetchall()
  
          if not columns:
              return f"No table found: {schema}.{table_name}"
  
          # Get constraints
          cursor.execute("""
              SELECT
                  constraint_name,
                  constraint_type,
                  column_name
              FROM v_catalog.constraint_columns
              WHERE table_schema = %s
              AND table_name = %s;
          """, (schema, table_name))
          constraints = cursor.fetchall()
  
          # Format the output
          result = f"Table Structure for {schema}.{table_name}:\n\n"
          result += "Columns:\n"
          for col in columns:
              result += f"- {col[0]}: {col[1]}"
              if col[2]:  # character_maximum_length
                  result += f"({col[2]})"
              elif col[3]:  # numeric_precision
                  result += f"({col[3]},{col[4]})"
              result += f" {'NULL' if col[5] == 'YES' else 'NOT NULL'}"
              if col[6]:  # column_default
                  result += f" DEFAULT {col[6]}"
              result += "\n"
  
          if constraints:
              result += "\nConstraints:\n"
              for const in constraints:
                  result += f"- {const[0]} ({const[1]}): {const[2]}\n"
  
          return result
  
      except Exception as e:
          error_msg = f"Error getting table structure: {str(e)}"
          await ctx.error(error_msg)
          return error_msg
      finally:
          if cursor:
              cursor.close()
          if conn:
              manager.release_connection(conn)