Lesson 17: Initial Data Ingestion and Tooling

1. Setup:

    1.1. Explain to the user the numbered steps of the workflow. Be concise. Keep them numbered so that the user
    can easily refer to them later.
    
    1.2. Ask the user for the research directory, if not provided. Ask the user if any modification is needed for the
    workflow (e.g. running from a specific step, or adding user feedback to specific steps).

    1.3 Extract the URLs from the ARTICLE_GUIDELINE_FILE with the "extract_guidelines_urls" tool. This tool reads the
    ARTICLE_GUIDELINE_FILE and extracts three groups of references from the guidelines:
    • "github_urls" - all GitHub links;
    • "youtube_videos_urls" - all YouTube video links;
    • "other_urls" - all remaining HTTP/HTTPS links;
    • "local_files" - relative paths to local files mentioned in the guidelines (e.g. "code.py", "src/main.py").
    Only extensions allowed are: ".py", ".ipynb", and ".md".
    The extracted data is saved to the GUIDELINES_FILENAMES_FILE within the NOVA_FOLDER directory.

2. Process the extracted resources in parallel:

    You can run the following sub-steps (2.1 to 2.4) in parallel. In a single turn, you can call all the
    necessary tools for these steps.

    2.1 Local files - run the "process_local_files" tool to read every file path listed under "local_files" in the
    GUIDELINES_FILENAMES_FILE and copy its content into the LOCAL_FILES_FROM_RESEARCH_FOLDER subfolder within
    NOVA_FOLDER, giving each copy an appropriate filename (path separators are replaced with underscores).

    2.2 Other URL links - run the "scrape_and_clean_other_urls" tool to read the `other_urls` list from the
    GUIDELINES_FILENAMES_FILE and scrape/clean them. The tool writes the cleaned markdown files inside the
    URLS_FROM_GUIDELINES_FOLDER subfolder within NOVA_FOLDER.

    2.3 GitHub URLs - run the "process_github_urls" tool to process the `github_urls` list from the
    GUIDELINES_FILENAMES_FILE with gitingest and save a Markdown summary for each URL inside the
    URLS_FROM_GUIDELINES_CODE_FOLDER subfolder within NOVA_FOLDER.

    2.4 YouTube URLs - run the "transcribe_youtube_urls" tool to process the `youtube_videos_urls` list from the
    GUIDELINES_FILENAMES_FILE, transcribe each video, and save the transcript as a Markdown file inside the
    URLS_FROM_GUIDELINES_YOUTUBE_FOLDER subfolder within NOVA_FOLDER.
        Note: Please be aware that video transcription can be a time-consuming process. For reference,
        transcribing a 39-minute video can take approximately 4.5 minutes.

def register_mcp_tools(mcp: FastMCP) -> None:
    """Register all MCP tools with the server instance."""
 
		# Step 1: Extract URLs and file references from guidelines
		@mcp.tool()
    async def extract_guidelines_urls(research_directory: str) -> Dict[str, Any]:
        """
        Extract URLs and local file references from article guidelines.
 
        Reads the ARTICLE_GUIDELINE_FILE file in the research directory and extracts:
        - GitHub URLs
        - Other HTTP/HTTPS URLs
        - Local file references (files mentioned in quotes with extensions)
 
        Results are saved to GUIDELINES_FILENAMES_FILE in the research directory.
        """
        result = extract_guidelines_urls_tool(research_directory)
        return result
 
		# Step 2.1: Process local files
		@mcp.tool()
    async def process_local_files(research_directory: str) -> Dict[str, Any]:
        """Process local files referenced in the article guidelines."""
        result = process_local_files_tool(research_directory)
        return result
 
		# Step 2.2: Scrape web URLs
		@mcp.tool()
    async def scrape_and_clean_other_urls(research_directory: str, concurrency_limit: int = 4) -> Dict[str, Any]:
        """Scrape and clean other URLs from GUIDELINES_FILENAMES_FILE."""
        result = await scrape_and_clean_other_urls_tool(research_directory, concurrency_limit)
        return result
 
		# Step 2.3: Process GitHub repositories
		@mcp.tool()
    async def process_github_urls(research_directory: str) -> Dict[str, Any]:
        """
        Process GitHub URLs from GUIDELINES_FILENAMES_FILE using gitingest.
 
        Reads the GUIDELINES_FILENAMES_FILE file and processes each URL listed
        under 'github_urls' using gitingest to extract repository summaries, file trees,
        and content. The results are saved as markdown files in the
        URLS_FROM_GUIDELINES_CODE_FOLDER subfolder.
        """
        result = await process_github_urls_tool(research_directory)
        return result
 
		# Step 2.4: Transcribe YouTube videos
		@mcp.tool()
    async def transcribe_youtube_urls(research_directory: str) -> Dict[str, Any]:
        """
        Transcribe YouTube video URLs from GUIDELINES_FILENAMES_FILE using Gemini 2.5 Pro.
 
        Reads the GUIDELINES_FILENAMES_FILE file and processes each URL listed
        under 'youtube_videos_urls'. Each video is transcribed, and the results are
        saved as markdown files in the URLS_FROM_GUIDELINES_YOUTUBE_FOLDER subfolder.
        """
        result = await transcribe_youtube_videos_tool(research_directory)
        return result

