get_table_schema
Retrieve schema details for a specific Iceberg table by specifying its namespace and table name, enabling efficient metadata queries within the IcebergMCP server.
Instructions
Provides the schema for a given Iceberg table
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| namespace | Yes | ||
| table_name | Yes |
Implementation Reference
- iceberg_mcp/iceberg_server.py:41-63 (handler)The handler function for the 'get_table_schema' tool, decorated with @mcp.tool(). It loads the Iceberg table from the Glue catalog and extracts the schema fields into SchemaField dicts.@mcp.tool() 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:17-22 (schema)TypedDict used as the return type for schema fields in get_table_schema tool.class SchemaField(TypedDict): id: int name: str type: str required: bool doc: str | None
- Helper function called by get_table_schema to create and return the GlueCatalog instance.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