From e2dcf4acfe039ac36f77ab6e0e447ca38993d891 Mon Sep 17 00:00:00 2001 From: Mark Hoerth Date: Mon, 20 Jul 2026 20:12:41 -0700 Subject: [PATCH] mcp-server: route agent discovery through Gravitino metadata Agents given both the Gravitino MCP server and a SQL query engine were routing metadata questions to SQL, enumerating catalogs and schemas one level at a time, and reading data through whichever catalog answered first. Two changes keep discovery in the metadata layer. Add a server-level FastMCP instructions string, defined once and applied to both server construction paths, stating that these tools are the first stop for what exists, where it lives, and how it is governed, that they return metadata and never rows, and that tag and policy tools should be consulted before reading potentially sensitive data. Add a find_metadata tool that resolves a table name to its fully-qualified locations across every relational catalog and schema in a single call, so an agent no longer crawls to locate a table. The crawl is composed from the existing catalog, schema, and table listing operations in a testable helper. A catalog or schema that cannot be listed is recorded under skipped rather than aborting the search. Signed-off-by: Mark Hoerth --- mcp-server/mcp_server/server.py | 24 +++ mcp-server/mcp_server/tools/metadata.py | 154 ++++++++++++++++ mcp-server/tests/unit/tools/test_metadata.py | 177 +++++++++++++++++++ 3 files changed, 355 insertions(+) mode change 100644 => 100755 mcp-server/mcp_server/server.py mode change 100644 => 100755 mcp-server/mcp_server/tools/metadata.py create mode 100755 mcp-server/tests/unit/tools/test_metadata.py diff --git a/mcp-server/mcp_server/server.py b/mcp-server/mcp_server/server.py old mode 100644 new mode 100755 index b8909120da3..898cdfbdaa3 --- a/mcp-server/mcp_server/server.py +++ b/mcp-server/mcp_server/server.py @@ -44,6 +44,28 @@ from mcp_server.core.setting import Setting from mcp_server.tools import load_tools +GRAVITINO_MCP_INSTRUCTIONS = """\ +The Gravitino MCP server is the authoritative source for metadata and \ +governance across the metalake. It answers what data exists, where it lives, \ +how it is structured, and how it is governed, spanning catalogs, schemas, \ +tables, columns, models, filesets, topics, statistics, jobs, tags, and \ +policies. + +Use these tools first for any question about discovery or governance. That \ +includes locating a catalog, schema, or table; inspecting a table's columns \ +and properties; and finding which tags or policies apply to an object. Do not \ +route these questions to a SQL query engine. A query engine discovers \ +structure only by enumerating it one level at a time and executes against the \ +live source, while the metadata tools answer the same questions directly from \ +Gravitino in a single call. + +The Gravitino tools return metadata only and never return table rows. A SQL \ +query engine is the right tool for reading or computing over data, and only \ +after the target object and its governing catalog have been identified here. \ +Before reading data that may hold personal, financial, or otherwise sensitive \ +information, consult the tag and policy tools to determine the object's \ +classification.""" + def _get_principal_from_request(fallback_authorization: str = "") -> str: """Derive a display principal for audit attribution. @@ -100,12 +122,14 @@ def _create_gravitino_mcp(setting: Setting) -> FastMCP: if setting.tags is not None and len(setting.tags) > 0: mcp = FastMCP( "Gravitino MCP Server", + instructions=GRAVITINO_MCP_INSTRUCTIONS, lifespan=_create_lifespan_manager(GravitinoContext(setting)), include_tags=setting.tags, ) else: mcp = FastMCP( "Gravitino MCP Server", + instructions=GRAVITINO_MCP_INSTRUCTIONS, lifespan=_create_lifespan_manager(GravitinoContext(setting)), ) diff --git a/mcp-server/mcp_server/tools/metadata.py b/mcp-server/mcp_server/tools/metadata.py old mode 100644 new mode 100755 index 0782fd54bae..9b1c0955ba6 --- a/mcp-server/mcp_server/tools/metadata.py +++ b/mcp-server/mcp_server/tools/metadata.py @@ -15,9 +15,101 @@ # specific language governing permissions and limitations # under the License. +import json + from fastmcp import Context, FastMCP +async def _find_table_metadata( + client, name: str, case_sensitive: bool +) -> dict: + """Locate tables by name across relational catalogs and schemas. + + Composes the existing catalog, schema, and table listing operations into + a single server-side crawl. A catalog or schema that cannot be listed is + recorded under "skipped" so one bad node does not abort the search. + """ + if not name or not name.strip(): + raise ValueError("name must be a non-empty string") + needle = name if case_sensitive else name.casefold() + + def _matches(candidate: str) -> bool: + hay = candidate if case_sensitive else candidate.casefold() + return needle in hay + + matches = [] + skipped = [] + searched_catalogs = 0 + searched_schemas = 0 + + catalogs = json.loads( + await client.as_catalog_operation().get_list_of_catalogs() + ) + for catalog in catalogs: + if not isinstance(catalog, dict): + continue + if str(catalog.get("type", "")).casefold() != "relational": + continue + catalog_name = catalog.get("name", "") + searched_catalogs += 1 + try: + schemas = json.loads( + await client.as_schema_operation().get_list_of_schemas( + catalog_name + ) + ) + except Exception as exc: # pylint: disable=broad-except + skipped.append({"location": catalog_name, "reason": str(exc)}) + continue + for schema in schemas: + schema_name = ( + schema.get("name", "") + if isinstance(schema, dict) + else str(schema) + ) + searched_schemas += 1 + try: + tables = json.loads( + await client.as_table_operation().get_list_of_tables( + catalog_name, schema_name + ) + ) + except Exception as exc: # pylint: disable=broad-except + skipped.append( + { + "location": f"{catalog_name}.{schema_name}", + "reason": str(exc), + } + ) + continue + for table in tables: + table_name = ( + table.get("name", "") + if isinstance(table, dict) + else str(table) + ) + if _matches(table_name): + matches.append( + { + "catalog": catalog_name, + "schema": schema_name, + "table": table_name, + "fullName": ( + f"{catalog_name}.{schema_name}.{table_name}" + ), + "catalogType": catalog.get("type", ""), + } + ) + return { + "matches": matches, + "searched": { + "catalogs": searched_catalogs, + "schemas": searched_schemas, + }, + "skipped": skipped, + } + + def load_metadata_tool(mcp: FastMCP): @mcp.tool(tags={"tag", "policy"}) async def metadata_type_to_fullname_formats(ctx: Context) -> dict: @@ -42,3 +134,65 @@ async def metadata_type_to_fullname_formats(ctx: Context) -> dict: "fileset": "{catalog}.{schema}.{fileset}", "column": "{catalog}.{schema}.{table}.{column}", } + + @mcp.tool(tags={"table"}) + async def find_metadata( + ctx: Context, name: str, case_sensitive: bool = False + ) -> str: + """ + Locate table metadata by name across every relational catalog and + schema in the metalake, in a single call. + + Use this to answer "where does table live" or "which catalogs + contain a table called " without knowing the catalog or schema + in advance. Prefer it over listing catalogs and schemas by hand, and + prefer it over discovering table locations through a SQL query engine. + The tool returns metadata only and never returns table rows. + + The lookup crawls the metalake server-side: it lists relational + catalogs, then schemas within each, then tables within each schema, + and returns every table whose name matches. Matching is a + case-insensitive substring match by default. A catalog or schema that + cannot be listed, for example a schema reported as managed by multiple + catalogs, is skipped and recorded under "skipped" rather than aborting + the search. + + Args: + ctx (Context): The request context object. + name (str): The table name or name fragment to search for. + case_sensitive (bool): Match the name by case when True. Defaults + to False (case-insensitive). + + Returns: + str: A JSON string with the following structure: + { + "matches": [ + { + "catalog": "catalog_postgres", + "schema": "hr", + "table": "employees", + "fullName": "catalog_postgres.hr.employees", + "catalogType": "relational" + } + ], + "searched": {"catalogs": 3, "schemas": 7}, + "skipped": [ + { + "location": "catalog_hive.sales", + "reason": "Schema managed by multiple catalogs" + } + ] + } + + Special Considerations: + - Only relational catalogs are searched. Filesets, topics, and + models are out of scope for this tool. + - On a large metalake the crawl issues one listing call per catalog + and per schema, so it is heavier than a direct load when the + catalog and schema are already known. In that case use + get_list_of_tables or get_table_metadata_details instead. + """ + client = ctx.request_context.lifespan_context.rest_client() + return json.dumps( + await _find_table_metadata(client, name, case_sensitive) + ) diff --git a/mcp-server/tests/unit/tools/test_metadata.py b/mcp-server/tests/unit/tools/test_metadata.py new file mode 100755 index 00000000000..c1f8295272c --- /dev/null +++ b/mcp-server/tests/unit/tools/test_metadata.py @@ -0,0 +1,177 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +import asyncio +import json +import unittest + +from mcp_server.tools.metadata import _find_table_metadata + + +class _FakeCatalogOperation: + def __init__(self, catalogs): + self._catalogs = catalogs + + async def get_list_of_catalogs(self): + return json.dumps(self._catalogs) + + +class _FakeSchemaOperation: + def __init__(self, schemas_by_catalog): + self._schemas_by_catalog = schemas_by_catalog + + async def get_list_of_schemas(self, catalog_name): + entry = self._schemas_by_catalog[catalog_name] + if isinstance(entry, Exception): + raise entry + return json.dumps(entry) + + +class _FakeTableOperation: + def __init__(self, tables_by_schema): + self._tables_by_schema = tables_by_schema + + async def get_list_of_tables(self, catalog_name, schema_name): + entry = self._tables_by_schema[(catalog_name, schema_name)] + if isinstance(entry, Exception): + raise entry + return json.dumps(entry) + + +class _FakeClient: + """Minimal stand-in exposing only the three operations the crawl uses.""" + + def __init__(self, catalogs, schemas_by_catalog, tables_by_schema): + self._catalog_operation = _FakeCatalogOperation(catalogs) + self._schema_operation = _FakeSchemaOperation(schemas_by_catalog) + self._table_operation = _FakeTableOperation(tables_by_schema) + + def as_catalog_operation(self): + return self._catalog_operation + + def as_schema_operation(self): + return self._schema_operation + + def as_table_operation(self): + return self._table_operation + + +def _sample_client(): + """A metalake with two relational catalogs, one fileset catalog to be + ignored, and one schema that fails to list (managed by multiple catalogs). + """ + catalogs = [ + {"name": "catalog_postgres", "type": "relational"}, + {"name": "catalog_hive", "type": "relational"}, + {"name": "catalog_fileset", "type": "fileset"}, + ] + schemas_by_catalog = { + "catalog_postgres": [{"name": "hr"}, {"name": "public"}], + "catalog_hive": [{"name": "product"}, {"name": "sales"}], + } + tables_by_schema = { + ("catalog_postgres", "hr"): [{"name": "employees"}, {"name": "salaries"}], + ("catalog_postgres", "public"): [], + ("catalog_hive", "product"): [{"name": "Employees"}], + ("catalog_hive", "sales"): Exception( + "Schema managed by multiple catalogs" + ), + } + return _FakeClient(catalogs, schemas_by_catalog, tables_by_schema) + + +class TestFindTableMetadata(unittest.TestCase): + """Unit tests for the find_metadata crawl helper.""" + + def test_case_insensitive_match_across_catalogs(self): + """A case-insensitive search returns hits from every relational + catalog, including a name that differs only by case.""" + result = asyncio.run( + _find_table_metadata(_sample_client(), "employees", False) + ) + full_names = {match["fullName"] for match in result["matches"]} + self.assertEqual( + full_names, + { + "catalog_postgres.hr.employees", + "catalog_hive.product.Employees", + }, + ) + + def test_fileset_catalog_is_ignored(self): + """Only relational catalogs are crawled; the fileset catalog is not + searched and does not appear in the counts.""" + result = asyncio.run( + _find_table_metadata(_sample_client(), "employees", False) + ) + self.assertEqual(result["searched"]["catalogs"], 2) + self.assertEqual(result["searched"]["schemas"], 4) + + def test_unlistable_schema_is_skipped_not_fatal(self): + """A schema that cannot be listed is recorded under 'skipped' rather + than aborting the crawl.""" + result = asyncio.run( + _find_table_metadata(_sample_client(), "employees", False) + ) + self.assertEqual( + result["skipped"], + [ + { + "location": "catalog_hive.sales", + "reason": "Schema managed by multiple catalogs", + } + ], + ) + + def test_case_sensitive_excludes_differing_case(self): + """A case-sensitive search drops a name that differs only by case.""" + result = asyncio.run( + _find_table_metadata(_sample_client(), "employees", True) + ) + full_names = {match["fullName"] for match in result["matches"]} + self.assertEqual(full_names, {"catalog_postgres.hr.employees"}) + + def test_substring_match(self): + """Matching is a substring match by default.""" + result = asyncio.run( + _find_table_metadata(_sample_client(), "ployee", False) + ) + full_names = {match["fullName"] for match in result["matches"]} + self.assertEqual( + full_names, + { + "catalog_postgres.hr.employees", + "catalog_hive.product.Employees", + }, + ) + + def test_no_match_returns_empty(self): + """A name that matches nothing returns no matches but still reports + what was searched.""" + result = asyncio.run( + _find_table_metadata(_sample_client(), "no_such_table", False) + ) + self.assertEqual(result["matches"], []) + self.assertEqual(result["searched"]["catalogs"], 2) + + def test_empty_name_is_rejected(self): + """A blank name raises rather than matching every table.""" + with self.assertRaises(ValueError): + asyncio.run(_find_table_metadata(_sample_client(), " ", False)) + + +if __name__ == "__main__": + unittest.main()