Building Autonomous Research Agents with Firecrawl & Brave Search

┌────────────────────────────────────────────────────────────────────────┐
│                         USER RESEARCH OBJECTIVE                        │
│            "Analyze enterprise pricing & latency SLAs of Vector DBs"   │
└───────────────────────────────────┬────────────────────────────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│                     TIER 1: DISCOVERY & TRIAGE                         │
│  • Query expansion & entity decomposition                              │
│  • Brave Search API (Independent web index, no Google/Bing bias)       │
│  • Domain authority filtering & URL deduplication                      │
└───────────────────────────────────┬────────────────────────────────────┘
                                    │ Ranked URL Candidates
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│                   TIER 2: RESILIENT EXTRACTION (MCP)                   │
│  • Firecrawl MCP Engine (`scrape_url`, `crawl_site`, `map_site`)       │
│  • Headless browser cluster + residential proxy rotation               │
│  • Dynamic JS execution + Shadow DOM / iframe traversal                │
│  • HTML-to-Clean-Markdown transformation (Noise stripped)             │
│  • Token pruning & strict citation regex pipelines                     │
└───────────────────────────────────┬────────────────────────────────────┘
                                    │ Clean Markdown + Metadata
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│                 TIER 3: RECURSIVE SYNTHESIS & VERIFICATION             │
│  • Chunking & semantic embedding into Supabase / Pinecone              │
│  • Recursive claim extraction & cross-source corroboration            │
│  • Structured JSON synthesis + Markdown report with verifiable URLs    │
└───────────────────────────────────┬────────────────────────────────────┘
                                    │
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│              AUTOMATED OUTPUTS & WEBHOOK NOTIFICATIONS                 │
│         Slack / Discord Webhooks • S3 Markdown Artifacts • DB Sync      │
└────────────────────────────────────────────────────────────────────────┘

RAW WEB PAGE (250KB HTML / ~65,000 Tokens)
├── Scripts, Analytics, Pixel Trackers  ──► [ STRIPPED BY FIRECRAWL ]
├── SVG Icons, CSS Class Names          ──► [ STRIPPED BY FIRECRAWL ]
├── Navigation Menus & Cookie Banners   ──► [ STRIPPED BY FIRECRAWL ]
└── Core Semantic Article & Data Tables ──► CLEAN MDX (3.2KB / ~850 Tokens)
                                            (98.7% Token Reduction)

pip install httpx mcp pydantic openai supabase python-dotenv rich

BRAVE_SEARCH_API_KEY=BSAx...
FIRECRAWL_API_KEY=fc-...
OPENAI_API_KEY=sk-...
SUPABASE_URL=https://xyzcompany.supabase.co
SUPABASE_SERVICE_ROLE_KEY=eyJhbGciOi...
DISCORD_WEBHOOK_URL=https://discord.com/api/webhooks/...

"""
Autonomous Research Agent with Firecrawl MCP & Brave Search API
Author: Prompt Genius Architecture Team
Date: 2026-08-19
"""

import asyncio
import os
import re
import json
import logging
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field
import httpx
from openai import AsyncOpenAI
from supabase import create_client, Client

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("autonomous_researcher")

# ---------------------------------------------------------------------------
# Data Models
# ---------------------------------------------------------------------------

class SearchResultItem(BaseModel):
    title: str
    url: str
    description: str
    score: float = 0.0

class CrawledDocument(BaseModel):
    url: str
    title: Optional[str] = None
    markdown: str
    token_count: int
    metadata: Dict[str, Any] = Field(default_factory=dict)

class ResearchClaim(BaseModel):
    claim: str
    source_url: str
    confidence: float
    corroborated: bool = False

class ResearchDossier(BaseModel):
    topic: str
    executive_summary: str
    key_findings: List[str]
    claims: List[ResearchClaim]
    sources_consulted: List[str]
    generated_at: str

# ---------------------------------------------------------------------------
# Brave Search Discovery Engine
# ---------------------------------------------------------------------------

