This is an automated email from the ASF dual-hosted git repository.
mchades pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new fc1c4a2ede [#11567][#11574] feat(mcp-server): write tools for tag,
catalog, schema, table, model, topic, fileset (#11621)
fc1c4a2ede is described below
commit fc1c4a2edee83fd7d9228ea390b951e4652595ee
Author: Qi Yu <[email protected]>
AuthorDate: Thu Jul 9 13:02:24 2026 +0800
[#11567][#11574] feat(mcp-server): write tools for tag, catalog, schema,
table, model, topic, fileset (#11621)
### What changes were proposed in this pull request?
This is the **write/modify** PR (second in the stack), adding governed
write tools to the MCP server.
- **#11567**: enable tag write tools (create/alter/delete/associate),
protected by Gravitino authorization.
- **#11574**: implement write operations for catalog, schema, table,
model, topic, and fileset (create/alter/drop, plus register/link/delete
for models).
Write tools are exposed by default; authorization is enforced by
Gravitino core.
> **Stacked on #11622.** This PR depends on the
authentication/authorization/audit
> foundation in #11622. Until #11622 merges, this PR's diff will also
show those
> commits; review after #11622 or focus on the two write commits here.
Merge order:
> #11622 first, then this PR.
### Why are the changes needed?
The MCP server previously exposed only read-only tools. These write
tools let agents manage Gravitino metadata through MCP, with per-request
authorization (from #11622) ensuring each principal's permissions are
enforced.
Fix: #11567
Fix: #11574
### Does this PR introduce _any_ user-facing change?
Yes. New MCP write tools for tag, catalog, schema, table, model, topic,
and fileset.
### How was this patch tested?
`./gradlew :mcp-server:formatCheckPython :mcp-server:pylint` clean
(pylint 10.00/10) and 123 unit tests pass.
---
docs/gravitino-mcp-server.md | 31 ++-
mcp-server/mcp_server/client/catalog_operation.py | 24 ++
mcp-server/mcp_server/client/fileset_operation.py | 31 +++
mcp-server/mcp_server/client/model_operation.py | 78 +++++++
.../plain/plain_rest_client_catalog_operation.py | 46 ++++
.../plain/plain_rest_client_fileset_operation.py | 68 +++++-
.../plain/plain_rest_client_model_operation.py | 147 ++++++++++++-
.../plain/plain_rest_client_schema_operation.py | 32 +++
.../plain/plain_rest_client_table_operation.py | 63 ++++++
.../plain/plain_rest_client_tag_operation.py | 9 +-
.../plain/plain_rest_client_topic_operation.py | 53 ++++-
mcp-server/mcp_server/client/plain/utils.py | 6 +
mcp-server/mcp_server/client/schema_operation.py | 18 ++
mcp-server/mcp_server/client/table_operation.py | 33 +++
mcp-server/mcp_server/client/tag_operation.py | 4 +-
mcp-server/mcp_server/client/topic_operation.py | 28 +++
mcp-server/mcp_server/tools/catalog.py | 109 +++++++++
mcp-server/mcp_server/tools/fileset.py | 104 ++++++++-
mcp-server/mcp_server/tools/model.py | 243 +++++++++++++++++++++
mcp-server/mcp_server/tools/schema.py | 79 +++++++
mcp-server/mcp_server/tools/table.py | 124 +++++++++++
mcp-server/mcp_server/tools/tag.py | 18 +-
mcp-server/mcp_server/tools/topic.py | 83 +++++++
mcp-server/tests/integration/test_authz_e2e.py | 9 -
mcp-server/tests/unit/client/test_url_encoding.py | 139 +++++++++++-
mcp-server/tests/unit/tools/mock_operation.py | 182 +++++++++++++++
mcp-server/tests/unit/tools/test_catalog.py | 87 ++++++++
mcp-server/tests/unit/tools/test_fileset.py | 84 +++++++
mcp-server/tests/unit/tools/test_model.py | 162 ++++++++++++++
mcp-server/tests/unit/tools/test_schema.py | 69 ++++++
mcp-server/tests/unit/tools/test_table.py | 104 +++++++++
mcp-server/tests/unit/tools/test_tag.py | 52 ++++-
mcp-server/tests/unit/tools/test_topic.py | 56 +++++
33 files changed, 2319 insertions(+), 56 deletions(-)
diff --git a/docs/gravitino-mcp-server.md b/docs/gravitino-mcp-server.md
index 1940e6d893..7d09a53730 100644
--- a/docs/gravitino-mcp-server.md
+++ b/docs/gravitino-mcp-server.md
@@ -64,20 +64,44 @@ Gravitino MCP server supports the following tools, and you
could export tool by
| Tool name | Description
| Tag | Since version |
|-------------------------------------|--------------------------------------------------------------------------------|--------------|---------------|
| `get_list_of_catalogs` | Retrieve a list of all catalogs in the
system. | `catalog` | 1.0.0 |
+| `create_catalog` | Create a new catalog.
| `catalog` | 2.0.0 |
+| `alter_catalog` | Alter an existing catalog.
| `catalog` | 2.0.0 |
+| `drop_catalog` | Drop a catalog.
| `catalog` | 2.0.0 |
+| `set_catalog_in_use` | Enable or disable a catalog.
| `catalog` | 2.0.0 |
| `get_list_of_schemas` | Retrieve a list of schemas belonging
to a specific catalog. | `schema` | 1.0.0 |
+| `create_schema` | Create a new schema.
| `schema` | 2.0.0 |
+| `alter_schema` | Alter an existing schema.
| `schema` | 2.0.0 |
+| `drop_schema` | Drop a schema.
| `schema` | 2.0.0 |
| `get_list_of_tables` | Retrieve a list of tables within a
specific catalog and schema. | `table` | 1.0.0 |
| `get_table_metadata_details` | Retrieve comprehensive metadata
details for a specific table. | `table` | 1.0.0 |
+| `create_table` | Create a new table.
| `table` | 2.0.0 |
+| `alter_table` | Alter an existing table.
| `table` | 2.0.0 |
+| `drop_table` | Drop a table.
| `table` | 2.0.0 |
| `list_of_models` | Retrieve a list of models within a
specific catalog and schema. | `model` | 1.0.0 |
-| `load_model` | Retrieve comprehensive metadata
details for a specific model. | `model` | 1.0.0 |
+| `load_model` | Retrieve comprehensive metadata
details for a specific model. | `model` | 1.0.0 |
| `list_model_versions` | Retrieve a list of versions for a
specific model. | `model` | 1.0.0 |
-| `load_model_version` | Retrieve comprehensive metadata
details for a specific model version. | `model` | 1.0.0 |
+| `load_model_version` | Retrieve comprehensive metadata
details for a specific model version. | `model` | 1.0.0 |
| `load_model_version_by_alias` | Retrieve comprehensive metadata
details for a specific model version by alias. | `model` | 1.0.0 |
+| `register_model` | Register a new model.
| `model` | 2.0.0 |
+| `delete_model` | Delete a model.
| `model` | 2.0.0 |
+| `link_model_version` | Link a new version to a model.
| `model` | 2.0.0 |
+| `delete_model_version` | Delete a model version.
| `model` | 2.0.0 |
+| `delete_model_version_by_alias` | Delete a model version by one of its
aliases. | `model` | 2.0.0 |
+| `alter_model` | Alter an existing model.
| `model` | 2.0.0 |
+| `alter_model_version` | Alter a model version.
| `model` | 2.0.0 |
+| `alter_model_version_by_alias` | Alter a model version by one of its
aliases. | `model` | 2.0.0 |
| `metadata_type_to_fullname_formats` | Retrieve the metadata type to fullname
formats mapping. | `metadata` | 1.0.0 |
| `list_of_topics` | Retrieve a list of topics within a
specific catalog and schema. | `topic` | 1.0.0 |
| `load_topic` | Retrieve comprehensive metadata
details for a specific topic. | `topic` | 1.0.0 |
+| `create_topic` | Create a new topic.
| `topic` | 2.0.0 |
+| `alter_topic` | Alter an existing topic.
| `topic` | 2.0.0 |
+| `delete_topic` | Delete a topic.
| `topic` | 2.0.0 |
| `list_of_filesets` | Retrieve a list of filesets within a
specific catalog and schema. | `fileset` | 1.0.0 |
| `load_fileset` | Retrieve comprehensive metadata
details for a specific fileset. | `fileset` | 1.0.0 |
| `list_files_in_fileset` | Retrieve a list of files within a
specific fileset. | `fileset` | 1.0.0 |
+| `create_fileset` | Create a new fileset.
| `fileset` | 2.0.0 |
+| `alter_fileset` | Alter an existing fileset.
| `fileset` | 2.0.0 |
+| `drop_fileset` | Drop a fileset.
| `fileset` | 2.0.0 |
| `list_of_jobs` | Retrieve a list of jobs
| `job` | 1.0.0 |
| `get_job_by_id` | Retrieve a job by its ID.
| `job` | 1.0.0 |
| `list_of_job_templates` | Retrieve a list of job templates.
| `job` | 1.0.0 |
@@ -90,6 +114,9 @@ Gravitino MCP server supports the following tools, and you
could export tool by
| `list_metadata_by_tag` | Retrieve a list of metadata items
associated with a specific tag. | `tag` | 1.0.0 |
| `associate_tag_with_metadata` | Associate tags with a specific
metadata item. | `tag` | 1.0.0 |
| `disassociate_tag_from_metadata` | Disassociate tags from a specific
metadata item. | `tag` | 1.0.0 |
+| `create_tag` | Create a new tag.
| `tag` | 2.0.0 |
+| `alter_tag` | Alter an existing tag.
| `tag` | 2.0.0 |
+| `delete_tag` | Delete a tag.
| `tag` | 2.0.0 |
| `list_statistics_for_metadata` | Retrieve a list of statistics
associated with a specific metadata item. | `statistics` | 1.0.0
|
| `list_statistics_for_partition` | Retrieve a list of statistics
associated with a specific partition. | `statistics` | 1.0.0
|
| `get_list_of_policies` | Retrieve a list of policies in the
system. | `policy` | 1.0.0 |
diff --git a/mcp-server/mcp_server/client/catalog_operation.py
b/mcp-server/mcp_server/client/catalog_operation.py
index 08cf7a5b97..9849f9e439 100644
--- a/mcp-server/mcp_server/client/catalog_operation.py
+++ b/mcp-server/mcp_server/client/catalog_operation.py
@@ -32,3 +32,27 @@ class CatalogOperation(ABC):
str: JSON-formatted string containing catalog information.
"""
pass
+
+ @abstractmethod
+ # pylint: disable=too-many-positional-arguments
+ async def create_catalog(
+ self,
+ name: str,
+ catalog_type: str,
+ provider: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def alter_catalog(self, catalog_name: str, updates: list) -> str:
+ pass
+
+ @abstractmethod
+ async def drop_catalog(self, catalog_name: str, force: bool) -> str:
+ pass
+
+ @abstractmethod
+ async def set_catalog_in_use(self, catalog_name: str, in_use: bool) -> str:
+ pass
diff --git a/mcp-server/mcp_server/client/fileset_operation.py
b/mcp-server/mcp_server/client/fileset_operation.py
index a158c11cda..fd1922ab74 100644
--- a/mcp-server/mcp_server/client/fileset_operation.py
+++ b/mcp-server/mcp_server/client/fileset_operation.py
@@ -80,3 +80,34 @@ class FilesetOperation(ABC):
str: JSON-formatted string containing list of files in the fileset
"""
pass
+
+ @abstractmethod
+ # pylint: disable=too-many-positional-arguments
+ async def create_fileset(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ fileset_type: str,
+ storage_location: str,
+ comment: str,
+ properties: dict,
+ storage_locations: dict = None,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def alter_fileset(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ fileset_name: str,
+ updates: list,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def drop_fileset(
+ self, catalog_name: str, schema_name: str, fileset_name: str
+ ) -> str:
+ pass
diff --git a/mcp-server/mcp_server/client/model_operation.py
b/mcp-server/mcp_server/client/model_operation.py
index c63eda326a..45d5b520e3 100644
--- a/mcp-server/mcp_server/client/model_operation.py
+++ b/mcp-server/mcp_server/client/model_operation.py
@@ -106,3 +106,81 @@ class ModelOperation(ABC):
str: JSON-formatted string containing full model version metadata
"""
pass
+
+ @abstractmethod
+ # pylint: disable=too-many-positional-arguments
+ async def register_model(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def delete_model(
+ self, catalog_name: str, schema_name: str, model_name: str
+ ) -> str:
+ pass
+
+ @abstractmethod
+ # pylint: disable=too-many-positional-arguments
+ async def link_model_version(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ uri: str,
+ aliases: list,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def delete_model_version(
+ self, catalog_name: str, schema_name: str, model_name: str, version:
int
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def delete_model_version_by_alias(
+ self, catalog_name: str, schema_name: str, model_name: str, alias: str
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def alter_model(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ updates: list,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ # pylint: disable=too-many-positional-arguments
+ async def alter_model_version(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ version: int,
+ updates: list,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ # pylint: disable=too-many-positional-arguments
+ async def alter_model_version_by_alias(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ alias: str,
+ updates: list,
+ ) -> str:
+ pass
diff --git
a/mcp-server/mcp_server/client/plain/plain_rest_client_catalog_operation.py
b/mcp-server/mcp_server/client/plain/plain_rest_client_catalog_operation.py
index 843cf3bfe7..492b3ca9d9 100644
--- a/mcp-server/mcp_server/client/plain/plain_rest_client_catalog_operation.py
+++ b/mcp-server/mcp_server/client/plain/plain_rest_client_catalog_operation.py
@@ -21,6 +21,7 @@ from mcp_server.client import CatalogOperation
from mcp_server.client.plain.utils import (
encode_path_segment,
extract_content_from_response,
+ extract_response,
)
@@ -34,3 +35,48 @@ class PlainRESTClientCatalogOperation(CatalogOperation):
f"/api/metalakes/{encode_path_segment(self.metalake_name)}/catalogs?details=true"
)
return extract_content_from_response(response, "catalogs", [])
+
+ # pylint: disable=too-many-positional-arguments
+ async def create_catalog(
+ self,
+ name: str,
+ catalog_type: str,
+ provider: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ response = await self.rest_client.post(
+
f"/api/metalakes/{encode_path_segment(self.metalake_name)}/catalogs",
+ json={
+ "name": name,
+ "type": catalog_type,
+ "provider": provider,
+ "comment": comment,
+ "properties": properties,
+ },
+ )
+ return extract_content_from_response(response, "catalog", {})
+
+ async def alter_catalog(self, catalog_name: str, updates: list) -> str:
+ response = await self.rest_client.put(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}",
+ json={"updates": updates},
+ )
+ return extract_content_from_response(response, "catalog", {})
+
+ async def drop_catalog(self, catalog_name: str, force: bool) -> str:
+ response = await self.rest_client.delete(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}",
+ params={"force": force},
+ )
+ return extract_content_from_response(response, "dropped", False)
+
+ async def set_catalog_in_use(self, catalog_name: str, in_use: bool) -> str:
+ response = await self.rest_client.patch(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}",
+ json={"inUse": in_use},
+ )
+ return extract_response(response)
diff --git
a/mcp-server/mcp_server/client/plain/plain_rest_client_fileset_operation.py
b/mcp-server/mcp_server/client/plain/plain_rest_client_fileset_operation.py
index 4c81eece1d..de95819843 100644
--- a/mcp-server/mcp_server/client/plain/plain_rest_client_fileset_operation.py
+++ b/mcp-server/mcp_server/client/plain/plain_rest_client_fileset_operation.py
@@ -16,7 +16,10 @@
# under the License.
from mcp_server.client.fileset_operation import FilesetOperation
-from mcp_server.client.plain.utils import encode_path_segment
+from mcp_server.client.plain.utils import (
+ encode_path_segment,
+ extract_content_from_response,
+)
class PlainRESTClientFilesetOperation(FilesetOperation):
@@ -32,7 +35,7 @@ class PlainRESTClientFilesetOperation(FilesetOperation):
f"/catalogs/{encode_path_segment(catalog_name)}"
f"/schemas/{encode_path_segment(schema_name)}/filesets"
)
- return response.json().get("identifiers", [])
+ return extract_content_from_response(response, "identifiers", [])
async def load_fileset(
self, catalog_name: str, schema_name: str, fileset_name: str
@@ -43,7 +46,7 @@ class PlainRESTClientFilesetOperation(FilesetOperation):
f"/schemas/{encode_path_segment(schema_name)}"
f"/filesets/{encode_path_segment(fileset_name)}"
)
- return response.json().get("fileset", {})
+ return extract_content_from_response(response, "fileset", {})
# pylint: disable=too-many-positional-arguments
async def list_files_in_fileset(
@@ -61,4 +64,61 @@ class PlainRESTClientFilesetOperation(FilesetOperation):
f"/filesets/{encode_path_segment(fileset_name)}/files",
params={"sub_path": sub_path, "location_name": location_name},
)
- return response.json().get("files", [])
+ return extract_content_from_response(response, "files", [])
+
+ # pylint: disable=too-many-positional-arguments
+ async def create_fileset(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ fileset_type: str,
+ storage_location: str,
+ comment: str,
+ properties: dict,
+ storage_locations: dict = None,
+ ) -> str:
+ request = {
+ "name": name,
+ "type": fileset_type,
+ "comment": comment,
+ "properties": properties,
+ }
+ if storage_location:
+ request["storageLocation"] = storage_location
+ if storage_locations:
+ request["storageLocations"] = storage_locations
+ response = await self.rest_client.post(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}/filesets",
+ json=request,
+ )
+ return extract_content_from_response(response, "fileset", {})
+
+ async def alter_fileset(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ fileset_name: str,
+ updates: list,
+ ) -> str:
+ response = await self.rest_client.put(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/filesets/{encode_path_segment(fileset_name)}",
+ json={"updates": updates},
+ )
+ return extract_content_from_response(response, "fileset", {})
+
+ async def drop_fileset(
+ self, catalog_name: str, schema_name: str, fileset_name: str
+ ) -> str:
+ response = await self.rest_client.delete(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/filesets/{encode_path_segment(fileset_name)}"
+ )
+ return extract_content_from_response(response, "dropped", False)
diff --git
a/mcp-server/mcp_server/client/plain/plain_rest_client_model_operation.py
b/mcp-server/mcp_server/client/plain/plain_rest_client_model_operation.py
index 7fce30bd91..466dcb423c 100644
--- a/mcp-server/mcp_server/client/plain/plain_rest_client_model_operation.py
+++ b/mcp-server/mcp_server/client/plain/plain_rest_client_model_operation.py
@@ -16,7 +16,11 @@
# under the License.
from mcp_server.client import ModelOperation
-from mcp_server.client.plain.utils import encode_path_segment
+from mcp_server.client.plain.utils import (
+ encode_path_segment,
+ extract_content_from_response,
+ extract_response,
+)
class PlainRESTClientModelOperation(ModelOperation):
@@ -34,7 +38,7 @@ class PlainRESTClientModelOperation(ModelOperation):
f"/catalogs/{encode_path_segment(catalog_name)}"
f"/schemas/{encode_path_segment(schema_name)}/models"
)
- return response.json().get("identifiers", [])
+ return extract_content_from_response(response, "identifiers", [])
async def load_model(
self, catalog_name: str, schema_name: str, model_name: str
@@ -45,7 +49,7 @@ class PlainRESTClientModelOperation(ModelOperation):
f"/schemas/{encode_path_segment(schema_name)}"
f"/models/{encode_path_segment(model_name)}"
)
- return response.json().get("model", {})
+ return extract_content_from_response(response, "model", {})
async def list_model_versions(
self, catalog_name: str, schema_name: str, model_name: str
@@ -56,7 +60,7 @@ class PlainRESTClientModelOperation(ModelOperation):
f"/schemas/{encode_path_segment(schema_name)}"
f"/models/{encode_path_segment(model_name)}/versions?details=true"
)
- return response.json().get("infos", [])
+ return extract_content_from_response(response, "infos", [])
async def load_model_version(
self, catalog_name: str, schema_name: str, model_name: str, version:
int
@@ -68,7 +72,7 @@ class PlainRESTClientModelOperation(ModelOperation):
f"/models/{encode_path_segment(model_name)}"
f"/versions/{encode_path_segment(version)}"
)
- return response.json().get("modelVersion", {})
+ return extract_content_from_response(response, "modelVersion", {})
async def load_model_version_by_alias(
self, catalog_name: str, schema_name: str, model_name: str, alias: str
@@ -80,4 +84,135 @@ class PlainRESTClientModelOperation(ModelOperation):
f"/models/{encode_path_segment(model_name)}"
f"/aliases/{encode_path_segment(alias)}"
)
- return response.json().get("modelVersion", {})
+ return extract_content_from_response(response, "modelVersion", {})
+
+ # pylint: disable=too-many-positional-arguments
+ async def register_model(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ response = await self.rest_client.post(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}/models",
+ json={"name": name, "comment": comment, "properties": properties},
+ )
+ return extract_content_from_response(response, "model", {})
+
+ async def delete_model(
+ self, catalog_name: str, schema_name: str, model_name: str
+ ) -> str:
+ response = await self.rest_client.delete(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/models/{encode_path_segment(model_name)}"
+ )
+ return extract_content_from_response(response, "dropped", False)
+
+ # pylint: disable=too-many-positional-arguments
+ async def link_model_version(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ uri: str,
+ aliases: list,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ response = await self.rest_client.post(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/models/{encode_path_segment(model_name)}/versions",
+ json={
+ "uri": uri,
+ "aliases": aliases,
+ "comment": comment,
+ "properties": properties,
+ },
+ )
+ return extract_response(response)
+
+ async def delete_model_version(
+ self, catalog_name: str, schema_name: str, model_name: str, version:
int
+ ) -> str:
+ response = await self.rest_client.delete(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/models/{encode_path_segment(model_name)}"
+ f"/versions/{encode_path_segment(version)}"
+ )
+ return extract_content_from_response(response, "dropped", False)
+
+ async def delete_model_version_by_alias(
+ self, catalog_name: str, schema_name: str, model_name: str, alias: str
+ ) -> str:
+ response = await self.rest_client.delete(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/models/{encode_path_segment(model_name)}"
+ f"/aliases/{encode_path_segment(alias)}"
+ )
+ return extract_content_from_response(response, "dropped", False)
+
+ async def alter_model(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ updates: list,
+ ) -> str:
+ response = await self.rest_client.put(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/models/{encode_path_segment(model_name)}",
+ json={"updates": updates},
+ )
+ return extract_content_from_response(response, "model", {})
+
+ # pylint: disable=too-many-positional-arguments
+ async def alter_model_version(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ version: int,
+ updates: list,
+ ) -> str:
+ response = await self.rest_client.put(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/models/{encode_path_segment(model_name)}"
+ f"/versions/{encode_path_segment(version)}",
+ json={"updates": updates},
+ )
+ return extract_content_from_response(response, "modelVersion", {})
+
+ # pylint: disable=too-many-positional-arguments
+ async def alter_model_version_by_alias(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ alias: str,
+ updates: list,
+ ) -> str:
+ response = await self.rest_client.put(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/models/{encode_path_segment(model_name)}"
+ f"/aliases/{encode_path_segment(alias)}",
+ json={"updates": updates},
+ )
+ return extract_content_from_response(response, "modelVersion", {})
diff --git
a/mcp-server/mcp_server/client/plain/plain_rest_client_schema_operation.py
b/mcp-server/mcp_server/client/plain/plain_rest_client_schema_operation.py
index 4c2c76ed44..fee6fe5b10 100644
--- a/mcp-server/mcp_server/client/plain/plain_rest_client_schema_operation.py
+++ b/mcp-server/mcp_server/client/plain/plain_rest_client_schema_operation.py
@@ -35,3 +35,35 @@ class PlainRESTClientSchemaOperation(SchemaOperation):
f"/catalogs/{encode_path_segment(catalog_name)}/schemas"
)
return extract_content_from_response(response, "identifiers", [])
+
+ async def create_schema(
+ self, catalog_name: str, name: str, comment: str, properties: dict
+ ) -> str:
+ response = await self.rest_client.post(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}/schemas",
+ json={"name": name, "comment": comment, "properties": properties},
+ )
+ return extract_content_from_response(response, "schema", {})
+
+ async def alter_schema(
+ self, catalog_name: str, schema_name: str, updates: list
+ ) -> str:
+ response = await self.rest_client.put(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}",
+ json={"updates": updates},
+ )
+ return extract_content_from_response(response, "schema", {})
+
+ async def drop_schema(
+ self, catalog_name: str, schema_name: str, cascade: bool
+ ) -> str:
+ response = await self.rest_client.delete(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}",
+ params={"cascade": cascade},
+ )
+ return extract_content_from_response(response, "dropped", False)
diff --git
a/mcp-server/mcp_server/client/plain/plain_rest_client_table_operation.py
b/mcp-server/mcp_server/client/plain/plain_rest_client_table_operation.py
index e1bee62cab..26ace2f05e 100644
--- a/mcp-server/mcp_server/client/plain/plain_rest_client_table_operation.py
+++ b/mcp-server/mcp_server/client/plain/plain_rest_client_table_operation.py
@@ -50,3 +50,66 @@ class PlainRESTClientTableOperation(TableOperation):
f"/tables/{encode_path_segment(table_name)}"
)
return extract_content_from_response(response, "table", {})
+
+ # pylint: disable=too-many-positional-arguments
+ async def create_table(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ comment: str,
+ columns: list,
+ properties: dict,
+ partitioning: list = None,
+ distribution: dict = None,
+ sort_orders: list = None,
+ indexes: list = None,
+ ) -> str:
+ request = {
+ "name": name,
+ "comment": comment,
+ "columns": columns,
+ "properties": properties,
+ }
+ optional_fields = {
+ "partitioning": partitioning,
+ "distribution": distribution,
+ "sortOrders": sort_orders,
+ "indexes": indexes,
+ }
+ request.update({k: v for k, v in optional_fields.items() if v})
+ response = await self.rest_client.post(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}/tables",
+ json=request,
+ )
+ return extract_content_from_response(response, "table", {})
+
+ async def alter_table(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ table_name: str,
+ updates: list,
+ ) -> str:
+ response = await self.rest_client.put(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/tables/{encode_path_segment(table_name)}",
+ json={"updates": updates},
+ )
+ return extract_content_from_response(response, "table", {})
+
+ async def drop_table(
+ self, catalog_name: str, schema_name: str, table_name: str, purge: bool
+ ) -> str:
+ response = await self.rest_client.delete(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/tables/{encode_path_segment(table_name)}",
+ params={"purge": purge},
+ )
+ return extract_content_from_response(response, "dropped", False)
diff --git
a/mcp-server/mcp_server/client/plain/plain_rest_client_tag_operation.py
b/mcp-server/mcp_server/client/plain/plain_rest_client_tag_operation.py
index b5fabf9cb1..ab2b20c04b 100644
--- a/mcp-server/mcp_server/client/plain/plain_rest_client_tag_operation.py
+++ b/mcp-server/mcp_server/client/plain/plain_rest_client_tag_operation.py
@@ -15,7 +15,6 @@
# specific language governing permissions and limitations
# under the License.
-from mcp_server.client.plain.exception import GravitinoException
from mcp_server.client.plain.utils import (
encode_path_segment,
extract_content_from_response,
@@ -62,15 +61,11 @@ class PlainRESTClientTagOperation(TagOperation):
)
return extract_content_from_response(response, "tag", {})
- async def delete_tag(self, name: str) -> None:
+ async def delete_tag(self, name: str) -> str:
response = await self.rest_client.delete(
f"/api/metalakes/{encode_path_segment(self.metalake_name)}/tags/{encode_path_segment(name)}"
)
- if response.status_code != 200:
- raise GravitinoException(
- f"Failed to delete tag {name}: {response.text}"
- )
- return None
+ return extract_content_from_response(response, "dropped", False)
async def associate_tag_with_metadata(
self,
diff --git
a/mcp-server/mcp_server/client/plain/plain_rest_client_topic_operation.py
b/mcp-server/mcp_server/client/plain/plain_rest_client_topic_operation.py
index ff04720197..68279e9adf 100644
--- a/mcp-server/mcp_server/client/plain/plain_rest_client_topic_operation.py
+++ b/mcp-server/mcp_server/client/plain/plain_rest_client_topic_operation.py
@@ -15,7 +15,10 @@
# specific language governing permissions and limitations
# under the License.
-from mcp_server.client.plain.utils import encode_path_segment
+from mcp_server.client.plain.utils import (
+ encode_path_segment,
+ extract_content_from_response,
+)
from mcp_server.client.topic_operation import TopicOperation
@@ -34,7 +37,7 @@ class PlainRESTClientTopicOperation(TopicOperation):
f"/catalogs/{encode_path_segment(catalog_name)}"
f"/schemas/{encode_path_segment(schema_name)}/topics"
)
- return response.json().get("identifiers", [])
+ return extract_content_from_response(response, "identifiers", [])
async def load_topic(
self, catalog_name: str, schema_name: str, topic_name: str
@@ -45,4 +48,48 @@ class PlainRESTClientTopicOperation(TopicOperation):
f"/schemas/{encode_path_segment(schema_name)}"
f"/topics/{encode_path_segment(topic_name)}"
)
- return response.json().get("topic", {})
+ return extract_content_from_response(response, "topic", {})
+
+ # pylint: disable=too-many-positional-arguments
+ async def create_topic(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ response = await self.rest_client.post(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}/topics",
+ json={"name": name, "comment": comment, "properties": properties},
+ )
+ return extract_content_from_response(response, "topic", {})
+
+ async def alter_topic(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ topic_name: str,
+ updates: list,
+ ) -> str:
+ response = await self.rest_client.put(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/topics/{encode_path_segment(topic_name)}",
+ json={"updates": updates},
+ )
+ return extract_content_from_response(response, "topic", {})
+
+ async def delete_topic(
+ self, catalog_name: str, schema_name: str, topic_name: str
+ ) -> str:
+ response = await self.rest_client.delete(
+ f"/api/metalakes/{encode_path_segment(self.metalake_name)}"
+ f"/catalogs/{encode_path_segment(catalog_name)}"
+ f"/schemas/{encode_path_segment(schema_name)}"
+ f"/topics/{encode_path_segment(topic_name)}"
+ )
+ return extract_content_from_response(response, "dropped", False)
diff --git a/mcp-server/mcp_server/client/plain/utils.py
b/mcp-server/mcp_server/client/plain/utils.py
index e5c998a57a..5f9b5eb41a 100644
--- a/mcp-server/mcp_server/client/plain/utils.py
+++ b/mcp-server/mcp_server/client/plain/utils.py
@@ -33,6 +33,12 @@ def extract_content_from_response(response, field: str,
default="") -> str:
return json.dumps(response_json.get(field, default))
+def extract_response(response) -> str:
+ response_json = response.json()
+ _handle_gravitino_exception(response_json)
+ return json.dumps(response_json)
+
+
def _handle_gravitino_exception(response: dict):
error_code = response.get("code", 0)
if error_code != 0:
diff --git a/mcp-server/mcp_server/client/schema_operation.py
b/mcp-server/mcp_server/client/schema_operation.py
index b90d7b96f2..438aa02e26 100644
--- a/mcp-server/mcp_server/client/schema_operation.py
+++ b/mcp-server/mcp_server/client/schema_operation.py
@@ -35,3 +35,21 @@ class SchemaOperation(ABC):
str: JSON-formatted string containing catalog information.
"""
pass
+
+ @abstractmethod
+ async def create_schema(
+ self, catalog_name: str, name: str, comment: str, properties: dict
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def alter_schema(
+ self, catalog_name: str, schema_name: str, updates: list
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def drop_schema(
+ self, catalog_name: str, schema_name: str, cascade: bool
+ ) -> str:
+ pass
diff --git a/mcp-server/mcp_server/client/table_operation.py
b/mcp-server/mcp_server/client/table_operation.py
index 2d188ebf25..d2737bd5e0 100644
--- a/mcp-server/mcp_server/client/table_operation.py
+++ b/mcp-server/mcp_server/client/table_operation.py
@@ -55,3 +55,36 @@ class TableOperation(ABC):
str: JSON-formatted string containing full table metadata
"""
pass
+
+ @abstractmethod
+ # pylint: disable=too-many-positional-arguments
+ async def create_table(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ comment: str,
+ columns: list,
+ properties: dict,
+ partitioning: list = None,
+ distribution: dict = None,
+ sort_orders: list = None,
+ indexes: list = None,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def alter_table(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ table_name: str,
+ updates: list,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def drop_table(
+ self, catalog_name: str, schema_name: str, table_name: str, purge: bool
+ ) -> str:
+ pass
diff --git a/mcp-server/mcp_server/client/tag_operation.py
b/mcp-server/mcp_server/client/tag_operation.py
index afb4df7b32..43614ec497 100644
--- a/mcp-server/mcp_server/client/tag_operation.py
+++ b/mcp-server/mcp_server/client/tag_operation.py
@@ -78,7 +78,7 @@ class TagOperation(ABC):
pass
@abstractmethod
- async def delete_tag(self, name: str) -> None:
+ async def delete_tag(self, name: str) -> str:
"""
Delete a tag by its name.
@@ -86,7 +86,7 @@ class TagOperation(ABC):
name: Name of the tag to delete
Returns:
- None
+ str: JSON-formatted string indicating whether the tag was deleted
"""
pass
diff --git a/mcp-server/mcp_server/client/topic_operation.py
b/mcp-server/mcp_server/client/topic_operation.py
index 6bc0bc94ab..7a90ae960c 100644
--- a/mcp-server/mcp_server/client/topic_operation.py
+++ b/mcp-server/mcp_server/client/topic_operation.py
@@ -53,3 +53,31 @@ class TopicOperation(ABC):
str: JSON-formatted string containing full topic metadata
"""
pass
+
+ @abstractmethod
+ # pylint: disable=too-many-positional-arguments
+ async def create_topic(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def alter_topic(
+ self,
+ catalog_name: str,
+ schema_name: str,
+ topic_name: str,
+ updates: list,
+ ) -> str:
+ pass
+
+ @abstractmethod
+ async def delete_topic(
+ self, catalog_name: str, schema_name: str, topic_name: str
+ ) -> str:
+ pass
diff --git a/mcp-server/mcp_server/tools/catalog.py
b/mcp-server/mcp_server/tools/catalog.py
index 50115610e8..b1b59f4015 100644
--- a/mcp-server/mcp_server/tools/catalog.py
+++ b/mcp-server/mcp_server/tools/catalog.py
@@ -108,3 +108,112 @@ def load_catalog_tools(mcp: FastMCP):
"""
client = ctx.request_context.lifespan_context.rest_client()
return await client.as_catalog_operation().get_list_of_catalogs()
+
+ @mcp.tool(tags={"catalog"})
+ # pylint: disable=too-many-positional-arguments
+ async def create_catalog(
+ ctx: Context,
+ name: str,
+ catalog_type: str,
+ provider: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ """
+ Create a new catalog within the metalake.
+
+ Authorization is enforced by Gravitino: a principal without the
+ required grant receives an authorization denial.
+
+ Args:
+ ctx (Context): The request context object.
+ name (str): Name of the catalog to create.
+ catalog_type (str): Catalog type, one of "relational", "fileset",
+ "messaging", "model".
+ provider (str): Provider implementation, e.g. "hive",
+ "lakehouse-iceberg", "jdbc-postgresql", "kafka". May be empty
+ for model/fileset catalogs that have a single provider.
+ comment (str): Human-readable description.
+ properties (dict): Catalog configuration properties.
+
+ Returns:
+ str: JSON-formatted string containing the created catalog.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_catalog_operation().create_catalog(
+ name, catalog_type, provider, comment, properties
+ )
+
+ @mcp.tool(tags={"catalog"})
+ async def alter_catalog(
+ ctx: Context, catalog_name: str, updates: list
+ ) -> str:
+ """
+ Alter an existing catalog.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog to alter.
+ updates (list): List of update operations. Example:
+ [
+ {"@type": "rename", "newName": "new_catalog"},
+ {"@type": "updateComment", "newComment": "updated"},
+ {"@type": "setProperty", "property": "k", "value": "v"},
+ {"@type": "removeProperty", "property": "k"}
+ ]
+
+ Returns:
+ str: JSON-formatted string containing the altered catalog.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_catalog_operation().alter_catalog(
+ catalog_name, updates
+ )
+
+ @mcp.tool(tags={"catalog"})
+ async def drop_catalog(
+ ctx: Context, catalog_name: str, force: bool = False
+ ) -> str:
+ """
+ Drop a catalog by its name.
+
+ A catalog that is in use (enabled) cannot be dropped unless `force`
+ is true; disable it first via `set_catalog_in_use` or pass
+ `force=True`.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog to drop.
+ force (bool): Whether to drop the catalog even if it is in use
+ or non-empty. Defaults to False.
+
+ Returns:
+ str: JSON-formatted string indicating whether the catalog was
dropped.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_catalog_operation().drop_catalog(
+ catalog_name, force
+ )
+
+ @mcp.tool(tags={"catalog"})
+ async def set_catalog_in_use(
+ ctx: Context, catalog_name: str, in_use: bool
+ ) -> str:
+ """
+ Enable or disable a catalog.
+
+ A disabled catalog rejects all operations except drop; a catalog must
+ be disabled (or force-dropped) before it can be dropped.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog to enable or disable.
+ in_use (bool): True to enable the catalog, False to disable it.
+
+ Returns:
+ str: JSON-formatted string containing the base response.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_catalog_operation().set_catalog_in_use(
+ catalog_name, in_use
+ )
diff --git a/mcp-server/mcp_server/tools/fileset.py
b/mcp-server/mcp_server/tools/fileset.py
index c3997ac16d..5aa7ec7cf6 100644
--- a/mcp-server/mcp_server/tools/fileset.py
+++ b/mcp-server/mcp_server/tools/fileset.py
@@ -96,7 +96,7 @@ def load_fileset_tools(mcp: FastMCP):
catalog_name, schema_name, fileset_name
)
- # pylint:disable=too-many-positional-arguments
+ # pylint: disable=too-many-positional-arguments
@mcp.tool(tags={"fileset"})
async def list_files_in_fileset(
ctx: Context,
@@ -143,3 +143,105 @@ def load_fileset_tools(mcp: FastMCP):
return await client.as_fileset_operation().list_files_in_fileset(
catalog_name, schema_name, fileset_name, location_name, sub_path
)
+
+ # pylint: disable=too-many-positional-arguments
+ @mcp.tool(tags={"fileset"})
+ async def create_fileset(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ fileset_type: str,
+ storage_location: str,
+ comment: str,
+ properties: dict,
+ storage_locations: dict = None,
+ ) -> str:
+ """
+ Create a new fileset within a schema.
+
+ Authorization is enforced by Gravitino: a principal without the
+ required grant receives an authorization denial.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ name (str): Name of the fileset to create.
+ fileset_type (str): Fileset type, one of "managed" or "external".
+ storage_location (str): Default storage location URI, e.g.
+ "file:/tmp/fileset1". May be empty for a managed fileset,
+ in which case the location is derived from catalog/schema
+ properties.
+ comment (str): Human-readable description.
+ properties (dict): Fileset configuration properties.
+ storage_locations (dict): Optional named storage locations for a
+ multiple-location fileset, mapping location name to URI.
+ Example: {"loc1": "s3a://bucket/fs", "loc2": "file:/tmp/fs"}
+
+ Returns:
+ str: JSON-formatted string containing the created fileset.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_fileset_operation().create_fileset(
+ catalog_name,
+ schema_name,
+ name,
+ fileset_type,
+ storage_location,
+ comment,
+ properties,
+ storage_locations,
+ )
+
+ @mcp.tool(tags={"fileset"})
+ async def alter_fileset(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ fileset_name: str,
+ updates: list,
+ ) -> str:
+ """
+ Alter an existing fileset.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ fileset_name (str): Name of the fileset to alter.
+ updates (list): List of update operations. Example:
+ [
+ {"@type": "rename", "newName": "new_fileset"},
+ {"@type": "updateComment", "newComment": "updated"},
+ {"@type": "setProperty", "property": "k", "value": "v"}
+ ]
+
+ Returns:
+ str: JSON-formatted string containing the altered fileset.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_fileset_operation().alter_fileset(
+ catalog_name, schema_name, fileset_name, updates
+ )
+
+ @mcp.tool(tags={"fileset"})
+ async def drop_fileset(
+ ctx: Context, catalog_name: str, schema_name: str, fileset_name: str
+ ) -> str:
+ """
+ Drop a fileset by its name.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ fileset_name (str): Name of the fileset to drop.
+
+ Returns:
+ str: JSON-formatted string indicating whether the fileset was
dropped.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_fileset_operation().drop_fileset(
+ catalog_name, schema_name, fileset_name
+ )
diff --git a/mcp-server/mcp_server/tools/model.py
b/mcp-server/mcp_server/tools/model.py
index 4e5f21fdb5..064f16c70c 100644
--- a/mcp-server/mcp_server/tools/model.py
+++ b/mcp-server/mcp_server/tools/model.py
@@ -230,3 +230,246 @@ def load_model_tools(mcp: FastMCP):
return await client.as_model_operation().load_model_version_by_alias(
catalog_name, schema_name, model_name, alias
)
+
+ @mcp.tool(tags={"model"})
+ # pylint: disable=too-many-positional-arguments
+ async def register_model(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ """
+ Register a new model within a schema.
+
+ Authorization is enforced by Gravitino: a principal without the
+ required grant receives an authorization denial.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ name (str): Name of the model to register.
+ comment (str): Human-readable description.
+ properties (dict): Model configuration properties.
+
+ Returns:
+ str: JSON-formatted string containing the registered model.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_model_operation().register_model(
+ catalog_name, schema_name, name, comment, properties
+ )
+
+ @mcp.tool(tags={"model"})
+ async def delete_model(
+ ctx: Context, catalog_name: str, schema_name: str, model_name: str
+ ) -> str:
+ """
+ Delete a model by its name.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ model_name (str): Name of the model to delete.
+
+ Returns:
+ str: JSON-formatted string indicating whether the model was
deleted.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_model_operation().delete_model(
+ catalog_name, schema_name, model_name
+ )
+
+ # pylint: disable=too-many-positional-arguments
+ @mcp.tool(tags={"model"})
+ async def link_model_version(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ uri: str,
+ aliases: list,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ """
+ Link a new version to an existing model.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ model_name (str): Name of the model.
+ uri (str): URI of the model version artifact.
+ aliases (list): List of string aliases for this version.
+ comment (str): Human-readable description.
+ properties (dict): Model version configuration properties.
+
+ Returns:
+ str: JSON-formatted string containing the success response.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_model_operation().link_model_version(
+ catalog_name,
+ schema_name,
+ model_name,
+ uri,
+ aliases,
+ comment,
+ properties,
+ )
+
+ @mcp.tool(tags={"model"})
+ async def delete_model_version(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ version: int,
+ ) -> str:
+ """
+ Delete a specific version of a model.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ model_name (str): Name of the model.
+ version (int): Version number to delete.
+
+ Returns:
+ str: JSON-formatted string indicating whether the version was
deleted.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_model_operation().delete_model_version(
+ catalog_name, schema_name, model_name, version
+ )
+
+ @mcp.tool(tags={"model"})
+ async def delete_model_version_by_alias(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ alias: str,
+ ) -> str:
+ """
+ Delete a model version by one of its aliases.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ model_name (str): Name of the model.
+ alias (str): Alias of the version to delete, e.g. "latest".
+
+ Returns:
+ str: JSON-formatted string indicating whether the version was
deleted.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_model_operation().delete_model_version_by_alias(
+ catalog_name, schema_name, model_name, alias
+ )
+
+ @mcp.tool(tags={"model"})
+ async def alter_model(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ updates: list,
+ ) -> str:
+ """
+ Alter an existing model.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ model_name (str): Name of the model to alter.
+ updates (list): List of update operations. Example:
+ [
+ {"@type": "rename", "newName": "new_model"},
+ {"@type": "updateComment", "newComment": "updated"},
+ {"@type": "setProperty", "property": "k", "value": "v"},
+ {"@type": "removeProperty", "property": "k"}
+ ]
+
+ Returns:
+ str: JSON-formatted string containing the altered model.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_model_operation().alter_model(
+ catalog_name, schema_name, model_name, updates
+ )
+
+ # pylint: disable=too-many-positional-arguments
+ @mcp.tool(tags={"model"})
+ async def alter_model_version(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ version: int,
+ updates: list,
+ ) -> str:
+ """
+ Alter a specific version of a model.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ model_name (str): Name of the model.
+ version (int): Version number to alter.
+ updates (list): List of update operations. Example:
+ [
+ {"@type": "updateComment", "newComment": "updated"},
+ {"@type": "setProperty", "property": "k", "value": "v"},
+ {"@type": "removeProperty", "property": "k"},
+ {"@type": "updateUri", "newUri": "s3://bucket/model-v2"},
+ {"@type": "updateAliases", "aliasesToAdd": ["prod"],
+ "aliasesToRemove": ["staging"]}
+ ]
+
+ Returns:
+ str: JSON-formatted string containing the altered model version.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_model_operation().alter_model_version(
+ catalog_name, schema_name, model_name, version, updates
+ )
+
+ # pylint: disable=too-many-positional-arguments
+ @mcp.tool(tags={"model"})
+ async def alter_model_version_by_alias(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ model_name: str,
+ alias: str,
+ updates: list,
+ ) -> str:
+ """
+ Alter a model version identified by one of its aliases.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ model_name (str): Name of the model.
+ alias (str): Alias of the version to alter, e.g. "latest".
+ updates (list): List of update operations; same format as
+ `alter_model_version`.
+
+ Returns:
+ str: JSON-formatted string containing the altered model version.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_model_operation().alter_model_version_by_alias(
+ catalog_name, schema_name, model_name, alias, updates
+ )
diff --git a/mcp-server/mcp_server/tools/schema.py
b/mcp-server/mcp_server/tools/schema.py
index 3a37353a4b..41ec06d3ce 100644
--- a/mcp-server/mcp_server/tools/schema.py
+++ b/mcp-server/mcp_server/tools/schema.py
@@ -61,3 +61,82 @@ def load_schema_tools(mcp: FastMCP):
return await client.as_schema_operation().get_list_of_schemas(
catalog_name
)
+
+ @mcp.tool(tags={"schema"})
+ async def create_schema(
+ ctx: Context,
+ catalog_name: str,
+ name: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ """
+ Create a new schema within a catalog.
+
+ Authorization is enforced by Gravitino: a principal without the
+ required grant receives an authorization denial.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog to create the schema in.
+ name (str): Name of the schema to create.
+ comment (str): Human-readable description.
+ properties (dict): Schema configuration properties.
+
+ Returns:
+ str: JSON-formatted string containing the created schema.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_schema_operation().create_schema(
+ catalog_name, name, comment, properties
+ )
+
+ @mcp.tool(tags={"schema"})
+ async def alter_schema(
+ ctx: Context, catalog_name: str, schema_name: str, updates: list
+ ) -> str:
+ """
+ Alter an existing schema.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog containing the schema.
+ schema_name (str): Name of the schema to alter.
+ updates (list): List of update operations. Example:
+ [
+ {"@type": "setProperty", "property": "k", "value": "v"},
+ {"@type": "removeProperty", "property": "k"}
+ ]
+
+ Returns:
+ str: JSON-formatted string containing the altered schema.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_schema_operation().alter_schema(
+ catalog_name, schema_name, updates
+ )
+
+ @mcp.tool(tags={"schema"})
+ async def drop_schema(
+ ctx: Context, catalog_name: str, schema_name: str, cascade: bool =
False
+ ) -> str:
+ """
+ Drop a schema by its name.
+
+ A non-empty schema cannot be dropped unless `cascade` is true.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog containing the schema.
+ schema_name (str): Name of the schema to drop.
+ cascade (bool): Whether to also drop all entities under the
+ schema. Defaults to False, in which case dropping a
+ non-empty schema fails.
+
+ Returns:
+ str: JSON-formatted string indicating whether the schema was
dropped.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_schema_operation().drop_schema(
+ catalog_name, schema_name, cascade
+ )
diff --git a/mcp-server/mcp_server/tools/table.py
b/mcp-server/mcp_server/tools/table.py
index af6c62da6f..01b861a1bd 100644
--- a/mcp-server/mcp_server/tools/table.py
+++ b/mcp-server/mcp_server/tools/table.py
@@ -257,3 +257,127 @@ def load_table_tools(mcp: FastMCP):
return await client.as_table_operation().load_table(
catalog_name, schema_name, table_name
)
+
+ @mcp.tool(tags={"table"})
+ # pylint: disable=too-many-positional-arguments
+ async def create_table(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ comment: str,
+ columns: list,
+ properties: dict,
+ partitioning: list = None,
+ distribution: dict = None,
+ sort_orders: list = None,
+ indexes: list = None,
+ ) -> str:
+ """
+ Create a new table within a schema.
+
+ Authorization is enforced by Gravitino: a principal without the
+ required grant receives an authorization denial.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ name (str): Name of the table to create.
+ comment (str): Human-readable description.
+ columns (list): List of column definitions. Each column is a dict:
+ {
+ "name": "id",
+ "type": "integer",
+ "comment": "primary id",
+ "nullable": false,
+ "autoIncrement": false
+ }
+ properties (dict): Table configuration properties.
+ partitioning (list): Optional partitioning strategies. Example:
+ [{"strategy": "identity", "fieldName": ["dt"]}]
+ distribution (dict): Optional distribution (bucketing). Example:
+ {"strategy": "hash", "number": 4,
+ "funcArgs": [{"type": "field", "fieldName": ["id"]}]}
+ sort_orders (list): Optional sort orders. Example:
+ [{"sortTerm": {"type": "field", "fieldName": ["id"]},
+ "direction": "asc", "nullOrdering": "nulls_first"}]
+ indexes (list): Optional indexes. Example:
+ [{"indexType": "primary_key", "name": "pk",
+ "fieldNames": [["id"]]}]
+
+ Returns:
+ str: JSON-formatted string containing the created table.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_table_operation().create_table(
+ catalog_name,
+ schema_name,
+ name,
+ comment,
+ columns,
+ properties,
+ partitioning,
+ distribution,
+ sort_orders,
+ indexes,
+ )
+
+ @mcp.tool(tags={"table"})
+ async def alter_table(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ table_name: str,
+ updates: list,
+ ) -> str:
+ """
+ Alter an existing table.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ table_name (str): Name of the table to alter.
+ updates (list): List of update operations. Example:
+ [
+ {"@type": "rename", "newName": "new_table"},
+ {"@type": "updateComment", "newComment": "updated"},
+ {"@type": "setProperty", "property": "k", "value": "v"}
+ ]
+
+ Returns:
+ str: JSON-formatted string containing the altered table.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_table_operation().alter_table(
+ catalog_name, schema_name, table_name, updates
+ )
+
+ @mcp.tool(tags={"table"})
+ async def drop_table(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ table_name: str,
+ purge: bool = False,
+ ) -> str:
+ """
+ Drop a table by its name.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ table_name (str): Name of the table to drop.
+ purge (bool): Whether to purge the table data in addition to
+ removing the metadata. Defaults to False. Not all catalogs
+ support purging.
+
+ Returns:
+ str: JSON-formatted string indicating whether the table was
dropped.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_table_operation().drop_table(
+ catalog_name, schema_name, table_name, purge
+ )
diff --git a/mcp-server/mcp_server/tools/tag.py
b/mcp-server/mcp_server/tools/tag.py
index f6daafa678..93eb89d62c 100644
--- a/mcp-server/mcp_server/tools/tag.py
+++ b/mcp-server/mcp_server/tools/tag.py
@@ -120,7 +120,8 @@ def load_tag_tool(mcp: FastMCP):
client = ctx.request_context.lifespan_context.rest_client()
return await client.as_tag_operation().list_of_tags()
- # Disable the alter_tag tool by default as it can be destructive.
+ # alter_tag is destructive; it is exposed and access is enforced by
Gravitino
+ # authorization rather than by hiding the tool.
@mcp.tool(tags={"tag"})
async def alter_tag(ctx: Context, name: str, updates: list) -> str:
"""
@@ -173,9 +174,10 @@ def load_tag_tool(mcp: FastMCP):
client = ctx.request_context.lifespan_context.rest_client()
return await client.as_tag_operation().alter_tag(name, updates)
- # Disable the delete_tag tool by default as it can be destructive.
+ # delete_tag is destructive; it is exposed and access is enforced by
Gravitino
+ # authorization rather than by hiding the tool.
@mcp.tool(tags={"tag"})
- async def delete_tag(ctx: Context, name: str) -> None:
+ async def delete_tag(ctx: Context, name: str) -> str:
"""
Delete a tag by its name.
@@ -185,10 +187,7 @@ def load_tag_tool(mcp: FastMCP):
name (str): Name of the tag to delete
Returns:
- None
-
- Raises:
- Exception: If the deletion fails, an exception is raised with an
error message.
+ str: JSON-formatted string indicating whether the tag was deleted.
"""
client = ctx.request_context.lifespan_context.rest_client()
return await client.as_tag_operation().delete_tag(name)
@@ -364,8 +363,3 @@ def load_tag_tool(mcp: FastMCP):
"""
client = ctx.request_context.lifespan_context.rest_client()
return await client.as_tag_operation().list_metadata_by_tag(tag_name)
-
- mcp.disable(
- names={"create_tag", "alter_tag", "delete_tag"},
- components={"tool"},
- )
diff --git a/mcp-server/mcp_server/tools/topic.py
b/mcp-server/mcp_server/tools/topic.py
index 93d2aa57d5..2455fa2fd9 100644
--- a/mcp-server/mcp_server/tools/topic.py
+++ b/mcp-server/mcp_server/tools/topic.py
@@ -132,3 +132,86 @@ def load_topic_tools(mcp: FastMCP):
return await client.as_topic_operation().load_topic(
catalog_name, schema_name, topic_name
)
+
+ @mcp.tool(tags={"topic"})
+ # pylint: disable=too-many-positional-arguments
+ async def create_topic(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ name: str,
+ comment: str,
+ properties: dict,
+ ) -> str:
+ """
+ Create a new topic within a schema.
+
+ Authorization is enforced by Gravitino: a principal without the
+ required grant receives an authorization denial.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ name (str): Name of the topic to create.
+ comment (str): Human-readable description.
+ properties (dict): Topic configuration properties.
+
+ Returns:
+ str: JSON-formatted string containing the created topic.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_topic_operation().create_topic(
+ catalog_name, schema_name, name, comment, properties
+ )
+
+ @mcp.tool(tags={"topic"})
+ async def alter_topic(
+ ctx: Context,
+ catalog_name: str,
+ schema_name: str,
+ topic_name: str,
+ updates: list,
+ ) -> str:
+ """
+ Alter an existing topic.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ topic_name (str): Name of the topic to alter.
+ updates (list): List of update operations. Example:
+ [
+ {"@type": "updateComment", "newComment": "updated"},
+ {"@type": "setProperty", "property": "k", "value": "v"}
+ ]
+
+ Returns:
+ str: JSON-formatted string containing the altered topic.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_topic_operation().alter_topic(
+ catalog_name, schema_name, topic_name, updates
+ )
+
+ @mcp.tool(tags={"topic"})
+ async def delete_topic(
+ ctx: Context, catalog_name: str, schema_name: str, topic_name: str
+ ) -> str:
+ """
+ Delete a topic by its name.
+
+ Args:
+ ctx (Context): The request context object.
+ catalog_name (str): Name of the catalog.
+ schema_name (str): Name of the schema.
+ topic_name (str): Name of the topic to delete.
+
+ Returns:
+ str: JSON-formatted string indicating whether the topic was
deleted.
+ """
+ client = ctx.request_context.lifespan_context.rest_client()
+ return await client.as_topic_operation().delete_topic(
+ catalog_name, schema_name, topic_name
+ )
diff --git a/mcp-server/tests/integration/test_authz_e2e.py
b/mcp-server/tests/integration/test_authz_e2e.py
index 79f070fadc..488e6245db 100644
--- a/mcp-server/tests/integration/test_authz_e2e.py
+++ b/mcp-server/tests/integration/test_authz_e2e.py
@@ -83,15 +83,6 @@ def test_authorization_scoped_discovery(gravitino_fixture,
integration_env):
assert admin_catalogs != bob_catalogs
[email protected](
- reason="Depends on the 'create_tag' write tool, which is disabled by
default "
- "in the MCP server (destructive-tool gating in tools/tag.py). With it
disabled "
- "the call fails as 'Unknown tool' rather than an authorization denial. No
other "
- "enabled write tool works as a drop-in here:
associate_tag/policy_with_metadata "
- "and run_job all require fixture entities that the setup does not create,
so they "
- "would fail with 'not found' instead of a denial. Re-enable this test once
the PR "
- "that enables 'create_tag' is merged."
-)
def test_write_denied_for_readonly_principal(
gravitino_fixture, integration_env
):
diff --git a/mcp-server/tests/unit/client/test_url_encoding.py
b/mcp-server/tests/unit/client/test_url_encoding.py
index 36eb7cb8df..70de5dd831 100644
--- a/mcp-server/tests/unit/client/test_url_encoding.py
+++ b/mcp-server/tests/unit/client/test_url_encoding.py
@@ -65,6 +65,7 @@ def _make_mock_client(response_json: dict):
client.post = AsyncMock(return_value=response)
client.put = AsyncMock(return_value=response)
client.delete = AsyncMock(return_value=response)
+ client.patch = AsyncMock(return_value=response)
return client
@@ -98,6 +99,20 @@ class TestCatalogOperationUrlEncoding(unittest.TestCase):
self.assertIn(_ENCODED_PATH_TRAVERSAL, url)
self.assertNotIn("../../", url)
+ def test_drop_catalog_passes_force_as_query_param(self):
+ client = _make_mock_client({"code": 0, "dropped": True})
+ op = PlainRESTClientCatalogOperation(METALAKE, client)
+ result = asyncio.run(op.drop_catalog("catalog", True))
+ self.assertEqual("true", result)
+ self.assertTrue(_called_params(client.delete)["force"])
+
+ def test_set_catalog_in_use_sends_patch_with_in_use(self):
+ client = _make_mock_client({"code": 0})
+ op = PlainRESTClientCatalogOperation(METALAKE, client)
+ result = asyncio.run(op.set_catalog_in_use("catalog", False))
+ self.assertEqual('{"code": 0}', result)
+ self.assertEqual({"inUse": False}, client.patch.call_args[1]["json"])
+
class TestSchemaOperationUrlEncoding(unittest.TestCase):
def test_get_list_of_schemas_encodes_catalog_name(self):
@@ -116,6 +131,12 @@ class TestSchemaOperationUrlEncoding(unittest.TestCase):
self.assertIn(_ENCODED_QUERY_INJECTION, url)
self.assertNotIn("?admin=true", url)
+ def test_drop_schema_passes_cascade_as_query_param(self):
+ client = _make_mock_client({"code": 0, "dropped": True})
+ op = PlainRESTClientSchemaOperation(METALAKE, client)
+ asyncio.run(op.drop_schema("catalog", "schema", True))
+ self.assertTrue(_called_params(client.delete)["cascade"])
+
class TestTableOperationUrlEncoding(unittest.TestCase):
def test_get_list_of_tables_encodes_path_traversal(self):
@@ -135,6 +156,32 @@ class TestTableOperationUrlEncoding(unittest.TestCase):
self.assertIn(_ENCODED_QUERY_INJECTION, url)
self.assertNotIn("?admin=true", url)
+ def test_create_table_omits_unset_optional_fields(self):
+ client = _make_mock_client({"table": {}})
+ op = PlainRESTClientTableOperation(METALAKE, client)
+ asyncio.run(op.create_table("catalog", "schema", "t", "", [], {}))
+ request = client.post.call_args[1]["json"]
+ self.assertNotIn("partitioning", request)
+ self.assertNotIn("distribution", request)
+ self.assertNotIn("sortOrders", request)
+ self.assertNotIn("indexes", request)
+
+ def test_create_table_includes_partitioning_when_set(self):
+ client = _make_mock_client({"table": {}})
+ op = PlainRESTClientTableOperation(METALAKE, client)
+ partitioning = [{"strategy": "identity", "fieldName": ["dt"]}]
+ asyncio.run(
+ op.create_table("catalog", "schema", "t", "", [], {}, partitioning)
+ )
+ request = client.post.call_args[1]["json"]
+ self.assertEqual(partitioning, request["partitioning"])
+
+ def test_drop_table_passes_purge_as_query_param(self):
+ client = _make_mock_client({"code": 0, "dropped": True})
+ op = PlainRESTClientTableOperation(METALAKE, client)
+ asyncio.run(op.drop_table("catalog", "schema", "t", True))
+ self.assertTrue(_called_params(client.delete)["purge"])
+
class TestModelOperationUrlEncoding(unittest.TestCase):
def test_list_of_models_encodes_path_traversal(self):
@@ -165,6 +212,71 @@ class TestModelOperationUrlEncoding(unittest.TestCase):
self.assertIn(_ENCODED_PATH_TRAVERSAL, url)
self.assertNotIn("../../", url)
+ def test_delete_model_reads_dropped_response(self):
+ client = _make_mock_client({"code": 0, "dropped": True})
+ op = PlainRESTClientModelOperation(METALAKE, client)
+ result = asyncio.run(op.delete_model("catalog", "schema", "model"))
+ self.assertEqual("true", result)
+
+ def test_link_model_version_returns_base_response(self):
+ client = _make_mock_client({"code": 0})
+ op = PlainRESTClientModelOperation(METALAKE, client)
+ result = asyncio.run(
+ op.link_model_version(
+ "catalog", "schema", "model", "s3://bucket/model", [], "", {}
+ )
+ )
+ self.assertEqual('{"code": 0}', result)
+
+ def test_delete_model_version_reads_dropped_response(self):
+ client = _make_mock_client({"code": 0, "dropped": True})
+ op = PlainRESTClientModelOperation(METALAKE, client)
+ result = asyncio.run(
+ op.delete_model_version("catalog", "schema", "model", 1)
+ )
+ self.assertEqual("true", result)
+
+ def test_delete_model_version_by_alias_encodes_alias(self):
+ client = _make_mock_client({"code": 0, "dropped": True})
+ op = PlainRESTClientModelOperation(METALAKE, client)
+ result = asyncio.run(
+ op.delete_model_version_by_alias(
+ "catalog", "schema", "model", _PATH_TRAVERSAL
+ )
+ )
+ self.assertEqual("true", result)
+ url = _called_url(client.delete)
+ self.assertIn(_ENCODED_PATH_TRAVERSAL, url)
+ self.assertNotIn("../../", url)
+
+ def test_alter_model_encodes_model_name(self):
+ client = _make_mock_client({"model": {}})
+ op = PlainRESTClientModelOperation(METALAKE, client)
+ asyncio.run(op.alter_model("catalog", "schema", _QUERY_INJECTION, []))
+ url = _called_url(client.put)
+ self.assertIn(_ENCODED_QUERY_INJECTION, url)
+ self.assertNotIn("?admin=true", url)
+
+ def test_alter_model_version_reads_model_version_response(self):
+ client = _make_mock_client({"modelVersion": {"version": 1}})
+ op = PlainRESTClientModelOperation(METALAKE, client)
+ result = asyncio.run(
+ op.alter_model_version("catalog", "schema", "model", 1, [])
+ )
+ self.assertEqual('{"version": 1}', result)
+
+ def test_alter_model_version_by_alias_encodes_alias(self):
+ client = _make_mock_client({"modelVersion": {}})
+ op = PlainRESTClientModelOperation(METALAKE, client)
+ asyncio.run(
+ op.alter_model_version_by_alias(
+ "catalog", "schema", "model", _PATH_TRAVERSAL, []
+ )
+ )
+ url = _called_url(client.put)
+ self.assertIn(_ENCODED_PATH_TRAVERSAL, url)
+ self.assertNotIn("../../", url)
+
class TestTopicOperationUrlEncoding(unittest.TestCase):
def test_list_of_topics_encodes_path_traversal(self):
@@ -185,6 +297,28 @@ class TestTopicOperationUrlEncoding(unittest.TestCase):
class TestFilesetOperationUrlEncoding(unittest.TestCase):
+ def test_create_fileset_omits_empty_storage_location(self):
+ client = _make_mock_client({"fileset": {}})
+ op = PlainRESTClientFilesetOperation(METALAKE, client)
+ asyncio.run(
+ op.create_fileset("catalog", "schema", "fs", "managed", "", "", {})
+ )
+ request = client.post.call_args[1]["json"]
+ self.assertNotIn("storageLocation", request)
+ self.assertNotIn("storageLocations", request)
+
+ def test_create_fileset_includes_storage_locations_when_set(self):
+ client = _make_mock_client({"fileset": {}})
+ op = PlainRESTClientFilesetOperation(METALAKE, client)
+ locations = {"loc1": "file:/tmp/fs1", "loc2": "file:/tmp/fs2"}
+ asyncio.run(
+ op.create_fileset(
+ "catalog", "schema", "fs", "managed", "", "", {}, locations
+ )
+ )
+ request = client.post.call_args[1]["json"]
+ self.assertEqual(locations, request["storageLocations"])
+
def test_list_of_filesets_encodes_path_traversal(self):
client = _make_mock_client({"identifiers": []})
op = PlainRESTClientFilesetOperation(METALAKE, client)
@@ -242,9 +376,10 @@ class TestTagOperationUrlEncoding(unittest.TestCase):
self.assertNotIn("?admin=true", url)
def test_delete_tag_encodes_tag_name(self):
- client = _make_mock_client({})
+ client = _make_mock_client({"code": 0, "dropped": True})
op = PlainRESTClientTagOperation(METALAKE, client)
- asyncio.run(op.delete_tag(_PATH_TRAVERSAL))
+ result = asyncio.run(op.delete_tag(_PATH_TRAVERSAL))
+ self.assertEqual("true", result)
url = _called_url(client.delete)
self.assertIn(_ENCODED_PATH_TRAVERSAL, url)
self.assertNotIn("../../", url)
diff --git a/mcp-server/tests/unit/tools/mock_operation.py
b/mcp-server/tests/unit/tools/mock_operation.py
index e1fc540442..e1953bc474 100644
--- a/mcp-server/tests/unit/tools/mock_operation.py
+++ b/mcp-server/tests/unit/tools/mock_operation.py
@@ -69,11 +69,40 @@ class MockCatalogOperation(CatalogOperation):
async def get_list_of_catalogs(self) -> str:
return "mock_catalogs"
+ # pylint: disable=too-many-positional-arguments
+ async def create_catalog(
+ self, name, catalog_type, provider, comment, properties
+ ) -> str:
+ return f"mock_catalog_created: {name}, {catalog_type}, {provider}"
+
+ async def alter_catalog(self, catalog_name, updates) -> str:
+ return f"mock_catalog_altered: {catalog_name} with updates {updates}"
+
+ async def drop_catalog(self, catalog_name, force) -> str:
+ return f"mock_catalog_dropped: {catalog_name}, force={force}"
+
+ async def set_catalog_in_use(self, catalog_name, in_use) -> str:
+ return f"mock_catalog_set_in_use: {catalog_name}, in_use={in_use}"
+
class MockSchemaOperation(SchemaOperation):
async def get_list_of_schemas(self, catalog_name: str) -> str:
return "mock_schemas"
+ async def create_schema(
+ self, catalog_name, name, comment, properties
+ ) -> str:
+ return f"mock_schema_created: {catalog_name}.{name}"
+
+ async def alter_schema(self, catalog_name, schema_name, updates) -> str:
+ return f"mock_schema_altered: {catalog_name}.{schema_name} with
updates {updates}"
+
+ async def drop_schema(self, catalog_name, schema_name, cascade) -> str:
+ return (
+ f"mock_schema_dropped: {catalog_name}.{schema_name}, "
+ f"cascade={cascade}"
+ )
+
class MockTableOperation(TableOperation):
async def get_list_of_tables(
@@ -86,6 +115,41 @@ class MockTableOperation(TableOperation):
) -> str:
return "mock_table"
+ # pylint: disable=R0917
+ async def create_table(
+ self,
+ catalog_name,
+ schema_name,
+ name,
+ comment,
+ columns,
+ properties,
+ partitioning=None,
+ distribution=None,
+ sort_orders=None,
+ indexes=None,
+ ) -> str:
+ return (
+ f"mock_table_created: {catalog_name}.{schema_name}.{name}, "
+ f"partitioning={partitioning}"
+ )
+
+ async def alter_table(
+ self, catalog_name, schema_name, table_name, updates
+ ) -> str:
+ return (
+ f"mock_table_altered: {catalog_name}.{schema_name}.{table_name} "
+ f"with updates {updates}"
+ )
+
+ async def drop_table(
+ self, catalog_name, schema_name, table_name, purge
+ ) -> str:
+ return (
+ f"mock_table_dropped: {catalog_name}.{schema_name}.{table_name}, "
+ f"purge={purge}"
+ )
+
class MockFilesetOperation(FilesetOperation):
async def list_of_filesets(
@@ -110,6 +174,38 @@ class MockFilesetOperation(FilesetOperation):
) -> str:
return "mock_files_in_fileset"
+ # pylint: disable=R0917
+ async def create_fileset(
+ self,
+ catalog_name,
+ schema_name,
+ name,
+ fileset_type,
+ storage_location,
+ comment,
+ properties,
+ storage_locations=None,
+ ) -> str:
+ return (
+ f"mock_fileset_created: {catalog_name}.{schema_name}.{name}, "
+ f"storage_locations={storage_locations}"
+ )
+
+ async def alter_fileset(
+ self, catalog_name, schema_name, fileset_name, updates
+ ) -> str:
+ return (
+ f"mock_fileset_altered:
{catalog_name}.{schema_name}.{fileset_name} "
+ f"with updates {updates}"
+ )
+
+ async def drop_fileset(
+ self, catalog_name, schema_name, fileset_name
+ ) -> str:
+ return (
+ f"mock_fileset_dropped:
{catalog_name}.{schema_name}.{fileset_name}"
+ )
+
class MockPolicyOperation(PolicyOperation):
async def associate_policy_with_metadata(
@@ -170,6 +266,75 @@ class MockModelOperation(ModelOperation):
) -> str:
return "mock_model_version_by_alias"
+ # pylint: disable=too-many-positional-arguments
+ async def register_model(
+ self, catalog_name, schema_name, name, comment, properties
+ ) -> str:
+ return f"mock_model_registered: {catalog_name}.{schema_name}.{name}"
+
+ async def delete_model(self, catalog_name, schema_name, model_name) -> str:
+ return f"mock_model_deleted: {catalog_name}.{schema_name}.{model_name}"
+
+ # pylint: disable=R0917
+ async def link_model_version(
+ self,
+ catalog_name,
+ schema_name,
+ model_name,
+ uri,
+ aliases,
+ comment,
+ properties,
+ ) -> str:
+ return (
+ f"mock_model_version_linked:
{catalog_name}.{schema_name}.{model_name} "
+ f"uri={uri} aliases={aliases}"
+ )
+
+ async def delete_model_version(
+ self, catalog_name, schema_name, model_name, version
+ ) -> str:
+ return (
+ f"mock_model_version_deleted:
{catalog_name}.{schema_name}.{model_name} "
+ f"version={version}"
+ )
+
+ async def delete_model_version_by_alias(
+ self, catalog_name, schema_name, model_name, alias
+ ) -> str:
+ return (
+ f"mock_model_version_deleted_by_alias: "
+ f"{catalog_name}.{schema_name}.{model_name} alias={alias}"
+ )
+
+ async def alter_model(
+ self, catalog_name, schema_name, model_name, updates
+ ) -> str:
+ return (
+ f"mock_model_altered: {catalog_name}.{schema_name}.{model_name} "
+ f"with updates {updates}"
+ )
+
+ # pylint: disable=R0917
+ async def alter_model_version(
+ self, catalog_name, schema_name, model_name, version, updates
+ ) -> str:
+ return (
+ f"mock_model_version_altered: "
+ f"{catalog_name}.{schema_name}.{model_name} version={version} "
+ f"with updates {updates}"
+ )
+
+ # pylint: disable=R0917
+ async def alter_model_version_by_alias(
+ self, catalog_name, schema_name, model_name, alias, updates
+ ) -> str:
+ return (
+ f"mock_model_version_altered_by_alias: "
+ f"{catalog_name}.{schema_name}.{model_name} alias={alias} "
+ f"with updates {updates}"
+ )
+
class MockTopicOperation(TopicOperation):
async def list_of_topics(self, catalog_name: str, schema_name: str) -> str:
@@ -180,6 +345,23 @@ class MockTopicOperation(TopicOperation):
) -> str:
return "mock_topic"
+ # pylint: disable=too-many-positional-arguments
+ async def create_topic(
+ self, catalog_name, schema_name, name, comment, properties
+ ) -> str:
+ return f"mock_topic_created: {catalog_name}.{schema_name}.{name}"
+
+ async def alter_topic(
+ self, catalog_name, schema_name, topic_name, updates
+ ) -> str:
+ return (
+ f"mock_topic_altered: {catalog_name}.{schema_name}.{topic_name} "
+ f"with updates {updates}"
+ )
+
+ async def delete_topic(self, catalog_name, schema_name, topic_name) -> str:
+ return f"mock_topic_deleted: {catalog_name}.{schema_name}.{topic_name}"
+
class MockTagOperation(TagOperation):
async def list_of_tags(self) -> str:
diff --git a/mcp-server/tests/unit/tools/test_catalog.py
b/mcp-server/tests/unit/tools/test_catalog.py
index ca6c03f735..6b626b9015 100644
--- a/mcp-server/tests/unit/tools/test_catalog.py
+++ b/mcp-server/tests/unit/tools/test_catalog.py
@@ -45,3 +45,90 @@ class TestCatalogTool(unittest.TestCase):
self.assertEqual("mock_catalogs", result.content[0].text)
asyncio.run(_test_list_catalogs(self.mcp))
+
+ def test_create_catalog(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "create_catalog",
+ {
+ "name": "cat1",
+ "catalog_type": "relational",
+ "provider": "hive",
+ "comment": "c",
+ "properties": {"k": "v"},
+ },
+ )
+ self.assertEqual(
+ "mock_catalog_created: cat1, relational, hive",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_alter_catalog(self):
+ async def _test(mcp_server):
+ updates = [{"@type": "rename", "newName": "cat2"}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "alter_catalog",
+ {"catalog_name": "cat1", "updates": updates},
+ )
+ self.assertEqual(
+ f"mock_catalog_altered: cat1 with updates {updates}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_drop_catalog(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "drop_catalog", {"catalog_name": "cat1"}
+ )
+ self.assertEqual(
+ "mock_catalog_dropped: cat1, force=False",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_drop_catalog_force(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "drop_catalog", {"catalog_name": "cat1", "force": True}
+ )
+ self.assertEqual(
+ "mock_catalog_dropped: cat1, force=True",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_set_catalog_in_use(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "set_catalog_in_use",
+ {"catalog_name": "cat1", "in_use": False},
+ )
+ self.assertEqual(
+ "mock_catalog_set_in_use: cat1, in_use=False",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_write_tools_exposed(self):
+ """Write tools must be present (authorization enforced by
Gravitino)."""
+
+ async def _test(mcp_server):
+ names = {t.name for t in await mcp_server.list_tools()}
+ self.assertIn("create_catalog", names)
+ self.assertIn("alter_catalog", names)
+ self.assertIn("drop_catalog", names)
+ self.assertIn("set_catalog_in_use", names)
+
+ asyncio.run(_test(self.mcp))
diff --git a/mcp-server/tests/unit/tools/test_fileset.py
b/mcp-server/tests/unit/tools/test_fileset.py
index a0c0371491..5b341f0b7a 100644
--- a/mcp-server/tests/unit/tools/test_fileset.py
+++ b/mcp-server/tests/unit/tools/test_fileset.py
@@ -75,3 +75,87 @@ class TestFilesetTool(unittest.TestCase):
)
asyncio.run(_test_list_files_in_fileset(self.mcp))
+
+ def test_create_fileset(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "create_fileset",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "name": "fs",
+ "fileset_type": "managed",
+ "storage_location": "file:/tmp/fs",
+ "comment": "c",
+ "properties": {"k": "v"},
+ },
+ )
+ self.assertEqual(
+ "mock_fileset_created: cat.sch.fs, storage_locations=None",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_create_fileset_with_multiple_locations(self):
+ async def _test(mcp_server):
+ locations = {"loc1": "file:/tmp/fs1", "loc2": "file:/tmp/fs2"}
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "create_fileset",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "name": "fs",
+ "fileset_type": "managed",
+ "storage_location": "",
+ "comment": "c",
+ "properties": {"k": "v"},
+ "storage_locations": locations,
+ },
+ )
+ self.assertEqual(
+ "mock_fileset_created: cat.sch.fs, "
+ f"storage_locations={locations}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_alter_fileset(self):
+ async def _test(mcp_server):
+ updates = [{"@type": "rename", "newName": "fs2"}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "alter_fileset",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "fileset_name": "fs",
+ "updates": updates,
+ },
+ )
+ self.assertEqual(
+ f"mock_fileset_altered: cat.sch.fs with updates {updates}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_drop_fileset(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "drop_fileset",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "fileset_name": "fs",
+ },
+ )
+ self.assertEqual(
+ "mock_fileset_dropped: cat.sch.fs", result.content[0].text
+ )
+
+ asyncio.run(_test(self.mcp))
diff --git a/mcp-server/tests/unit/tools/test_model.py
b/mcp-server/tests/unit/tools/test_model.py
index ff27e64d3b..d550c8ba44 100644
--- a/mcp-server/tests/unit/tools/test_model.py
+++ b/mcp-server/tests/unit/tools/test_model.py
@@ -106,3 +106,165 @@ class TestModelTool(unittest.TestCase):
)
asyncio.run(_test_load_model_version_by_alias(self.mcp))
+
+ def test_register_model(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "register_model",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "name": "m",
+ "comment": "c",
+ "properties": {"k": "v"},
+ },
+ )
+ self.assertEqual(
+ "mock_model_registered: cat.sch.m", result.content[0].text
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_delete_model(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "delete_model",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "model_name": "m",
+ },
+ )
+ self.assertEqual(
+ "mock_model_deleted: cat.sch.m", result.content[0].text
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_link_model_version(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "link_model_version",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "model_name": "m",
+ "uri": "s3://bucket/model",
+ "aliases": ["latest"],
+ "comment": "c",
+ "properties": {"k": "v"},
+ },
+ )
+ self.assertEqual(
+ "mock_model_version_linked: cat.sch.m "
+ "uri=s3://bucket/model aliases=['latest']",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_delete_model_version(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "delete_model_version",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "model_name": "m",
+ "version": 1,
+ },
+ )
+ self.assertEqual(
+ "mock_model_version_deleted: cat.sch.m version=1",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_delete_model_version_by_alias(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "delete_model_version_by_alias",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "model_name": "m",
+ "alias": "latest",
+ },
+ )
+ self.assertEqual(
+ "mock_model_version_deleted_by_alias: cat.sch.m "
+ "alias=latest",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_alter_model(self):
+ async def _test(mcp_server):
+ updates = [{"@type": "rename", "newName": "m2"}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "alter_model",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "model_name": "m",
+ "updates": updates,
+ },
+ )
+ self.assertEqual(
+ f"mock_model_altered: cat.sch.m with updates {updates}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_alter_model_version(self):
+ async def _test(mcp_server):
+ updates = [{"@type": "updateComment", "newComment": "c2"}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "alter_model_version",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "model_name": "m",
+ "version": 1,
+ "updates": updates,
+ },
+ )
+ self.assertEqual(
+ "mock_model_version_altered: cat.sch.m version=1 "
+ f"with updates {updates}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_alter_model_version_by_alias(self):
+ async def _test(mcp_server):
+ updates = [{"@type": "updateUri", "newUri": "s3://bucket/m2"}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "alter_model_version_by_alias",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "model_name": "m",
+ "alias": "latest",
+ "updates": updates,
+ },
+ )
+ self.assertEqual(
+ "mock_model_version_altered_by_alias: cat.sch.m "
+ f"alias=latest with updates {updates}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
diff --git a/mcp-server/tests/unit/tools/test_schema.py
b/mcp-server/tests/unit/tools/test_schema.py
index 443c3968dc..172b8d70f5 100644
--- a/mcp-server/tests/unit/tools/test_schema.py
+++ b/mcp-server/tests/unit/tools/test_schema.py
@@ -42,3 +42,72 @@ class TestSchemaTool(unittest.TestCase):
self.assertEqual("mock_schemas", result.content[0].text)
asyncio.run(_test_list_schemas(self.mcp))
+
+ def test_create_schema(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "create_schema",
+ {
+ "catalog_name": "cat",
+ "name": "sch",
+ "comment": "c",
+ "properties": {"k": "v"},
+ },
+ )
+ self.assertEqual(
+ "mock_schema_created: cat.sch", result.content[0].text
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_alter_schema(self):
+ async def _test(mcp_server):
+ updates = [{"@type": "setProperty", "property": "k", "value": "v"}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "alter_schema",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "updates": updates,
+ },
+ )
+ self.assertEqual(
+ f"mock_schema_altered: cat.sch with updates {updates}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_drop_schema(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "drop_schema",
+ {"catalog_name": "cat", "schema_name": "sch"},
+ )
+ self.assertEqual(
+ "mock_schema_dropped: cat.sch, cascade=False",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_drop_schema_cascade(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "drop_schema",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "cascade": True,
+ },
+ )
+ self.assertEqual(
+ "mock_schema_dropped: cat.sch, cascade=True",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
diff --git a/mcp-server/tests/unit/tools/test_table.py
b/mcp-server/tests/unit/tools/test_table.py
index 1b426e5512..208cfc928e 100644
--- a/mcp-server/tests/unit/tools/test_table.py
+++ b/mcp-server/tests/unit/tools/test_table.py
@@ -57,3 +57,107 @@ class TestTableTool(unittest.TestCase):
self.assertEqual("mock_table", result.content[0].text)
asyncio.run(_test_load_table(self.mcp))
+
+ def test_create_table(self):
+ async def _test(mcp_server):
+ columns = [{"name": "id", "type": "integer", "nullable": False}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "create_table",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "name": "tbl",
+ "comment": "c",
+ "columns": columns,
+ "properties": {"k": "v"},
+ },
+ )
+ self.assertEqual(
+ "mock_table_created: cat.sch.tbl, partitioning=None",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_create_table_with_partitioning(self):
+ async def _test(mcp_server):
+ columns = [{"name": "id", "type": "integer", "nullable": False}]
+ partitioning = [{"strategy": "identity", "fieldName": ["dt"]}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "create_table",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "name": "tbl",
+ "comment": "c",
+ "columns": columns,
+ "properties": {"k": "v"},
+ "partitioning": partitioning,
+ },
+ )
+ self.assertEqual(
+ "mock_table_created: cat.sch.tbl, "
+ f"partitioning={partitioning}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_alter_table(self):
+ async def _test(mcp_server):
+ updates = [{"@type": "rename", "newName": "tbl2"}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "alter_table",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "table_name": "tbl",
+ "updates": updates,
+ },
+ )
+ self.assertEqual(
+ f"mock_table_altered: cat.sch.tbl with updates {updates}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_drop_table(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "drop_table",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "table_name": "tbl",
+ },
+ )
+ self.assertEqual(
+ "mock_table_dropped: cat.sch.tbl, purge=False",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_drop_table_purge(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "drop_table",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "table_name": "tbl",
+ "purge": True,
+ },
+ )
+ self.assertEqual(
+ "mock_table_dropped: cat.sch.tbl, purge=True",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
diff --git a/mcp-server/tests/unit/tools/test_tag.py
b/mcp-server/tests/unit/tools/test_tag.py
index 1bb7046f4c..daff0df295 100644
--- a/mcp-server/tests/unit/tools/test_tag.py
+++ b/mcp-server/tests/unit/tools/test_tag.py
@@ -99,14 +99,50 @@ class TestTagTool(unittest.TestCase):
asyncio.run(_test_disassociate_tag_from_metadata(self.mcp))
- def test_destructive_tag_tools_disabled_by_default(self):
- async def _test_destructive_tag_tools_disabled_by_default(mcp_server):
+ def test_write_tag_tools_are_exposed(self):
+ """Tag write tools are registered/exposed (not hidden).
+
+ Authorization enforcement is delegated to Gravitino and covered by the
+ live integration test, not asserted here; this only checks exposure.
+ """
+
+ async def _test(mcp_server):
tool_names = {tool.name for tool in await mcp_server.list_tools()}
+ self.assertIn("create_tag", tool_names)
+ self.assertIn("alter_tag", tool_names)
+ self.assertIn("delete_tag", tool_names)
+
+ asyncio.run(_test(self.mcp))
- self.assertIn("get_tag_by_name", tool_names)
- self.assertIn("list_of_tags", tool_names)
- self.assertNotIn("create_tag", tool_names)
- self.assertNotIn("alter_tag", tool_names)
- self.assertNotIn("delete_tag", tool_names)
+ def test_create_tag(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "create_tag",
+ {"name": "t1", "comment": "c", "properties": {"k": "v"}},
+ )
+ self.assertEqual("mock_tag_created: t1",
result.content[0].text)
+
+ asyncio.run(_test(self.mcp))
+
+ def test_alter_tag(self):
+ async def _test(mcp_server):
+ updates = [{"@type": "rename", "newName": "t2"}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "alter_tag", {"name": "t1", "updates": updates}
+ )
+ self.assertEqual(
+ f"mock_tag_altered: t1 with updates {updates}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_delete_tag(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool("delete_tag", {"name": "t1"})
+ self.assertEqual("mock_tag_deleted: t1",
result.content[0].text)
- asyncio.run(_test_destructive_tag_tools_disabled_by_default(self.mcp))
+ asyncio.run(_test(self.mcp))
diff --git a/mcp-server/tests/unit/tools/test_topic.py
b/mcp-server/tests/unit/tools/test_topic.py
index 28422d0634..6f3f915bb8 100644
--- a/mcp-server/tests/unit/tools/test_topic.py
+++ b/mcp-server/tests/unit/tools/test_topic.py
@@ -57,3 +57,59 @@ class TestTopicTool(unittest.TestCase):
self.assertEqual("mock_topic", result.content[0].text)
asyncio.run(_test_load_topic(self.mcp))
+
+ def test_create_topic(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "create_topic",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "name": "tp",
+ "comment": "c",
+ "properties": {"k": "v"},
+ },
+ )
+ self.assertEqual(
+ "mock_topic_created: cat.sch.tp", result.content[0].text
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_alter_topic(self):
+ async def _test(mcp_server):
+ updates = [{"@type": "setProperty", "property": "k", "value": "v"}]
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "alter_topic",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "topic_name": "tp",
+ "updates": updates,
+ },
+ )
+ self.assertEqual(
+ f"mock_topic_altered: cat.sch.tp with updates {updates}",
+ result.content[0].text,
+ )
+
+ asyncio.run(_test(self.mcp))
+
+ def test_delete_topic(self):
+ async def _test(mcp_server):
+ async with Client(mcp_server) as client:
+ result = await client.call_tool(
+ "delete_topic",
+ {
+ "catalog_name": "cat",
+ "schema_name": "sch",
+ "topic_name": "tp",
+ },
+ )
+ self.assertEqual(
+ "mock_topic_deleted: cat.sch.tp", result.content[0].text
+ )
+
+ asyncio.run(_test(self.mcp))
