Add create_user_tool agent operation and MCP wrapper

Validates the representation through DynamicUnprivilegedToolCreatePayload
and delegates to DynamicToolManager.create_unprivileged_tool, mirroring
the POST /api/unprivileged_tools endpoint. The MCP wrapper carries the
full GalaxyUserTool schema in its docstring (required fields, the
common container-as-string mistake, a worked example) so agents can
construct valid representations without round-tripping.

lib/galaxy/agents/operations.py

@@ -895,3 +895,14 @@ class AgentOperationsManager:
             "tools": [t.to_dict() for t in tools],
             "count": len(tools),
         }
+
+    def create_user_tool(self, representation: dict[str, Any]) -> dict[str, Any]:
+        from galaxy.tool_util_models.dynamic_tool_models import DynamicUnprivilegedToolCreatePayload
+
+        user = self.trans.user
+        if not user:
+            raise ValueError("User must be authenticated")
+
+        payload = DynamicUnprivilegedToolCreatePayload(src="representation", representation=representation)
+        dynamic_tool = self.dynamic_tools_manager.create_unprivileged_tool(user, payload)
+        return dynamic_tool.to_dict()

lib/galaxy/webapps/galaxy/api/mcp.py

@@ -447,6 +447,64 @@ def get_mcp_app(gx_app):
             ops_manager = get_operations_manager(api_key, ctx)
             return ops_manager.list_user_tools(active)
 
+    @mcp.tool()
+    def create_user_tool(representation: dict[str, Any], api_key: str, ctx: MCPContext) -> dict[str, Any]:
+        """Create a user-defined tool in Galaxy from a YAML tool definition.
+
+        User-defined tools are lightweight, containerized tools that can be
+        created without admin privileges. They are stored in the database,
+        scoped to the creating user, and can be embedded in workflows
+        (importing the workflow automatically creates the tool for the
+        importing user).
+
+        Requires the USER_TOOL_EXECUTE role on the calling user and
+        enable_beta_tool_formats=true in the Galaxy config; both are enforced
+        by the underlying manager and surface as permission/config errors here.
+
+        Args:
+            representation: The tool definition as a dictionary matching the
+                GalaxyUserTool schema. Required fields:
+                - class: "GalaxyUserTool" (exactly this string)
+                - id: tool identifier (lowercase, no spaces, 3-255 chars)
+                - version: version string (e.g. "0.1.0")
+                - name: display name shown in Galaxy tool menu
+                - container: container image as a STRING (e.g. "python:3.12-slim"),
+                  NOT a dict -- this is a common mistake
+                - shell_command: the command to execute, with $(inputs.name.path)
+                  for data inputs and $(inputs.name) for parameter inputs
+                - inputs: list of input dicts, each with "name" and "type"
+                  (type can be: "data", "integer", "float", "text", "boolean")
+                - outputs: list of output dicts, each with "name", "type": "data",
+                  "format" (e.g. "tabular", "vcf", "bed"), and "from_work_dir"
+
+        Returns:
+            Dict with the created tool's id, uuid, tool_id, active status, and
+            the validated representation.
+
+        Example:
+            create_user_tool({
+                "class": "GalaxyUserTool",
+                "id": "my_filter",
+                "version": "0.1.0",
+                "name": "My Filter",
+                "container": "python:3.12-slim",
+                "shell_command": "python3 -c 'import sys; ...'",
+                "inputs": [{"name": "input1", "type": "data", "format": "tabular"}],
+                "outputs": [
+                    {"name": "output1", "type": "data",
+                     "format": "tabular", "from_work_dir": "out.tsv"}
+                ]
+            })
+
+        NEXT STEPS:
+        - Run the tool: run_user_tool(history_id, tool_uuid, inputs)
+        - List your tools: list_user_tools()
+        - Delete a tool: delete_user_tool(uuid)
+        """
+        with _mcp_error_handler("create_user_tool"):
+            ops_manager = get_operations_manager(api_key, ctx)
+            return ops_manager.create_user_tool(representation)
+
     mcp_app = mcp.http_app(path="/")
     mcp_app.state.mcp_server = mcp
 

test/integration/test_agents.py

@@ -342,6 +342,7 @@ class TestMCPServerSmoke(IntegrationTestCase):
             "invoke_workflow",
             "get_job_status",
             "list_user_tools",
+            "create_user_tool",
         }
         assert expected.issubset(tool_names), f"Missing tools: {expected - tool_names}"
 
@@ -428,3 +429,24 @@ class TestMCPServerSmoke(IntegrationTestCase):
         data = result.data
         assert data["tools"] == []
         assert data["count"] == 0
+
+    def test_mcp_create_user_tool(self):
+        """create_user_tool() persists a UDT and returns its uuid."""
+        from fastmcp import Client
+        from galaxy_test.base.populators import TOOL_WITH_SHELL_COMMAND
+
+        mcp_server = self._get_mcp_server()
+        _, api_key = self._setup_udt_user("udt_create_user@test.com")
+
+        async def _create():
+            async with Client(mcp_server) as client:
+                return await client.call_tool(
+                    "create_user_tool",
+                    {"api_key": api_key, "representation": TOOL_WITH_SHELL_COMMAND},
+                )
+
+        result = self._run_async(_create())
+        assert not result.is_error, result
+        data = result.data
+        assert "uuid" in data
+        assert data["representation"]["name"] == TOOL_WITH_SHELL_COMMAND["name"]