Add run_user_tool agent operation and MCP wrapper

Mirrors run_tool but passes tool_uuid in the payload, which
tools_service._create routes to the toolbox's unprivileged-tool
resolver. Closes UDT parity with the standalone galaxy-mcp server.

lib/galaxy/agents/operations.py

@@ -918,3 +918,12 @@ class AgentOperationsManager:
 
         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]:
+        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/webapps/galaxy/api/mcp.py

@@ -522,6 +522,47 @@ def get_mcp_app(gx_app):
             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
 

test/integration/test_agents.py

@@ -344,6 +344,7 @@ class TestMCPServerSmoke(IntegrationTestCase):
             "list_user_tools",
             "create_user_tool",
             "delete_user_tool",
+            "run_user_tool",
         }
         assert expected.issubset(tool_names), f"Missing tools: {expected - tool_names}"
 
@@ -474,3 +475,38 @@ class TestMCPServerSmoke(IntegrationTestCase):
         uuid, listed = self._run_async(_flow())
         uuids_after = {t["uuid"] for t in listed.data["tools"]}
         assert uuid not in uuids_after
+
+    def test_mcp_run_user_tool(self):
+        """run_user_tool() executes a UDT against an HDA input and produces an output."""
+        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_run_user@test.com")
+
+        populator = DatasetPopulator(self.galaxy_interactor)
+        history_id = populator.new_history()
+        dataset = populator.new_dataset(history_id=history_id, content="abc")
+
+        async def _flow():
+            async with Client(mcp_server) as client:
+                create = await client.call_tool(
+                    "create_user_tool",
+                    {"api_key": api_key, "representation": TOOL_WITH_SHELL_COMMAND},
+                )
+                uuid = create.data["uuid"]
+                return await client.call_tool(
+                    "run_user_tool",
+                    {
+                        "api_key": api_key,
+                        "history_id": history_id,
+                        "tool_uuid": uuid,
+                        "inputs": {"input": {"src": "hda", "id": dataset["id"]}},
+                    },
+                )
+
+        result = self._run_async(_flow())
+        assert not result.is_error, result
+        populator.wait_for_history(history_id, assert_ok=True)
+        output = populator.get_history_dataset_content(history_id)
+        assert output == "abc\n"