class BraveSearchEngine:
    """Handles independent web discovery using the Brave Search API."""
    
    BASE_URL = "https://api.search.brave.com/res/v1/web/search"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.getenv("BRAVE_SEARCH_API_KEY")
        if not self.api_key:
            raise ValueError("BRAVE_SEARCH_API_KEY is required.")
        self.client = httpx.AsyncClient(
            headers={
                "Accept": "application/json",
                "Accept-Encoding": "gzip",
                "X-Subscription-Token": self.api_key
            },
            timeout=10.0
        )

    async def search(self, query: str, count: int = 5) -> List[SearchResultItem]:
        logger.info(f"Executing Brave search for query: '{query}' (limit: {count})")
        params = {
            "q": query,
            "count": count,
            "text_decorations": False,
            "search_lang": "en"
        }
        try:
            response = await self.client.get(self.BASE_URL, params=params)
            response.raise_for_status()
            data = response.json()
            results = []
            
            web_results = data.get("web", {}).get("results", [])
            for item in web_results:
                results.append(SearchResultItem(
                    title=item.get("title", ""),
                    url=item.get("url", ""),
                    description=item.get("description", "")
                ))
            return results
        except Exception as e:
            logger.error(f"Brave Search API error: {e}")
            return []

    async def close(self):
        await self.client.aclose()

# ---------------------------------------------------------------------------
# Firecrawl MCP Client
# ---------------------------------------------------------------------------

class FirecrawlExtractor:
    """Connects to Firecrawl API / MCP server for headless JS scraping and clean MD extraction."""

    BASE_URL = "https://api.firecrawl.dev/v1"

    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key or os.getenv("FIRECRAWL_API_KEY")
        if not self.api_key:
            raise ValueError("FIRECRAWL_API_KEY is required.")
        self.client = httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {self.api_key}",
                "Content-Type": "application/json"
            },
            timeout=45.0
        )

    @staticmethod
    def _prune_markdown_tokens(raw_markdown: str, max_chars: int = 18000) -> str:
        """
        Token Optimization Pipeline:
        1. Strips image URLs and redundant base64 data strings.
        2. Normalizes excessive line breaks.
        3. Caps characters to stay within tight LLM context budgets.
        """
        # Remove base64 images and large media links
        cleaned = re.sub(r"![.*?](data:image/.*?)", "", raw_markdown)
        cleaned = re.sub(r"![.*?](https?://.*?)", "", cleaned)
        # Collapse multiple empty lines
        cleaned = re.sub(r"
{3,}", "

", cleaned)
        
        if len(cleaned) > max_chars:
            cleaned = cleaned[:max_chars] + "

... [Content truncated for context limits] ..."
            
        return cleaned.strip()

    async def scrape_url(self, url: str) -> Optional[CrawledDocument]:
        logger.info(f"Extracting clean markdown via Firecrawl: {url}")
        payload = {
            "url": url,
            "pageOptions": {
                "onlyMainContent": True,
                "includeHtml": False
            }
        }
        try:
            response = await self.client.post(f"{self.BASE_URL}/scrape", json=payload)
            if response.status_code != 200:
                logger.warning(f"Firecrawl scrape failed for {url} with status {response.status_code}: {response.text}")
                return None
            
            data = response.json()
            if not data.get("success", False):
                return None
                
            page_data = data.get("data", {})
            raw_md = page_data.get("markdown", "")
            pruned_md = self._prune_markdown_tokens(raw_md)
            
            # Approximate token count (1 token ≈ 4 chars)
            est_tokens = len(pruned_md) // 4

            return CrawledDocument(
                url=url,
                title=page_data.get("metadata", {}).get("title"),
                markdown=pruned_md,
                token_count=est_tokens,
                metadata=page_data.get("metadata", {})
            )
        except Exception as e:
            logger.error(f"Error scraping {url} with Firecrawl: {e}")
            return None

    async def close(self):
        await self.client.aclose()

# ---------------------------------------------------------------------------
# Tool Gating & Triage Logic
# ---------------------------------------------------------------------------

class ResearchOrchestrator:
    """Coordinates search query generation, tool gating, synthesis, and persistence."""

    def __init__(self):
        self.brave = BraveSearchEngine()
        self.firecrawl = FirecrawlExtractor()
        self.openai = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        
        # Initialize Supabase Client
        supabase_url = os.getenv("SUPABASE_URL")
        supabase_key = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
        self.supabase: Optional[Client] = (
            create_client(supabase_url, supabase_key)
            if supabase_url and supabase_key
            else None
        )

    async def generate_search_queries(self, objective: str) -> List[str]:
        """Decomposes an abstract research objective into 3 distinct atomic search queries."""
        prompt = f"""You are an autonomous research planner.
Objective: "{objective}"

Decompose this objective into exactly 3 diverse, high-signal search queries optimized for a search engine.
Avoid Boolean syntax (AND/OR). Return ONLY a JSON list of strings.

Example:
["vector database benchmarks latency 2026", "qdrant pinecone milvus pricing comparison", "weaviate enterprise sla downtime analysis"]
"""
        response = await self.openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )
        content = response.choices[0].message.content or "{}"
        parsed = json.loads(content)
        queries = parsed.get("queries", list(parsed.values())[0] if parsed else [])
        if not isinstance(queries, list):
            queries = [objective]
        return queries[:3]

    async def triage_and_extract(self, search_results: List[SearchResultItem], max_documents: int = 4) -> List[CrawledDocument]:
        """
        Tool Gating Logic:
        Evaluates search snippets and filters out low-signal domains (e.g., social feeds, login walls).
        Executes parallel Firecrawl extractions.
        """
        # Deduplicate URLs
        unique_urls = list({item.url: item for item in search_results}.values())
        
        # Filter obvious non-content URLs
        filtered_candidates = [
            item for item in unique_urls
            if not any(blocked in item.url.lower() for blocked in ["twitter.com", "facebook.com", "login", "signup", ".pdf"])
        ]
        
        target_items = filtered_candidates[:max_documents]
        logger.info(f"Targeting {len(target_items)} verified URLs for Firecrawl extraction.")
        
        # Parallel extraction with asyncio.gather
        tasks = [self.firecrawl.scrape_url(item.url) for item in target_items]
        scraped_docs = await asyncio.gather(*tasks)
        
        return [doc for doc in scraped_docs if doc is not None]

    async def synthesize_dossier(self, objective: str, documents: List[CrawledDocument]) -> ResearchDossier:
        """Performs recursive multi-document synthesis with strict citation tracking."""
        context_blocks = []
        for idx, doc in enumerate(documents, start=1):
            context_blocks.append(
                f"--- SOURCE [{idx}] ---
"
                f"URL: {doc.url}
"
                f"TITLE: {doc.title or 'Unknown'}
"
                f"CONTENT:
{doc.markdown}
"
            )
        
        aggregated_context = "

