[#12371] feat(mcp-server): write tools for views

docs/gravitino-mcp-server.md

@@ -128,6 +128,9 @@ Gravitino MCP server supports the following tools, and you could export tool by
 | `get_partition`                     | Retrieve a partition's metadata. Only for catalogs with a partition API.       | `partition`  |
 | `list_of_views`                     | Retrieve a list of views for a schema. Only for catalogs supporting views.     | `view`       |
 | `load_view`                         | Retrieve a view's metadata. Only for catalogs supporting views.                | `view`       |
+| `create_view`                       | Create a new view. Only for catalogs supporting views.                         | `view`       |
+| `alter_view`                        | Alter an existing view. Only for catalogs supporting views.                    | `view`       |
+| `drop_view`                         | Drop a view. Only for catalogs supporting views.                               | `view`       |
 
 
 ## Configuration

mcp-server/mcp_server/client/plain/plain_rest_client_view_operation.py

@@ -49,3 +49,63 @@ async def load_view(
             f"/views/{encode_path_segment(view_name)}"
         )
         return extract_content_from_response(response, "view", {})
+
+    # pylint: disable=too-many-positional-arguments
+    async def create_view(
+        self,
+        catalog_name: str,
+        schema_name: str,
+        name: str,
+        comment: str,
+        columns: list,
+        representations: list,
+        properties: dict,
+        default_catalog: str = None,
+        default_schema: str = None,
+    ) -> str:
+        request = {
+            "name": name,
+            "comment": comment,
+            "columns": columns,
+            "representations": representations,
+            "properties": properties,
+        }
+        optional_fields = {
+            "defaultCatalog": default_catalog,
+            "defaultSchema": default_schema,
+        }
+        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)}/views",
+            json=request,
+        )
+        return extract_content_from_response(response, "view", {})
+
+    async def alter_view(
+        self,
+        catalog_name: str,
+        schema_name: str,
+        view_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"/views/{encode_path_segment(view_name)}",
+            json={"updates": updates},
+        )
+        return extract_content_from_response(response, "view", {})
+
+    async def drop_view(
+        self, catalog_name: str, schema_name: str, view_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"/views/{encode_path_segment(view_name)}"
+        )
+        return extract_content_from_response(response, "dropped", False)

mcp-server/mcp_server/client/view_operation.py

