td_find_workflow

Instructions

Essential for locating specific workflows when you know the name.
Returns workflow IDs, project info, and latest execution status.

Common scenarios:
- User mentions workflow name, need to find details
- Looking for failing workflows with specific names
- Finding workflows within a specific project
- Getting workflow ID before detailed analysis
- Checking if a named workflow is running/failed

Filters: project_name (optional), status ('success', 'error', 'running').
Use exact_match=True for precise names, False for partial matches.

Implementation Reference

• td_mcp_server/search_tools.py:154-260 (handler)

  The handler function that executes the td_find_workflow tool. It searches for workflows matching the search_term, optionally filtered by project_name, exact_match, and status_filter. Returns matching workflows with details including project info and latest execution status.

  async def td_find_workflow(
      search_term: str,
      project_name: str | None = None,
      exact_match: bool = False,
      status_filter: str | None = None,
  ) -> dict[str, Any]:
      """Find workflows by name to get IDs and check execution status.
  
      Essential for locating specific workflows when you know the name.
      Returns workflow IDs, project info, and latest execution status.
  
      Common scenarios:
      - User mentions workflow name, need to find details
      - Looking for failing workflows with specific names
      - Finding workflows within a specific project
      - Getting workflow ID before detailed analysis
      - Checking if a named workflow is running/failed
  
      Filters: project_name (optional), status ('success', 'error', 'running').
      Use exact_match=True for precise names, False for partial matches.
      """
      if not search_term or not search_term.strip():
          return _format_error_response("Search term cannot be empty")
  
      client = _create_client(include_workflow=True)
      if isinstance(client, dict):
          return client
  
      try:
          # Get workflows (up to 1000)
          workflows = client.get_workflows(count=1000, all_results=True)
  
          found_workflows = []
          search_lower = search_term.lower()
          project_lower = project_name.lower() if project_name else None
  
          for workflow in workflows:
              workflow_name = workflow.name.lower()
              workflow_project = workflow.project.name.lower()
  
              # Check workflow name match
              name_match = False
              if exact_match:
                  name_match = workflow_name == search_lower
              else:
                  name_match = search_lower in workflow_name
  
              if not name_match:
                  continue
  
              # Check project filter if specified
              if project_lower:
                  if project_lower not in workflow_project:
                      continue
  
              # Check status filter if specified
              if status_filter:
                  if workflow.latest_sessions:
                      last_status = workflow.latest_sessions[0].last_attempt.status
                      if last_status != status_filter:
                          continue
                  else:
                      # No sessions means no status, skip if filtering by status
                      continue
  
              # Prepare workflow info
              workflow_info = {
                  "id": workflow.id,
                  "name": workflow.name,
                  "project": {
                      "id": workflow.project.id,
                      "name": workflow.project.name,
                  },
                  "timezone": workflow.timezone,
                  "scheduled": workflow.schedule is not None,
              }
  
              # Add latest status if available
              if workflow.latest_sessions:
                  latest = workflow.latest_sessions[0]
                  workflow_info["latest_session"] = {
                      "session_time": latest.session_time,
                      "status": latest.last_attempt.status,
                      "success": latest.last_attempt.success,
                  }
              else:
                  workflow_info["latest_session"] = None
  
              found_workflows.append(workflow_info)
  
          if found_workflows:
              return {
                  "found": True,
                  "count": len(found_workflows),
                  "workflows": found_workflows,
              }
  
          message = f"No workflows found matching '{search_term}'"
          if project_name:
              message += f" in project '{project_name}'"
          if status_filter:
              message += f" with status '{status_filter}'"
  
          return {"found": False, "count": 0, "message": message}
  
      except Exception as e:
          return _format_error_response(f"Failed to search workflows: {str(e)}")

• td_mcp_server/search_tools.py:27-31 (registration)

  The registration of the td_find_workflow tool using the mcp.tool() decorator within the register_mcp_tools function.

  mcp.tool()(td_find_project)
  mcp.tool()(td_find_workflow)
  mcp.tool()(td_get_project_by_name)
  mcp.tool()(td_smart_search)