".join(context_blocks)
        
        system_prompt = (
            "You are a Principal Intelligence Analyst. Your mission is to produce a verifiable, "
            "fact-checked Research Dossier based exclusively on the provided source materials.
"
            "Rules:
"
            "1. Every single key claim MUST be attributed to a specific source URL from the context.
"
            "2. Never hallucinate facts or extrapolate beyond provided evidence.
"
            "3. Structure output strictly according to the specified JSON schema."
        )
        
        user_prompt = f"""RESEARCH OBJECTIVE:
{objective}

AVAILABLE PRIMARY SOURCES:
{aggregated_context}

Respond in the following JSON format:
{{
  "topic": "{objective}",
  "executive_summary": "High-level summary of the findings...",
  "key_findings": ["Finding 1 with concrete metrics", "Finding 2..."],
  "claims": [
    {{
      "claim": "Specific factual proposition",
      "source_url": "https://source.com/page",
      "confidence": 0.95,
      "corroborated": true
    }}
  ],
  "sources_consulted": ["https://..."]
}}
"""
        response = await self.openai.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_prompt}
            ],
            response_format={"type": "json_object"}
        )
        
        raw_json = json.loads(response.choices[0].message.content or "{}")
        raw_json["generated_at"] = "2026-08-19T00:00:00Z"
        return ResearchDossier(**raw_json)

    async def persist_and_notify(self, dossier: ResearchDossier):
        """Persists dossier to Supabase and dispatches webhook notifications."""
        # 1. Supabase Persistence (if configured)
        if self.supabase:
            try:
                self.supabase.table("research_dossiers").insert({
                    "topic": dossier.topic,
                    "executive_summary": dossier.executive_summary,
                    "payload": dossier.model_dump(),
                    "created_at": dossier.generated_at
                }).execute()
                logger.info("Dossier successfully saved to Supabase.")
            except Exception as e:
                logger.error(f"Supabase persistence failed: {e}")

        # 2. Discord Webhook Dispatch (if configured)
        webhook_url = os.getenv("DISCORD_WEBHOOK_URL")
        if webhook_url:
            async with httpx.AsyncClient() as client:
                embed = {
                    "title": f"🔬 Research Completed: {dossier.topic}",
                    "description": dossier.executive_summary[:1800],
                    "color": 3447003,
                    "fields": [
                        {"name": "Sources Consulted", "value": str(len(dossier.sources_consulted)), "inline": True},
                        {"name": "Verified Claims", "value": str(len(dossier.claims)), "inline": True}
                    ]
                }
                await client.post(webhook_url, json={"embeds": [embed]})
                logger.info("Webhook notification sent.")

    async def run(self, objective: str) -> ResearchDossier:
        try:
            # Step 1: Query generation
            queries = await self.generate_search_queries(objective)
            
            # Step 2: Multi-query Brave Search
            search_tasks = [self.brave.search(q, count=4) for q in queries]
            search_results_nested = await asyncio.gather(*search_tasks)
            flat_results = [item for sublist in search_results_nested for item in sublist]
            
            # Step 3: Triage & Firecrawl extraction
            crawled_docs = await self.triage_and_extract(flat_results, max_documents=5)
            if not crawled_docs:
                raise RuntimeError("No documents were successfully extracted.")
                
            # Step 4: LLM Synthesis
            dossier = await self.synthesize_dossier(objective, crawled_docs)
            
            # Step 5: Save & Broadcast
            await self.persist_and_notify(dossier)
            
            return dossier
            
        finally:
            await self.brave.close()
            await self.firecrawl.close()

