IcebergMCP

Implementation Reference

• iceberg_mcp/iceberg_server.py:66-84 (handler)

  The handler function decorated with @mcp.tool() that implements the get_table_properties tool. Loads the table from the catalog, extracts partition specs, sort orders, snapshot summary (size and records), and table properties.

  @mcp.tool()
  def get_table_properties(
          namespace: str,
          table_name: str
  ) -> dict:
      catalog: Catalog = get_catalog()
      table_obj = catalog.load_table((namespace, table_name))
      partition_specs = [p.dict() for p in table_obj.metadata.partition_specs]
      sort_orders = [s.dict() for s in table_obj.metadata.sort_orders]
      current_snapshot = table_obj.current_snapshot()
      if not current_snapshot or not current_snapshot.summary:
          return {}
      return {
          "total_size_in_bytes": current_snapshot.summary["total-files-size"],
          "total_records": current_snapshot.summary["total-records"],
          "partition_specs": partition_specs,
          "sort_orders": sort_orders,
          **table_obj.properties
      }

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

  Helper function to initialize and return the GlueCatalog using AWS credentials from the configured profile and region.

  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