Update tools/YouTubeTool.py

tools/YouTubeTool.py

@@ -1,30 +1,74 @@
+import os
 from smolagents import Tool
-from youtube_transcript_api import YouTubeTranscriptApi, NoTranscriptFound, TranscriptsDisabled
-import re
+from googleapiclient.discovery import build
 
-class YouTubeTool(Tool):
-    name = "youtube_reader"
-    description= "Fetches the transcript for the youtube video url"
-    inputs= {"video_url": {"type": "string", "description": "URL or ID of the Youtube video"}}
+class YouTubeSearchTool(Tool):
+    """
+    A tool to search for YouTube videos using the Google Custom Search Engine (CSE) API,
+    by specifically restricting results to the youtube.com domain.
+    """
+    name = "youtube_search"
+    description = "Searches for and returns links and snippets for YouTube videos related to a specific query."
+
+    inputs = {
+        # Corrected the typo/broken line to read correctly
+        "query": {"type": "string", "description": "The search term to find relevant YouTube videos for."}
+    }
     output_type = "string"
 
-    def forward(self, video_url: str) -> str:
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+        # Retrieve credentials from environment variables
+        self.api_key = os.getenv("GOOGLE_API_KEY")
+        self.cse_id = os.getenv("GOOGLE_CSE_ID")
+
+        if not self.api_key or not self.cse_id:
+            # Note: This check relies on environment variables being set up
+            raise ValueError("GOOGLE_API_KEY or GOOGLE_CSE_ID secret not found.")
+
+        # Initialize the Google Custom Search service
+        self.service = build(
+            "customsearch", "v1", developerKey=self.api_key
+        )
+
+    def forward(self, query: str) -> str:
+        """
+        Executes a Google search query, restricting results to youtube.com.
+
+        Args:
+            query: The search term provided by the agent.
+
+        Returns:
+            A formatted string of search results (max 3), or an error message.
+        """
+        print(f"Executing YouTube search for: '{query}'")
+        
+        # Modify the query to ensure results are restricted to YouTube
+        youtube_query = f"{query} site:youtube.com"
+
         try:
-            # Extract video ID
-            match = re.search(r"v=([A-Za-z0-9_-]{11})", video_url)
-            if not match:
-                return "Invalid YouTube URL."
-            video_id = match.group(1)
-
-            # Fetch transcript
-            transcript_list = YouTubeTranscriptApi.get_transcript(video_id)
-            transcript_text = " ".join([t["text"] for t in transcript_list])
-
-            return transcript_text
-
-        except NoTranscriptFound:
-            return "Transcript not available for this video."
-        except TranscriptsDisabled:
-            return "Transcripts are disabled for this video."
+            # Execute the search request for up to 3 results
+            res = self.service.cse().list(
+                q=youtube_query,
+                cx=self.cse_id,
+                num=3
+            ).execute()
+
+            items = res.get('items', [])
+
+            if not items:
+                return "XX record info: No YouTube results found."
+
+            search_results = []
+            for i, item in enumerate(items):
+                search_results.append(
+                    f"RESULT {i+1}: '{item.get('title')}'\n"
+                    f"CONTENT: {item.get('snippet')}\n"
+                    f"SOURCE: {item.get('link')}"
+                )
+
+            return "\n\n---SEPARATOR---\n\n".join(search_results)
+
         except Exception as e:
-            return f"Error fetching transcript: {str(e)}"
+            return f"Error during YouTube Search API call: {e}"