# ---------------------------------------------------------------------------
# CLI Execution Entrypoint
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    async def main():
        agent = ResearchOrchestrator()
        objective = "Compare current 2026 latency benchmarks and cost per million tokens for Pinecone vs Qdrant Cloud"
        print(f"[*] Starting autonomous research on: '{objective}'")
        dossier = await agent.run(objective)
        print("
" + "="*80)
        print(f"EXECUTIVE SUMMARY:
{dossier.executive_summary}
")
        print("KEY FINDINGS:")
        for idx, finding in enumerate(dossier.key_findings, 1):
            print(f"  {idx}. {finding}")
        print("
VERIFIED SOURCES:")
        for s in dossier.sources_consulted:
            print(f"  - {s}")
        print("="*80)

    asyncio.run(main())

import re
from typing import List, Tuple

def validate_citations(dossier_claims: List[dict], extracted_urls: List[str]) -> List[Tuple[dict, bool]]:
    """
    Cross-checks all cited URLs against the verified extraction pool.
    Flags any synthetic or mutated URLs generated by the LLM.
    """
    verified_url_set = set(extracted_urls)
    results = []

    for item in dossier_claims:
        claimed_url = item.get("source_url", "")
        # Exact match check
        if claimed_url in verified_url_set:
            results.append((item, True))
        else:
            # Fuzzy origin check (domain match)
            domain_match = any(
                re.search(re.escape(claimed_url.split("//")[-1].split("/")[0]), v_url)
                for v_url in verified_url_set
            )
            results.append((item, domain_match))

    return results

# modal_agent_cron.py
import modal
from autonomous_researcher import ResearchOrchestrator

app = modal.App("firecrawl-research-agent")

image = (
    modal.Image.debian_slim()
    .pip_install("httpx", "openai", "supabase", "pydantic", "rich")
)

@app.function(
    image=image,
    schedule=modal.Cron("0 8 * * 1"), # Runs every Monday at 8:00 AM UTC
    secrets=[modal.Secret.from_name("research-agent-secrets")],
    timeout=600
)
async def scheduled_weekly_research():
    topics = [
        "State of AI Agent Frameworks (LangGraph vs CrewAI vs OpenAI SDK) in 2026",
        "Recent Breakthroughs in Prompt Injection Mitigation and Guardrail Architecture"
    ]
    agent = ResearchOrchestrator()
    for topic in topics:
        dossier = await agent.run(topic)
        print(f"Generated dossier for '{topic}': {len(dossier.claims)} verified claims.")

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY autonomous_researcher.py .
COPY server.py .

EXPOSE 8080
CMD ["uvicorn", "server:app", "--host", "0.0.0.0", "--port", "8080"]