Recommend IWC workflows alongside tools

The tool_recommendation agent now also surfaces IWC workflows when the
ask is analysis-shaped rather than tool-shaped: "which tool sorts a BAM?"
still returns a tool, but "RNA-seq from FASTQ to differential expression"
returns a workflow. Adds search_iwc_workflows / get_iwc_workflow_details
as pydantic-ai tools on the agent (going through the module-level iwc
helpers so they share the cached manifest with the MCP wrappers), extends
SimplifiedToolRecommendationResult with a recommended_workflows field,
and renders a Recommended IWC Workflows section in the formatted output.

Workflow recommendations produce a new WORKFLOW_IMPORT ActionSuggestion
(parameters: trs_id, name) so the UI can wire that to the existing
import_workflow_from_iwc operation -- one click from "this is the
analysis you want" to "imported into your library." When the agent
returns both a tool and a workflow (ambiguous ask), the tool keeps
priority 1 and the workflow drops to priority 2.

Prompt updated to teach the tool-vs-workflow heuristic and the new
agent tools. Five unit tests cover suggestion creation (with/without
trs_id, with/without a competing tool), the rendered workflow section,
and the search helper against a mocked manifest.

Author: dannon

lib/galaxy/agents/prompts/tool_recommendation.md

@@ -1,42 +1,57 @@
-# Galaxy Tool Recommendation Agent
+# Galaxy Analysis Recommendation Agent
 
-You are a Galaxy Project expert specializing in tool discovery and recommendation.
+You are a Galaxy Project expert specializing in **analysis discovery**. Your job is to recommend the _right kind of thing_ for the user's request:
 
-Your goal is to help users find the right tools for their bioinformatics tasks by providing practical recommendations with clear reasoning.
+- A **tool** when the user asks for a single, atomic operation ("which tool sorts a BAM?", "I need to merge FASTQ files").
+- An **IWC workflow** when the user asks for a complete, multi-step analysis ("RNA-seq from FASTQ to differential expression", "variant calling pipeline", "ChIP-seq analysis").
+- **Both** when the user is unsure and could reasonably want either.
+
+Default to a tool for narrow asks. Default to a workflow for end-to-end asks. When in doubt, return both and let the user choose.
 
 ## CRITICAL: Tool Availability
 
 **This Galaxy server only has certain tools installed. You MUST verify tools exist before recommending them.**
 
-1. **ALWAYS call `search_galaxy_tools` FIRST** before making any recommendations
-2. **ONLY recommend tools that appear in the search results** - if a tool doesn't show up in the search, it is NOT installed on this server
-3. If your search returns no results for a common tool (like BWA, HISAT2, etc.), that means it's not installed
+1. **For tool recommendations: ALWAYS call `search_galaxy_tools` FIRST** before naming a tool.
+2. **ONLY recommend tools that appear in the search results** -- if a tool doesn't show up in the search, it is NOT installed on this server.
+3. If your search returns no results for a common tool (like BWA, HISAT2, etc.), that means it's not installed.
 4. When a well-known tool is not installed, tell the user: "While [tool name] would typically be recommended for this task, it doesn't appear to be installed on this Galaxy server. You may want to contact your administrator to request its installation."
 
+IWC workflows are a separate catalog -- they can be recommended even if not yet installed on this server, because the user can import them via `import_workflow_from_iwc`.
+
 ## Available Tools
 
-- **`search_galaxy_tools(query)`** - Search for tools by keyword. Always start here.
-- **`get_galaxy_tool_details(tool_id)`** - Get detailed info (inputs, outputs, version) for a specific tool. Use after searching to provide better recommendations.
-- **`get_galaxy_tool_categories()`** - List available tool categories. Use when user asks "what kinds of tools are available?" or to understand the server's capabilities.
+- **`search_galaxy_tools(query)`** -- Search this server's installed tools by keyword. Always start here for atomic asks.
+- **`get_galaxy_tool_details(tool_id)`** -- Get inputs, outputs, version for a specific tool.
+- **`get_galaxy_tool_categories()`** -- List tool categories on this server.
+- **`search_iwc_workflows(query, limit=5)`** -- Search the IWC catalog for end-to-end workflows. Use for analysis-shaped requests.
+- **`get_iwc_workflow_details(trs_id)`** -- Get full details (steps, tools, readme) for one IWC workflow before recommending it.
 
 ## Recommendation Process
 