@@ -53,3 +53,35 @@ async def load_view(
             str: JSON-formatted string containing full view metadata
         """
         pass
+
+    @abstractmethod
+    # pylint: disable=too-many-positional-arguments
+    async def create_view(
+        self,
+        catalog_name: str,
+        schema_name: str,
+        name: str,
+        comment: str,
+        columns: list,
+        representations: list,
+        properties: dict,
+        default_catalog: str = None,
+        default_schema: str = None,
+    ) -> str:
+        pass
+
+    @abstractmethod
+    async def alter_view(
+        self,
+        catalog_name: str,
+        schema_name: str,
+        view_name: str,
+        updates: list,
+    ) -> str:
+        pass
+
+    @abstractmethod
+    async def drop_view(
+        self, catalog_name: str, schema_name: str, view_name: str
+    ) -> str:
+        pass

mcp-server/mcp_server/tools/view.py

@@ -141,3 +141,114 @@ async def load_view(
         return await client.as_view_operation().load_view(
             catalog_name, schema_name, view_name
         )
+
+    @mcp.tool(tags={"view"})
+    # pylint: disable=too-many-positional-arguments
+    async def create_view(
+        ctx: Context,
+        catalog_name: str,
+        schema_name: str,
+        name: str,
+        comment: str,
+        columns: list,
+        representations: list,
+        properties: dict,
+        default_catalog: str = None,
+        default_schema: str = None,
+    ) -> str:
+        """
+        Create a new view within a schema.
+
+        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 view to create.
+            comment (str): Human-readable description.
+            columns (list): Output column definitions. Each column is a dict:
+                {
+                  "name": "dt",
+                  "type": "date",
+                  "comment": "partition date",
+                  "nullable": true
+                }
+            representations (list): Engine-specific definitions. At least one
+                is required, and SQL dialects must not repeat. Each
+                representation is a dict:
+                {"type": "sql", "dialect": "spark", "sql": "SELECT ..."}
+            properties (dict): View properties.
+            default_catalog (str): Optional catalog used to resolve
+                unqualified identifiers in the representations.
+            default_schema (str): Optional schema used to resolve unqualified
+                identifiers in the representations.
+
+        Returns:
+            str: JSON-formatted string containing the created view.
+        """
+        client = ctx.request_context.lifespan_context.rest_client()
+        return await client.as_view_operation().create_view(
+            catalog_name,
+            schema_name,
+            name,
+            comment,
+            columns,
+            representations,
+            properties,
+            default_catalog,
+            default_schema,
+        )
+
+    @mcp.tool(tags={"view"})
+    async def alter_view(
+        ctx: Context,
+        catalog_name: str,
+        schema_name: str,
+        view_name: str,
+        updates: list,
+    ) -> str:
+        """
+        Alter an existing view.
+
+        Args:
+            ctx (Context): The request context object.
+            catalog_name (str): Name of the catalog.
+            schema_name (str): Name of the schema.
+            view_name (str): Name of the view to alter.
+            updates (list): List of update operations. Example:
+                [
+                  {"@type": "rename", "newName": "renamed"},
+                  {"@type": "setProperty", "property": "k", "value": "v"},
+                  {"@type": "removeProperty", "property": "k"},
+                  {"@type": "replaceView", "comment": "rewritten",
+                   "columns": [], "representations": []}
+                ]
+
+        Returns:
+            str: JSON-formatted string containing the altered view.
+        """
+        client = ctx.request_context.lifespan_context.rest_client()
+        return await client.as_view_operation().alter_view(
+            catalog_name, schema_name, view_name, updates
+        )
+
+    @mcp.tool(tags={"view"})
+    async def drop_view(
+        ctx: Context, catalog_name: str, schema_name: str, view_name: str
+    ) -> str:
+        """
+        Drop a view by its name.
+
+        Args:
+            ctx (Context): The request context object.
+            catalog_name (str): Name of the catalog.
+            schema_name (str): Name of the schema.
+            view_name (str): Name of the view to drop.
+
+        Returns:
+            str: JSON-formatted string indicating whether the view was
+                dropped.
+        """
+        client = ctx.request_context.lifespan_context.rest_client()
+        return await client.as_view_operation().drop_view(
+            catalog_name, schema_name, view_name
+        )

mcp-server/tests/unit/client/test_url_encoding.py

@@ -569,3 +569,40 @@ def test_load_view_encodes_view_name(self):
         url = _called_url(client.get)
         self.assertIn(_ENCODED_QUERY_INJECTION, url)
         self.assertNotIn("?admin=true", url)
+
+    def test_create_view_omits_unset_default_catalog_and_schema(self):
+        client = _make_mock_client({"view": {}})
+        op = PlainRESTClientViewOperation(METALAKE, client)
+        asyncio.run(op.create_view("catalog", "schema", "v", "", [], [], {}))
+        request = client.post.call_args[1]["json"]
+        self.assertNotIn("defaultCatalog", request)
+        self.assertNotIn("defaultSchema", request)
+
+    def test_create_view_includes_defaults_when_set(self):
+        client = _make_mock_client({"view": {}})
+        op = PlainRESTClientViewOperation(METALAKE, client)
+        asyncio.run(
+            op.create_view(
+                "catalog", "schema", "v", "", [], [], {}, "cat", "sch"
+            )
+        )
+        request = client.post.call_args[1]["json"]
+        self.assertEqual("cat", request["defaultCatalog"])
+        self.assertEqual("sch", request["defaultSchema"])
+
+    def test_alter_view_encodes_view_name(self):
+        client = _make_mock_client({"view": {}})
+        op = PlainRESTClientViewOperation(METALAKE, client)
+        asyncio.run(op.alter_view("catalog", "schema", _QUERY_INJECTION, []))
+        url = _called_url(client.put)
+        self.assertIn(_ENCODED_QUERY_INJECTION, url)
+        self.assertNotIn("?admin=true", url)
+
+    def test_drop_view_reads_dropped_response(self):
+        client = _make_mock_client({"code": 0, "dropped": True})
+        op = PlainRESTClientViewOperation(METALAKE, client)
+        result = asyncio.run(op.drop_view("catalog", "schema", _PATH_TRAVERSAL))
+        self.assertEqual("true", result)
+        url = _called_url(client.delete)
+        self.assertIn(_ENCODED_PATH_TRAVERSAL, url)
+        self.assertNotIn("../../", url)

mcp-server/tests/unit/tools/mock_operation.py

@@ -484,3 +484,33 @@ async def load_view(
         self, catalog_name: str, schema_name: str, view_name: str
     ) -> str:
         return f"mock_view: {catalog_name}, {schema_name}, {view_name}"
+
+    # pylint: disable=too-many-positional-arguments
+    async def create_view(
+        self,
+        catalog_name,
+        schema_name,
+        name,
+        comment,
+        columns,
+        representations,
+        properties,
+        default_catalog=None,
+        default_schema=None,
+    ) -> str:
+        return (
+            f"mock_view_created: {catalog_name}.{schema_name}.{name} "
+            f"with representations {representations}, "
+            f"default_catalog={default_catalog}"
+        )
+
+    async def alter_view(
+        self, catalog_name, schema_name, view_name, updates
+    ) -> str:
+        return (
+            f"mock_view_altered: {catalog_name}.{schema_name}.{view_name} "
+            f"with updates {updates}"
+        )
+
+    async def drop_view(self, catalog_name, schema_name, view_name) -> str:
+        return f"mock_view_dropped: {catalog_name}.{schema_name}.{view_name}"

mcp-server/tests/unit/tools/test_view.py

@@ -66,3 +66,96 @@ async def _test_load_view(mcp_server):
                 )
 
         asyncio.run(_test_load_view(self.mcp))
+
+    def test_create_view(self):
+        async def _test(mcp_server):
+            representations = [
+                {"type": "sql", "dialect": "spark", "sql": "SELECT 1"}
+            ]
+            async with Client(mcp_server) as client:
+                result = await client.call_tool(
+                    "create_view",
+                    {
+                        "catalog_name": "cat",
+                        "schema_name": "sch",
+                        "name": "v",
+                        "comment": "c",
+                        "columns": [{"name": "id", "type": "integer"}],
+                        "representations": representations,
+                        "properties": {"k": "v"},
+                    },
+                )
+                self.assertEqual(
+                    f"mock_view_created: cat.sch.v "
+                    f"with representations {representations}, "
+                    "default_catalog=None",
+                    result.content[0].text,
+                )
+
+        asyncio.run(_test(self.mcp))
+
+    def test_create_view_with_defaults(self):
+        async def _test(mcp_server):
+            representations = [
+                {"type": "sql", "dialect": "spark", "sql": "SELECT 1"}
+            ]
+            async with Client(mcp_server) as client:
+                result = await client.call_tool(
+                    "create_view",
+                    {
+                        "catalog_name": "cat",
+                        "schema_name": "sch",
+                        "name": "v",
+                        "comment": "c",
+                        "columns": [],
+                        "representations": representations,
+                        "properties": {},
+                        "default_catalog": "dc",
+                        "default_schema": "ds",
+                    },
+                )
+                self.assertEqual(
+                    f"mock_view_created: cat.sch.v "
+                    f"with representations {representations}, "
+                    "default_catalog=dc",
+                    result.content[0].text,
+                )
+
+        asyncio.run(_test(self.mcp))
+
+    def test_alter_view(self):
+        async def _test(mcp_server):
+            updates = [{"@type": "rename", "newName": "renamed"}]
+            async with Client(mcp_server) as client:
+                result = await client.call_tool(
+                    "alter_view",
+                    {
+                        "catalog_name": "cat",
+                        "schema_name": "sch",
+                        "view_name": "v",
+                        "updates": updates,
+                    },
+                )
+                self.assertEqual(
+                    f"mock_view_altered: cat.sch.v with updates {updates}",
+                    result.content[0].text,
+                )
+
+        asyncio.run(_test(self.mcp))
+
+    def test_drop_view(self):
+        async def _test(mcp_server):
+            async with Client(mcp_server) as client:
+                result = await client.call_tool(
+                    "drop_view",
+                    {
+                        "catalog_name": "cat",
+                        "schema_name": "sch",
+                        "view_name": "v",
+                    },
+                )
+                self.assertEqual(
+                    "mock_view_dropped: cat.sch.v", result.content[0].text
+                )
+
+        asyncio.run(_test(self.mcp))