Build an MCP Server over an Apache Iceberg Catalog

Apache Iceberg gives you open tables and a catalog any engine can read. MCP gives an agent a standard way to call tools. This pattern connects them: a Python MCP server that loads tables through an Iceberg catalog with PyIceberg, queries them with DuckDB, and uses Iceberg's own metadata to make decisions before a query runs. It runs on a laptop with a SQLite catalog, and the catalog is one function to swap for a REST catalog such as Apache Polaris.

The shape of the pattern

Catalog: PyIceberg load_catalog. Locally a SQL catalog in SQLite with a folder warehouse; in production a REST catalog.
Tables: Iceberg tables written with PyIceberg from Arrow data, partitioned so metadata is useful for pruning.
Engine: DuckDB reads Arrow from Iceberg scans and runs the compiled, governed SQL.
Agent interface: an MCP server with metric tools, covered in exposing a semantic layer over MCP.

Step 1: Install and create the catalog

PyIceberg's extras pull in what each piece needs: sql-sqlite for the catalog, pyarrow for reading and writing, and pyiceberg-core for partition transforms on write. Without pyiceberg-core, appending to a month-partitioned table fails with NotInstalledError.

[project]
dependencies = [
    "pyiceberg[sql-sqlite,pyarrow,pyiceberg-core]>=0.12,<0.13",
    "mcp>=2.2,<3",
    "duckdb>=1.5,<2",
    "pyyaml>=6",
]

from pyiceberg.catalog import Catalog, load_catalog

def get_catalog(home=None) -> Catalog:
    """Load (or create) the SQL catalog backed by SQLite and a local warehouse."""
    root = demo_home(home)
    warehouse = root / "warehouse"
    warehouse.mkdir(parents=True, exist_ok=True)
    return load_catalog(
        "demo",
        **{
            "type": "sql",
            "uri": f"sqlite:///{root / 'catalog.db'}",
            "warehouse": warehouse.as_uri(),
        },
    )

The SQL catalog creates its own tables in the SQLite file on first use. Everything the demo writes lives under one folder, so rm -rf .lakehouse resets it.

Step 2: Write tables, partitioned for the questions agents ask

Agents ask time-bounded questions: last quarter, this year, since the launch. Partitioning the fact table by month lets Iceberg skip whole months during planning, which Step 4 turns into a cost estimate. With a PyArrow schema, field IDs are assigned when the table is created, so the demo adds the partition field right after creation and before the first write:

customers_table = catalog.create_table(f"{NAMESPACE}.customers", schema=CUSTOMERS_SCHEMA)
customers_table.append(customers)

orders_table = catalog.create_table(f"{NAMESPACE}.orders", schema=ORDERS_SCHEMA)
# Partition by month of order_date so scan planning can prune by time,
# which the cost guardrail uses to estimate rows scanned.
with orders_table.update_spec() as update:
    update.add_field("order_date", MonthTransform(), "order_month")
orders_table.append(orders)

The seed script generates 500 customers and 36,000 orders from a fixed random seed, so every run and every test sees the same numbers:

$ slh-seed
Seeded Iceberg tables in /home/you/mcp-semantic-lakehouse-demo/.lakehouse
  sales.customers: 500 rows
  sales.orders: 36,000 rows

A test confirms the layout the rest of the server relies on:

def test_iceberg_tables_exist_and_orders_are_partitioned_by_month(home):
    catalog = get_catalog(home)
    orders = catalog.load_table("sales.orders")
    assert [f.name for f in orders.spec().fields] == ["order_month"]
    assert str(orders.spec().fields[0].transform) == "month"
    assert catalog.load_table("sales.customers").scan().to_arrow().num_rows == 500
    # One data file per month: Jan 2025 to Jun 2026.
    assert len(list(orders.scan().plan_files())) == 18

Step 3: Read Iceberg into the engine

An Iceberg scan returns Arrow, and DuckDB reads Arrow without copying through Python objects. PyIceberg also has a to_duckdb() shorthand that registers the Arrow table on a connection; the demo does the same steps by hand so the tables land in a named raw schema, away from the governed views agents query:

for key, identifier in model.sources.items():
    arrow = catalog.load_table(identifier).scan().to_arrow()
    con.register("_incoming", arrow)
    con.execute(f"CREATE TABLE {RAW_SCHEMA}.{key} AS SELECT * FROM _incoming")
    con.unregister("_incoming")

Loading into memory at startup keeps a demo fast and simple. At lakehouse scale, query the tables in place: point an engine at the same catalog, or pass a row_filter and selected_fields to scan() so PyIceberg only reads what a request needs.

Step 4: Use Iceberg metadata as a tool, not only as storage

Iceberg metadata is cheap to read and says a lot: how many rows a table holds, which files match a filter, when the current snapshot was committed. The server uses it to estimate how many rows a request would scan before running it:

row_filter = And(
    GreaterThanOrEqual(time_column, start.isoformat()),
    LessThan(time_column, end.isoformat()),
)
return sum(task.file.record_count for task in table.scan(row_filter=row_filter).plan_files())

plan_files() reads manifests only. For Q1 2026 it keeps three month files and returns 5,930 rows; with no time range the server reads total-records from the snapshot summary instead (36,000) and the cost guardrail refuses the query. The same metadata could back a freshness tool that tells the agent when the data was last committed, so it can say "as of" in its answer.

Step 5: Serve it over MCP

The server object is built once, and the service behind it opens the catalog and builds the governed engine on first use:

def build_server(service: SemanticLakehouse | None = None) -> MCPServer:
    """Create the MCP server. Pass a service in tests; otherwise one is built lazily."""
    state: dict[str, SemanticLakehouse] = {}

    def svc() -> SemanticLakehouse:
        if "s" not in state:
            try:
                state["s"] = service or SemanticLakehouse(role=os.environ.get("AGENT_ROLE"))
            except RuntimeError as exc:  # e.g. tables not seeded yet: tell the agent why
                raise ToolError(str(exc)) from exc
        return state["s"]
    ...


def main() -> None:
    build_server().run()  # stdio

Building lazily means an MCP client can start the server and list its tools even before anyone has seeded the tables; the first real call then returns a tool error the agent can read, "Iceberg table sales.orders not found. Run `slh-seed` first.", instead of a crash at startup. Raising ToolError matters here: the SDK hides the text of any other exception from the model.

Run the full walkthrough, a real MCP client talking to the server in process:

slh-seed
slh-demo
AGENT_ROLE=emea_sales_agent slh-demo

Step 6: Swap the local catalog for a REST catalog

Nothing above depends on SQLite except get_catalog. PyIceberg's REST catalog takes a URI, a client credential, and a warehouse name, which is what an Apache Polaris deployment or another Iceberg REST catalog gives you:

catalog = load_catalog(
    "prod",
    **{
        "type": "rest",
        "uri": "https://your-catalog.example.com/api/catalog",
        "credential": f"{client_id}:{client_secret}",
        "warehouse": "analytics",
    },
)

This snippet is configuration, not part of the tested demo. When you make the swap, give the MCP server its own catalog principal with read access to the source namespaces only, so the catalog enforces the same boundary the governed views do.

Done when

The MCP server reaches data only through an Iceberg catalog, with its own credentials.
Fact tables are partitioned on the column agents filter by most, usually time.
Cost or scope decisions read Iceberg metadata before any data file is opened.
The catalog is created in one function, so moving from local to REST changes configuration, not code.
A fresh clone goes from install to a working tool call in a few minutes.

Taking it to production

Two paths work. Keep this server and replace DuckDB-in-memory with an engine that reads Iceberg in place and enforces grants, or keep the MCP layer thin and delegate to a lakehouse platform that already exposes governed views and its own MCP server. Either way the parts that matter carry over: the catalog is the source of truth, agents only see governed views, and metadata is used to say no before a query costs money.

An MCP server over an Iceberg catalog