-1. Understand the user's task and data types
-2. **Call `search_galaxy_tools` with relevant keywords** (e.g., "alignment", "mapping", "fastq")
-3. Optionally call `get_galaxy_tool_details` on promising candidates to get input/output format info
-4. Recommend tools from the search results, using their exact IDs
-5. If no suitable tools are found, be honest about the limitation
+1. Decide: is the user asking for a single step (tool) or a complete analysis (workflow)?
+2. For tools: call `search_galaxy_tools`, optionally `get_galaxy_tool_details`, populate `primary_tools` from the search results.
+3. For workflows: call `search_iwc_workflows`, optionally `get_iwc_workflow_details` for the top hit, populate `recommended_workflows` with the entries from the search (preserve `trsID`, `name`, `description`, `step_count`, `tools_used`, `match_score`).
+4. If the ask is ambiguous, populate both `primary_tools` and `recommended_workflows`.
+5. Always explain _why_ in the `reasoning` field, including the tool-vs-workflow choice.
+
+## Workflow Recommendations
+
+When recommending a workflow:
+
+- Always preserve the exact `trsID` from `search_iwc_workflows` -- this is what the import action needs.
+- Mention the step count and the key tools the workflow uses, so the user can judge fit.
+- Prefer workflows whose `tools_used` overlap with what's installed on this server, but do not require it.
 
 ## Tool IDs
 
-- Use ONLY the exact `id` field from search results
-- Never guess or fabricate tool IDs based on your training knowledge
-- If you know a tool exists in Galaxy generally but it's not in the search results, it's NOT available on this server
+- Use ONLY the exact `id` field from `search_galaxy_tools` results.
+- Never guess or fabricate tool IDs based on training data.
+- If a tool exists in Galaxy generally but is not in the search results, it's NOT available on this server.
 
 ## Best Practices
 
-- Prioritize tools that are well-maintained and widely used
-- Consider the user's experience level
-- Explain why you're recommending specific tools
-- Mention important parameters or configuration options
-- Suggest workflows when multiple tools are needed
+- Match the scope of the recommendation to the scope of the ask.
+- Explain which kind of recommendation you chose and why.
+- Mention important parameters or configuration options for tools.
+- For workflows, mention what the user gets end-to-end (input format -> outputs).

lib/galaxy/agents/tools.py

@@ -1,7 +1,13 @@
 """
 Tool recommendation agent for suggesting appropriate Galaxy tools.
+
+Despite the historical name, this agent recommends both atomic Galaxy tools
+and end-to-end IWC workflows. Atomic asks ("which tool sorts a BAM?") still
+get a tool back; analysis-shaped asks ("RNA-seq from FASTQ to differential
+expression") get a workflow back.
 """
 
+import asyncio
 import logging
 import re
 from pathlib import Path
@@ -16,6 +22,7 @@
 from pydantic_ai import Agent
 from pydantic_ai.tools import RunContext
 
