Merge pull request #22625 from dannon/agent-ops-udt-parity

Add user-defined-tool operations to the agent-operations layer

Author: mvdbeek

lib/galaxy/agents/operations.py

@@ -10,8 +10,12 @@
     Optional,
 )
 
+from sqlalchemy import select
+
 from galaxy.managers.context import ProvidesUserContext
 from galaxy.managers.hdas import HDAManager
+from galaxy.managers.tools import DynamicToolManager
+from galaxy.model import UserDynamicToolAssociation
 from galaxy.schema import (
     FilterQueryParams,
     SerializationParams,
@@ -26,9 +30,12 @@
 from galaxy.schema.schema import (
     CreateHistoryPayload,
     DatasetSourceType,
+    InvocationIndexPayload,
+    WorkflowIndexPayload,
 )
 from galaxy.schema.workflows import InvokeWorkflowPayload
 from galaxy.structured_app import MinimalManagerApp
+from galaxy.tool_util_models.dynamic_tool_models import DynamicUnprivilegedToolCreatePayload
 
 log = logging.getLogger(__name__)
 
@@ -61,6 +68,7 @@ def __init__(self, app: MinimalManagerApp, trans: ProvidesUserContext):
         self._invocations_service: Optional[Any] = None
         self._hda_manager: Optional[HDAManager] = None
         self._dataset_collections_service: Optional[Any] = None
+        self._dynamic_tools_manager: Optional[Any] = None
 
     def _encode_id(self, value: int) -> str:
         return self.trans.security.encode_id(value)
@@ -148,6 +156,12 @@ def dataset_collections_service(self):
             self._dataset_collections_service = self.app[DatasetCollectionsService]
         return self._dataset_collections_service
 
+    @property
+    def dynamic_tools_manager(self):
+        if self._dynamic_tools_manager is None:
+            self._dynamic_tools_manager = self.app[DynamicToolManager]
+        return self._dynamic_tools_manager
+
     def connect(self) -> dict[str, Any]:
         config = self.app.config
         user = self.trans.user
@@ -429,8 +443,6 @@ def list_workflows(
         show_shared: bool = True,
         search: str | None = None,
     ) -> dict[str, Any]:
-        from galaxy.webapps.galaxy.services.workflows import WorkflowIndexPayload
-
         payload = WorkflowIndexPayload(
             limit=limit,
             offset=offset,
@@ -516,8 +528,6 @@ def get_invocations(
         if history_id:
             decoded_history_id = self.trans.security.decode_id(history_id)
 
-        from galaxy.webapps.galaxy.services.invocations import InvocationIndexPayload
-
         payload = InvocationIndexPayload(
             workflow_id=decoded_workflow_id,
             history_id=decoded_history_id,
@@ -873,3 +883,67 @@ def get_user(self) -> dict[str, Any]:
             "deleted": user.deleted,
             "create_time": user.create_time.isoformat() if user.create_time else None,
         }
+
+    # ==================== User-Defined Tools (UDT) ====================
+
+    def list_user_tools(self, active: bool = True) -> dict[str, Any]:
+        user = self.trans.user
+        if not user:
+            raise ValueError("User must be authenticated")
+
+        tools = list(self.dynamic_tools_manager.list_unprivileged_tools(user, active=active))
+        return {
+            "tools": [t.to_dict() for t in tools],
+            "count": len(tools),
+        }
+
+    def create_user_tool(self, representation: dict[str, Any]) -> dict[str, Any]:
+        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()
+
+    def delete_user_tool(self, uuid: str) -> dict[str, Any]:
+        user = self.trans.user
+        if not user:
+            raise ValueError("User must be authenticated")
+
+        dynamic_tool = self.dynamic_tools_manager.get_unprivileged_tool_by_uuid(user, uuid)
+        if dynamic_tool is None:
+            raise ValueError(f"User-defined tool {uuid!r} not found")
+
+        self.dynamic_tools_manager.deactivate_unprivileged_tool(user, dynamic_tool)
+        return {"uuid": uuid, "deactivated": True}
+
+    def run_user_tool(self, history_id: str, tool_uuid: str, inputs: dict[str, Any]) -> dict[str, Any]:
+        user = self.trans.user
+        if not user:
+            raise ValueError("User must be authenticated")
+
+        dynamic_tool = self.dynamic_tools_manager.get_unprivileged_tool_by_uuid(user, tool_uuid)
+        if dynamic_tool is None:
+            raise ValueError(f"User-defined tool {tool_uuid!r} not found")
+        # UDT deactivation is per-user by design: deactivate_unprivileged_tool only
+        # flips the user-association, leaving DynamicTool.active intact so other
+        # users sharing the underlying tool aren't affected. The runtime check has
+        # to look at the association, not just dynamic_tool.active.
+        session = self.dynamic_tools_manager.session()
+        assoc_active = session.scalar(
+            select(UserDynamicToolAssociation.active).where(
+                UserDynamicToolAssociation.user_id == user.id,
+                UserDynamicToolAssociation.dynamic_tool_id == dynamic_tool.id,
+            )
+        )
+        if not dynamic_tool.active or not assoc_active:
+            raise ValueError(f"User-defined tool {tool_uuid!r} is deactivated")
+
+        payload = {
+            "history_id": history_id,
+            "tool_uuid": tool_uuid,
+            "inputs": inputs,
+        }
+        result = self.tools_service._create(self.trans, payload)
+        return self._encode_ids_in_response(result)

lib/galaxy/schema/schema.py

@@ -1637,6 +1637,10 @@ class WorkflowIndexQueryPayload(Model):
     skip_step_counts: bool = False
 
 
+class WorkflowIndexPayload(WorkflowIndexQueryPayload):
+    missing_tools: bool = False
+
+
 class JobIndexSortByEnum(str, Enum):
     create_time = "create_time"
     update_time = "update_time"
@@ -1689,6 +1693,10 @@ class InvocationIndexQueryPayload(Model):
     include_nested_invocations: bool = True
 
 
+class InvocationIndexPayload(InvocationIndexQueryPayload):
+    instance: bool = Field(default=False, description="Is provided workflow id for Workflow instead of StoredWorkflow?")
+
+
 PageSortByEnum = Literal["create_time", "title", "update_time", "username"]
 
 

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

@@ -429,6 +429,140 @@ def get_user(api_key: str, ctx: MCPContext) -> dict[str, Any]:
             ops_manager = get_operations_manager(api_key, ctx)
             return ops_manager.get_user()
 
+    # ==================== User-Defined Tools (UDT) ====================
+
+    @mcp.tool()
+    def list_user_tools(api_key: str, ctx: MCPContext, active: bool = True) -> dict[str, Any]:
+        """List user-defined tools belonging to the current user.
+
+        Args:
+            active: If True (default), only show active tools. Set False to
+                include deactivated tools.
+
+        Returns:
+            Dict with 'tools' (list of user tools, each with id, uuid, tool_id,
+            name, and active status) and 'count'.
+        """
+        with _mcp_error_handler("list_user_tools"):
+            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.tool()
+    def delete_user_tool(uuid: str, api_key: str, ctx: MCPContext) -> dict[str, Any]:
+        """Deactivate a user-defined tool. Deactivated tools are not loaded into the toolbox.
+
+        Existing job history that referenced the tool is preserved; only
+        future runs are blocked.
+
+        Args:
+            uuid: The UUID of the tool to deactivate. Get this from list_user_tools().
+
+        Returns:
+            Dict confirming deactivation: {"uuid": ..., "deactivated": True}.
+        """
+        with _mcp_error_handler("delete_user_tool"):
+            ops_manager = get_operations_manager(api_key, ctx)
+            return ops_manager.delete_user_tool(uuid)
+
+    @mcp.tool()
+    def run_user_tool(
+        history_id: str,
+        tool_uuid: str,
+        inputs: dict[str, Any],
+        api_key: str,
+        ctx: MCPContext,
+    ) -> dict[str, Any]:
+        """Run a user-defined tool by UUID, producing outputs in the given history.
+
+        Resolution happens through the tool service's standard run path,
+        which accepts tool_uuid in the payload and dispatches via the
+        toolbox's unprivileged-tool resolver -- so this is functionally a
+        UUID-keyed counterpart to run_tool().
+
+        Args:
+            history_id: Galaxy history ID where outputs will be placed.
+            tool_uuid: The UUID of the user-defined tool (from create_user_tool
+                or list_user_tools).
+            inputs: Tool input parameters keyed by input name.
+                - Dataset inputs: {"input_name": {"src": "hda", "id": "<dataset_id>"}}
+                - Collection inputs: {"input_name": {"src": "hdca", "id": "<collection_id>"}}
+                - Scalar parameters: {"param_name": value}
+
+        Returns:
+            Dict with job info (job_id, history_id, state) and output dataset IDs.
+
+        Example:
+            run_user_tool(
+                history_id="abc123",
+                tool_uuid="61d15277-a911-45ef-aa66-5385146578cc",
+                inputs={
+                    "scorer_output": {"src": "hda", "id": "59ace41fc068d3ad"},
+                    "top_tracks_per_variant": 5,
+                },
+            )
+        """
+        with _mcp_error_handler("run_user_tool"):
+            ops_manager = get_operations_manager(api_key, ctx)
+            return ops_manager.run_user_tool(history_id, tool_uuid, inputs)
+
     mcp_app = mcp.http_app(path="/")
     mcp_app.state.mcp_server = mcp
 

lib/galaxy/webapps/galaxy/api/workflows.py

@@ -73,12 +73,14 @@
     AsyncTaskResultSummary,
     ClaimLandingPayload,
     CreateWorkflowLandingRequestPayload,
+    InvocationIndexPayload,
     InvocationSortByEnum,
     InvocationsStateCounts,
     SetSlugPayload,
     ShareWithPayload,
     ShareWithStatus,
     SharingStatus,
+    WorkflowIndexPayload,
     WorkflowJobMetric,
     WorkflowLandingRequest,
     WorkflowSortByEnum,
@@ -120,15 +122,11 @@
     ServesExportStores,
 )
 from galaxy.webapps.galaxy.services.invocations import (
-    InvocationIndexPayload,
     InvocationsService,
     PrepareStoreDownloadPayload,
     WriteInvocationStoreToPayload,
 )
-from galaxy.webapps.galaxy.services.workflows import (
-    WorkflowIndexPayload,
-    WorkflowsService,
-)
+from galaxy.webapps.galaxy.services.workflows import WorkflowsService
 from galaxy.workflow.extract import extract_workflow
 from galaxy.workflow.modules import module_factory
 

lib/galaxy/webapps/galaxy/services/invocations.py

@@ -4,8 +4,6 @@
     Any,
 )
 
-from pydantic import Field
-
 from galaxy.celery.helpers import async_task_summary
 from galaxy.celery.tasks import (
     prepare_invocation_download,
@@ -51,7 +49,7 @@
     AsyncTaskResultSummary,
     BcoGenerationParametersMixin,
     ExportObjectType,
-    InvocationIndexQueryPayload,
+    InvocationIndexPayload,
     StoreExportPayload,
     WriteStoreToPayload,
 )
@@ -71,10 +69,6 @@
 log = logging.getLogger(__name__)
 
 
-class InvocationIndexPayload(InvocationIndexQueryPayload):
-    instance: bool = Field(default=False, description="Is provided workflow id for Workflow instead of StoredWorkflow?")
-
-
 class PrepareStoreDownloadPayload(StoreExportPayload, BcoGenerationParametersMixin):
     pass
 

lib/galaxy/webapps/galaxy/services/workflows.py

@@ -29,7 +29,7 @@
 from galaxy.schema.invocation import WorkflowInvocationResponse
 from galaxy.schema.schema import (
     InvocationsStateCounts,
-    WorkflowIndexQueryPayload,
+    WorkflowIndexPayload,
 )
 from galaxy.schema.workflows import (
     InvokeWorkflowPayload,
@@ -45,10 +45,6 @@
 log = logging.getLogger(__name__)
 
 
-class WorkflowIndexPayload(WorkflowIndexQueryPayload):
-    missing_tools: bool = False
-
-
 class WorkflowsService(ServiceBase):
     def __init__(
         self,