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:
    Index information as a string

Implementation Reference

• src/mcp_vertica/mcp.py:460-520 (handler)

  The handler function for the 'list_indexes' tool. It is decorated with @mcp.tool() for registration and implements the logic to query Vertica's v_catalog.projections table to retrieve and format index (projection) information for a given table and schema.

  @mcp.tool()
  async def list_indexes(
      ctx: Context,
      table_name: str,
      schema: str = "public"
  ) -> str:
      """List all indexes for a specific table.
  
      Args:
          ctx: FastMCP context for progress reporting and logging
          table_name: Name of the table to inspect
          schema: Schema name (default: public)
  
      Returns:
          Index information as a string
      """
      await ctx.info(f"Listing indexes 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
          projection_name,
          is_super_projection,
          anchor_table_name
      FROM v_catalog.projections
      WHERE projection_schema = %s
      AND anchor_table_name = %s
      ORDER BY projection_name;
      """
  
      conn = None
      cursor = None
      try:
          conn = manager.get_connection()
          cursor = conn.cursor()
          cursor.execute(query, (schema, table_name))
          indexes = cursor.fetchall()
  
          if not indexes:
              return f"No projections found for table: {schema}.{table_name}"
  
          # Format the output for projections
          result = f"Projections for {schema}.{table_name}:\n\n"
          for proj in indexes:
              # proj[0]: projection_name, proj[1]: is_super_projection, proj[2]: anchor_table_name
              result += f"- {proj[0]} (Super Projection: {proj[1]}) [Table: {proj[2]}]\n"
          return result
  
      except Exception as e:
          error_msg = f"Error listing indexes: {str(e)}"
          await ctx.error(error_msg)
          return error_msg
      finally:
          if cursor:
              cursor.close()
          if conn:
              manager.release_connection(conn)