+from galaxy.agents import iwc
 from galaxy.schema.agents import ConfidenceLevel
 from .base import (
     ActionSuggestion,
@@ -33,11 +40,25 @@
 log = logging.getLogger(__name__)
 
 
+def _iwc_search(query: str, limit: int) -> list[dict[str, Any]]:
+    workflows = iwc.all_workflows(iwc.fetch_manifest())
+    return iwc.search_workflows(workflows, query, limit=limit)
+
+
+def _iwc_details(trs_id: str) -> Optional[dict[str, Any]]:
+    workflows = iwc.all_workflows(iwc.fetch_manifest())
+    for wf in workflows:
+        if wf.get("trsID") == trs_id:
+            return iwc.enrich_workflow(wf, include_full_readme=False)
+    return None
+
+
 class SimplifiedToolRecommendationResult(BaseModel):
     """Tool recommendation result using simple types for local LLM compatibility."""
 
-    primary_tools: list[dict[str, Any]]
+    primary_tools: list[dict[str, Any]] = []
     alternative_tools: list[dict[str, Any]] = []
+    recommended_workflows: list[dict[str, Any]] = []
     workflow_suggestion: Optional[str] = None
     parameter_guidance: dict[str, Any] = {}
     confidence: ConfidenceLiteral
@@ -121,6 +142,49 @@ async def get_galaxy_tool_categories(ctx: RunContext[GalaxyAgentDependencies]) -
                 return "No tool categories found"
             return "Available tool categories:\n" + "\n".join(f"- {cat}" for cat in categories)
 
+        @agent.tool
+        async def search_iwc_workflows(ctx: RunContext[GalaxyAgentDependencies], query: str, limit: int = 5) -> str:
+            """Search the IWC (Intergalactic Workflows Commission) catalog for workflows.
+
+            Use this when the user is asking for a multi-step analysis (e.g. "run
+            an RNA-seq pipeline", "variant calling from FASTQ") rather than a
+            single tool. Returns ranked workflow entries with trsID, name,
+            description, step count, and the tools each workflow uses.
+            """
+            results = await self.search_iwc_workflows(query, limit=limit)
+            if not results:
+                return f"No IWC workflows found matching '{query}'"
+            lines = [f"Found {len(results)} IWC workflows for '{query}':"]
+            for wf in results:
+                lines.append(
+                    f"- trsID: {wf['trsID']}, name: {wf['name']}, steps: {wf['step_count']}, "
+                    f"tools: {', '.join(wf.get('tools_used', [])[:6])}, "
+                    f"description: {(wf.get('description') or '')[:160]}"
+                )
+            return "\n".join(lines)
+
+        @agent.tool
+        async def get_iwc_workflow_details(ctx: RunContext[GalaxyAgentDependencies], trs_id: str) -> str:
+            """Fetch the full enriched IWC entry for a single workflow.
+
+            Use after search_iwc_workflows to get the complete tool list,
+            authors, categories, and readme summary before recommending.
+            """
+            details = await self.get_iwc_workflow_details(trs_id)
+            if details is None:
+                return f"No IWC workflow found with trsID {trs_id}"
+            lines = [
+                f"Name: {details.get('name')}",
+                f"trsID: {details.get('trsID')}",
+                f"Steps: {details.get('step_count')}",
+                f"Tags: {', '.join(details.get('tags', []))}",
+                f"Categories: {', '.join(details.get('categories', []))}",
+                f"Tools used: {', '.join(details.get('tools_used', []))}",
+                f"Description: {details.get('description', '')}",
+                f"Readme: {details.get('readme_summary', '')}",
+            ]
+            return "\n".join(lines)
+
         return agent
 
     def get_system_prompt(self) -> str:
@@ -202,6 +266,22 @@ async def get_tool_details(self, tool_id: str) -> dict[str, Any]:
             log.warning(f"Error getting tool details for {tool_id}: {e}")
             return {"id": tool_id, "error": str(e)}
 
+    async def search_iwc_workflows(self, query: str, limit: int = 5) -> list[dict[str, Any]]:
+        """Search the IWC manifest. Network-bound on cache miss; runs in a thread."""
+        try:
+            return await asyncio.to_thread(_iwc_search, query, limit)
+        except (OSError, ValueError) as e:
+            log.warning(f"IWC search failed for query={query!r}: {e}")
+            return []
+
+    async def get_iwc_workflow_details(self, trs_id: str) -> Optional[dict[str, Any]]:
+        """Fetch one workflow from the IWC manifest, fully enriched."""
+        try:
+            return await asyncio.to_thread(_iwc_details, trs_id)
+        except (OSError, ValueError) as e:
+            log.warning(f"IWC details lookup failed for {trs_id!r}: {e}")
+            return None
+
     async def get_tool_categories(self) -> list[str]:
         if not self.deps.toolbox:
             log.warning("Toolbox not available in agent dependencies")
@@ -300,8 +380,11 @@ async def process(self, query: str, context: Optional[dict[str, Any]] = None) ->
                     suggestions=suggestions,
                     agent_data={
                         "num_tools_found": len(recommendation.primary_tools),
+                        "num_workflows_found": len(recommendation.recommended_workflows),
                         "has_alternatives": bool(recommendation.alternative_tools),
-                        "has_workflow": bool(recommendation.workflow_suggestion),
+                        "has_workflow": bool(
+                            recommendation.recommended_workflows or recommendation.workflow_suggestion
+                        ),
                         "search_keywords": recommendation.search_keywords,
                     },
                     reasoning=recommendation.reasoning,
@@ -357,6 +440,23 @@ def _format_recommendation_response(self, recommendation: SimplifiedToolRecommen
                 tool_name = tool.get("name", tool.get("tool_name", "Unknown"))
                 parts.append(f"- **{tool_name}**: {tool.get('description', 'No description')}")
 
+        if recommendation.recommended_workflows:
+            parts.append("\n**Recommended IWC Workflows:**")
+            for i, wf in enumerate(recommendation.recommended_workflows[:3], 1):
+                wf_name = wf.get("name", "Unknown workflow")
+                trs_id = wf.get("trsID") or wf.get("trs_id") or ""
+                parts.append(f"\n{i}. **{wf_name}**")
+                if trs_id:
+                    parts.append(f"   - trsID: `{trs_id}`")
+                if wf.get("description"):
+                    parts.append(f"   - {wf['description']}")
+                if wf.get("step_count"):
+                    parts.append(f"   - Steps: {wf['step_count']}")
+                if wf.get("tools_used"):
+                    parts.append(f"   - Tools: {', '.join(wf['tools_used'][:6])}")
+                if wf.get("categories"):
+                    parts.append(f"   - Categories: {', '.join(wf['categories'])}")
+
         if recommendation.workflow_suggestion:
             parts.append(f"\n**Workflow Suggestion:**\n{recommendation.workflow_suggestion}")
 
@@ -366,12 +466,13 @@ def _format_recommendation_response(self, recommendation: SimplifiedToolRecommen
                 parts.append(f"- {param}: {value}")
 
         if recommendation.reasoning:
-            parts.append(f"\n**Why these tools?**\n{recommendation.reasoning}")
+            parts.append(f"\n**Why this recommendation?**\n{recommendation.reasoning}")
 
         return "\n".join(parts)
 
     def _create_suggestions(self, recommendation: SimplifiedToolRecommendationResult) -> list[ActionSuggestion]:
         suggestions = []
+        action_confidence = ConfidenceLevel(recommendation.confidence.lower())
 
         if recommendation.primary_tools:
             top_tool = recommendation.primary_tools[0]
@@ -381,7 +482,6 @@ def _create_suggestions(self, recommendation: SimplifiedToolRecommendationResult
             log.debug(f"Extracted tool_name={tool_name}, tool_id={tool_id}")
 
             if tool_id and self._verify_tool_exists(tool_id):
-                action_confidence = ConfidenceLevel(recommendation.confidence.lower())
                 suggestions.append(
                     ActionSuggestion(
                         action_type=ActionType.TOOL_RUN,
@@ -394,6 +494,21 @@ def _create_suggestions(self, recommendation: SimplifiedToolRecommendationResult
             elif tool_id:
                 log.warning(f"Tool '{tool_id}' recommended but not found in toolbox - skipping suggestion")
 
+        if recommendation.recommended_workflows:
+            top_wf = recommendation.recommended_workflows[0]
+            trs_id = top_wf.get("trsID") or top_wf.get("trs_id")
+            wf_name = top_wf.get("name", "IWC workflow")
+            if trs_id:
+                suggestions.append(
+                    ActionSuggestion(
+                        action_type=ActionType.WORKFLOW_IMPORT,
+                        description=f"Import {wf_name} from IWC",
+                        parameters={"trs_id": trs_id, "name": wf_name},
+                        confidence=action_confidence,
+                        priority=1 if not recommendation.primary_tools else 2,
+                    )
+                )
+
         return suggestions
 
     def _verify_tool_exists(self, tool_id: str) -> bool:

lib/galaxy/schema/agents.py

@@ -31,6 +31,7 @@ class ActionType(str, Enum):
     CONTACT_SUPPORT = "contact_support"
     VIEW_EXTERNAL = "view_external"
     DOCUMENTATION = "documentation"
+    WORKFLOW_IMPORT = "workflow_import"
 
 
 class ActionSuggestion(BaseModel):
@@ -54,6 +55,9 @@ def validate_parameters(self) -> "ActionSuggestion":
         elif self.action_type == ActionType.VIEW_EXTERNAL:
             if not self.parameters.get("url"):
                 raise ValueError("VIEW_EXTERNAL requires 'url' parameter")
+        elif self.action_type == ActionType.WORKFLOW_IMPORT:
+            if not self.parameters.get("trs_id"):
+                raise ValueError("WORKFLOW_IMPORT requires 'trs_id' parameter")
         return self