def extract_guidelines_urls_tool(research_folder: str) -> Dict[str, Any]:
    """
    Extract URLs and local file references from the article guidelines in the research folder.
    
    Reads the ARTICLE_GUIDELINE_FILE file and extracts:
    - GitHub URLs
    - YouTube video URLs  
    - Other HTTP/HTTPS URLs
    - Local file references
    
    Results are saved to GUIDELINES_FILENAMES_FILE in the research folder.
    """
    ...
 
    # Convert to Path object
    research_path = Path(research_folder)
    nova_path = research_path / NOVA_FOLDER
    guidelines_path = research_path / ARTICLE_GUIDELINE_FILE
    ...
    
    # Read guidelines content
    guidelines_content = read_file_safe(guidelines_path)
    
    # Extract URLs
    all_urls = extract_urls(guidelines_content)
    github_source_urls = [u for u in all_urls if "github.com" in u]
    youtube_source_urls = [u for u in all_urls if "youtube.com" in u]
    web_source_urls = [u for u in all_urls if "github.com" not in u and "youtube.com" not in u]
 
    # Extract local file paths
    local_file_paths = extract_local_paths(guidelines_content)
    
    # Prepare the extracted data structure
    extracted_data = {
        "github_urls": github_source_urls,
        "youtube_videos_urls": youtube_source_urls, 
        "other_urls": web_source_urls,
        "local_file_paths": local_file_paths,
    }
    
    # Save to JSON file
    output_path = nova_path / GUIDELINES_FILENAMES_FILE
    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(extracted_data, f, indent=2, ensure_ascii=False)
    
    return {
        "status": "success",
        "github_sources_count": len(github_source_urls),
        "youtube_sources_count": len(youtube_source_urls),
        "web_sources_count": len(web_source_urls),
        "local_files_count": len(local_file_paths),
        "output_path": str(output_path.resolve()),
        "message": f"Successfully extracted URLs from article guidelines in '{research_folder}'. "
                  f"Found {len(github_source_urls)} GitHub URLs, {len(youtube_source_urls)} YouTube videos URLs, "
                  f"{len(web_source_urls)} other URLs, and {len(local_file_paths)} local file references. "
                  f"Results saved to: {output_path.resolve()}"
    }

def extract_urls(text: str) -> list[str]:
    """Extract all HTTP/HTTPS URLs from the given text."""
    url_pattern = re.compile(r"https?://[^\s)>\"',]+")
    return url_pattern.findall(text)

from research_agent_part_2.mcp_server.src.tools import extract_guidelines_urls_tool
 
# Update this path to your actual sample research folder
research_folder = "/path/to/research_folder"
result = extract_guidelines_urls_tool(research_folder=research_folder)
print(result)

