IcebergMCP

Instructions

Implementation Reference

• iceberg_mcp/iceberg_server.py:87-106 (handler)

  Handler function for get_table_partitions tool. Loads the Iceberg table using the catalog and processes its partitions into a list of dictionaries containing partition details, record count, and size.

  @mcp.tool()
  def get_table_partitions(
          namespace: str,
          table_name: str
  ) -> list[dict[str, int]]:
      """Provides the partitions for a given Iceberg table""" 
      catalog: Catalog = get_catalog()
      table_obj = catalog.load_table((namespace, table_name))
      partitions = table_obj.inspect.partitions().to_pylist()
  
      result = []
      for p in partitions:
          result.append(
              {
                  "partition": p['partition'],
                  "record_count": p['record_count'],
                  "size_in_bytes": p['total_data_file_size_in_bytes'],
              }
          )
      return result

• iceberg_mcp/iceberg_server.py:108-125 (helper)

  Helper function to create and return the GlueCatalog instance used by the get_table_partitions handler and other tools.

  def get_catalog() -> GlueCatalog:
      try:
          session = boto3.Session(profile_name=iceberg_config.profile_name)
          credentials = session.get_credentials().get_frozen_credentials()
  
          catalog = GlueCatalog(
              "glue",
              **{
                  "client.access-key-id": credentials.access_key,
                  "client.secret-access-key": credentials.secret_key,
                  "client.session-token": credentials.token,
                  "client.region": iceberg_config.region,
              },
          )
      except Exception as e:
          logger.error(f"Error creating AWS connection: {str(e)}")
          raise
      return catalog