IcebergMCP

Instructions

Implementation Reference

• iceberg_mcp/iceberg_server.py:42-63 (handler)

  The handler function that loads the Iceberg table and extracts its schema fields into the defined SchemaField format.

  def get_table_schema(
      namespace: str,
      table_name: str
  ) -> list[SchemaField]:
      """Provides the schema for a given Iceberg table""" 
      catalog: Catalog = get_catalog()
      table_obj = catalog.load_table((namespace, table_name))
      schema = table_obj.schema()
  
      fields = []
      for field in schema.fields:
          fields.append(
              {
                  "id": field.field_id,
                  "name": field.name,
                  "type": str(field.field_type),
                  "required": field.required,
                  "doc": field.doc if field.doc else None,
              }
          )
  
      return fields

• iceberg_mcp/iceberg_server.py:41-41 (registration)

  Registers the get_table_schema function as an MCP tool using the FastMCP decorator.

  @mcp.tool()

• iceberg_mcp/iceberg_server.py:17-22 (schema)

  Defines the output schema structure for table schema fields.

  class SchemaField(TypedDict):
      id: int
      name: str
      type: str
      required: bool
      doc: str | None

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

  Helper function to initialize the GlueCatalog using AWS credentials from config.

  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