{'status': 'success', 'github_sources_count': 1, 'youtube_sources_count': 1, 'web_sources_count': 2, 'local_files_count': 0, 'output_path': '/path/to/research_folder/.nova/guidelines_filenames.json', 'message': "Successfully extracted URLs from article guidelines in '/path/to/research_folder'. Found 1 GitHub URLs, 1 YouTube videos URLs, 2 other URLs, and 0 local file references. Results saved to: /path/to/research_folder/.nova/guidelines_filenames.json"}

{
  "github_urls": ["https://github.com/..."],
  "youtube_videos_urls": ["https://youtube.com/..."],
  "other_urls": ["https://some-other-url.com/..."],
  "local_file_paths": ["path/to/local/file.py"]
}

def process_local_files_tool(research_directory: str) -> Dict[str, Any]:
    """
    Process local files referenced in the article guidelines.
 
    Reads the guidelines JSON file and copies each referenced local file
    to the local files subfolder. Path separators in filenames are
    replaced with underscores to avoid creating nested folders.
 
    Args:
        research_directory: Path to the research directory containing the guidelines JSON file
 
    Returns:
        Dict with status, processing results, and file paths
    """
    # ... (path setup and file reading logic) ...
 
    # Load JSON metadata
    data = json.loads(metadata_path.read_text(encoding="utf-8"))
    local_files = data.get("local_file_paths", [])
 
    if not local_files:
        # ... (return success if no files to process) ...
 
    # Create destination folder
    dest_folder = nova_path / LOCAL_FILES_FROM_RESEARCH_FOLDER
    dest_folder.mkdir(parents=True, exist_ok=True)
 
    # ... (initialize counters and lists) ...
 
    # Initialize notebook converter
    notebook_converter = NotebookToMarkdownConverter(include_outputs=True, include_metadata=False)
 
    for rel_path in local_files:
        src_path = research_path / rel_path
        dest_name = rel_path.replace("/", "_").replace("\\", "_")
 
        try:
            if src_path.suffix.lower() == ".ipynb":
                dest_name = dest_name.rsplit(".ipynb", 1)[0] + ".md"
                dest_path = dest_folder / dest_name
                markdown_content = notebook_converter.convert_notebook_to_string(src_path)
                dest_path.write_text(markdown_content, encoding="utf-8")
            else:
                dest_path = dest_folder / dest_name
                shutil.copy2(src_path, dest_path)
 
            processed += 1
            processed_files.append(dest_name)
        except Exception as e:
            errors.append(f"Failed to process {rel_path}: {str(e)}")
 
    # Build result message using the dedicated function
    result_message = build_result_message(research_directory, processed, local_files, dest_folder, warnings, errors)
 
    return {
        "status": "success" if processed > 0 else "warning",
        "files_processed": processed,
        "files_total": len(local_files),
        "processed_files": processed_files,
        "warnings": warnings,
        "errors": errors,
        "output_directory": str(dest_folder.resolve()),
        "message": result_message,
    }

async def scrape_and_clean_other_urls_tool(research_directory: str, concurrency_limit: int = 4) -> Dict[str, Any]:
    """
    Scrape and clean other URLs from guidelines file in the research folder.
    
    Reads the guidelines file and scrapes/cleans each URL listed
    under 'other_urls'. The cleaned markdown content is saved to the
    URLS_FROM_GUIDELINES_FOLDER subfolder with appropriate filenames.
    """    
    # Convert to Path object
    research_path = Path(research_directory)
    nova_path = research_path / NOVA_FOLDER
    
    # Look for GUIDELINES_FILENAMES_FILE file
    guidelines_file_path = nova_path / GUIDELINES_FILENAMES_FILE
    
    # Read the guidelines filenames file
    guidelines_data = json.loads(read_file_safe(guidelines_file_path))
    urls_to_scrape = guidelines_data.get("other_urls", [])
    
    if not urls_to_scrape:
        return {
            "status": "success",
            "urls_processed": [],
            "urls_failed": [],
            "total_urls": 0,
            "successful_urls_count": 0,
            "failed_urls_count": 0,
            "output_directory": str(nova_path / URLS_FROM_GUIDELINES_FOLDER),
            "message": "No other URLs found to scrape in the guidelines filenames file."
        }
    
    # Read article guidelines for context
    guidelines_path = research_path / ARTICLE_GUIDELINE_FILE
    guidelines_content = read_file_safe(guidelines_path)
    
    # Scrape URLs concurrently
    completed_results = await scrape_urls_concurrently(
        urls_to_scrape, 
        concurrency_limit, 
        guidelines_content
    )
    
    # Write results to files
    output_dir = nova_path / URLS_FROM_GUIDELINES_FOLDER
    saved_files, successful_scrapes = write_scraped_results_to_files(completed_results, output_dir)
    
    # Calculate statistics
    failed_urls = [res["url"] for res in completed_results if not res.get("success", False)]
    successful_urls = [res["url"] for res in completed_results if res.get("success", False)]
    
    return {
        "status": "success",
        "urls_processed": successful_urls,
        "urls_failed": failed_urls,
        "total_urls": len(urls_to_scrape),
        "successful_urls_count": successful_scrapes,
        "failed_urls_count": len(failed_urls),
        "output_directory": str(output_dir),
        "message": f"Successfully processed {successful_scrapes}/{len(urls_to_scrape)} URLs. "
                  f"Results saved to: {output_dir}"
    }
   

async def scrape_url(url: str, firecrawl_app: AsyncFirecrawl) -> dict:
    """
    Scrape a URL using Firecrawl with retries and return a dict with url, title, markdown.
 
    Uses maxAge=1 week for 500% faster scraping by leveraging cached data when available.
    This optimization significantly improves performance for documentation, articles, and
    relatively static content while maintaining freshness within acceptable limits.
    """
    max_retries = 3
    base_delay = 5  # seconds
    timeout_seconds = 120  # 2 minutes timeout per request
 
    for attempt in range(max_retries):
        try:
            # Add timeout to individual Firecrawl request
            # Use maxAge=1 week for 500% faster scraping with cached data
            res = await firecrawl_app.scrape(
                url, formats=["markdown"], maxAge=MAX_AGE_ONE_WEEK, timeout=timeout_seconds
            )
            title = res.metadata.title if res and res.metadata and res.metadata.title else "N/A"
            markdown_content = res.markdown if res and res.markdown else ""
            return {"url": url, "title": title, "markdown": markdown_content, "success": True}
        except asyncio.TimeoutError:
            # Manage retries
            ...
        except Exception as e:
            # Manage retries
            ...
    
    return {
        "url": url,
        "title": "Scraping Failed",
        "markdown": f"⚠️ Error scraping {url} after {max_retries} attempts.",
        "success": False,
    }

Your task is to clean markdown content scraped from a webpage by *only removing* all irrelevant sections such as
headers, footers, navigation bars, advertisements, sidebars, self-promotion, call-to-actions, etc.
Focus on keeping only the core textual content (and code content if there are code sections) that is pertinent to
the article guidelines provided below.
Return *only* the cleaned markdown.
Do not summarize or rewrite the original content. This task is only about *removing* irrelevant content.
Good content should be kept as is, do not touch it.

Here are the article guidelines:

{article_guidelines}


Here is the markdown content to clean:

{markdown_content}

async def clean_markdown(
    markdown_content: str, article_guidelines: str, url_for_log: str, chat_model: BaseChatModel
) -> str:
    """Clean markdown content via LLM and convert image syntax to URLs."""
    if not markdown_content.strip():
        return markdown_content
 
    prompt_text = PROMPT_CLEAN_MARKDOWN.format(article_guidelines=article_guidelines, markdown_content=markdown_content)
    timeout_seconds = 180  # 3 minutes timeout for LLM call
 
    try:
        # Add timeout to LLM API call
        response = await asyncio.wait_for(chat_model.ainvoke(prompt_text), timeout=timeout_seconds)
        cleaned_content = response.content if hasattr(response, "content") else str(response)
 
        if isinstance(cleaned_content, list):
            cleaned_content = "".join(str(part) for part in cleaned_content)
 
        # Post-process: convert markdown images to just URLs
        cleaned_content = convert_markdown_images_to_urls(cleaned_content)
 
        return cleaned_content
    except asyncio.TimeoutError:
        logger.error(f"LLM API call timed out after {timeout_seconds}s for {url_for_log}. Using original content.")
        return markdown_content
    except Exception as e:
        logger.error(f"Error cleaning markdown for {url_for_log}: {e}. Using original content.", exc_info=True)
        return markdown_content

from research_agent_part_2.mcp_server.src.tools import process_github_urls_tool
 
# Test GitHub URL processing
result = await process_github_urls_tool(research_directory=research_folder)
print(result)

{'status': 'success', 'urls_processed': 1, 'urls_total': 1, 'files_saved': 1, 'output_directory': '/.../.nova/urls_from_guidelines_code', 'message': "Processed 1/1 GitHub URLs from guidelines_filenames.json in '/Users/fabio/Desktop/course-ai-agents/lessons/research_agent_part_2/data/sample_research_folder'. Saved markdown summaries to urls_from_guidelines_code folder."}

async def transcribe_youtube(
    url: str,
    output_path: Path,
    timestamp: int = 30,
) -> None:
    """
    Transcribes a public YouTube video using a Gemini model and saves the
    result to a file.
 
    Args:
        url: The public URL of the YouTube video.
        output_path: The path to save the transcription markdown file.
        timestamp: The interval in seconds for inserting timestamps in the
                   transcription.
    """
# Create client internally using settings and track with Opik if configured
    base_client = genai.Client(api_key=settings.google_api_key.get_secret_value())
    client = track_genai_client(base_client)
    model_name = settings.youtube_transcription_model
 
    prompt = PROMPT_YOUTUBE_TRANSCRIPTION.format(timestamp=timestamp)
 
    parts: list[types.Part] = [
        types.Part(
            file_data=types.FileData(file_uri=url)# YouTube URL - no download needed
        ),
        types.Part(text=prompt),
    ]
 
    ...
    response: types.GenerateContentResponse = await client.aio.models.generate_content(
        model=model_name,
        contents=types.Content(parts=parts),
    )
    ...
 
    output_path.write_text(response.text, encoding="utf-8")

{
      "github_urls": ["https://github.com/example/repo"],
      "youtube_videos_urls": ["https://www.youtube.com/watch?v=..."],
      "other_urls": ["https://blog.example.com/article1", "https://docs.example.com/page2"],
      "local_file_paths": ["notebooks/analysis.ipynb"]
    }

{
      "status": "success",
      "successful_urls_count": 1,
      "failed_urls_count": 1,
      "urls_failed": ["https://docs.example.com/page2"],
      "message": "Successfully processed 1/2 URLs..."
    }

from research_agent_part_2.mcp_client.src.client import main as client_main
import sys
 
async def run_client():
    _argv_backup = sys.argv[:]
    sys.argv = ["client"]
    try:
        await client_main()
    finally:
        sys.argv = _argv_backup
 
# Start client with in-memory server
await run_client()

🤔 LLM's Thoughts: 
**Executing the Next Phase of the Workflow**

Okay, so I'm ready to move forward. My focus right now is definitely on steps 1 and 2 of this workflow. I remember step 1.3 is all about pulling those guidelines URLs, which is straightforward enough. But step 2, that's where the real action is. It's calling for some serious parallel processing! I've got four distinct sub-steps within step 2 that need to run concurrently: `process_local_files`, `scrape_and_clean_other_urls`, `process_github_urls`, and `transcribe_youtube_urls`.  

All of these tools, the core tools for these steps and sub-steps, need to be executed while taking into account the `research_directory`.  I'll manage the inputs and outputs, and make sure everything is coordinated.  Time to get these functions moving!

🔧 Function Call (Tool): 
  Tool:  extract_guidelines_urls 
...