🤯 I Built an Agentic Deep Researcher & It's MCP Server 🕵🏻‍♂️

deep_researcher_agent/
├── app.py              # Streamlit web UI
├── agents.py           # Multi-agent logic
├── server.py           # MCP server
├── assets/             # Images and demos
├── pyproject.toml      # Dependency config
└── README.md

import os
from agno.agent import Agent
from agno.models.nebius import Nebius
from dotenv import load_dotenv
from typing import Iterator
from agno.utils.log import logger
from agno.utils.pprint import pprint_run_response
from agno.tools.scrapegraph import ScrapeGraphTools
from agno.workflow import RunEvent, RunResponse, Workflow
from pydantic import BaseModel, Field

load_dotenv()

class DeepResearcherAgent(Workflow):
    """
    A multi-stage research workflow that:
    1. Gathers information from the web using advanced scraping tools.
    2. Analyzes and synthesizes the findings.
    3. Produces a clear, well-structured report.
    """

searcher: Agent = Agent(
        tools=[ScrapeGraphTools(api_key=os.getenv("SGAI_API_KEY"))],
        model=Nebius(
            id="deepseek-ai/DeepSeek-V3-0324", api_key=os.getenv("NEBIUS_API_KEY")
        ),
        show_tool_calls=True,
        markdown=True,
        description=("..."),
        instructions=(
            "...",
        ),
    )

 # Analyst: Synthesizes and interprets the research findings
    analyst: Agent = Agent(
        model=Nebius(
            id="deepseek-ai/DeepSeek-V3-0324", api_key=os.getenv("NEBIUS_API_KEY")
        ),
        markdown=True,
        description=(
            "..."
        ),
        instructions=(
            "..."
        ),
    )

  # Writer: Produces a final, polished report
    writer: Agent = Agent(
        model=Nebius(
            id="deepseek-ai/DeepSeek-V3-0324", api_key=os.getenv("NEBIUS_API_KEY")
        ),
        markdown=True,
        description=(
            "..."
        ),
        instructions=(
            "..."
        ),
    )

    def run(self, topic: str) -> Iterator[RunResponse]:
        """
        Orchestrates the research, analysis, and report writing process for a given topic.
        """
        logger.info(f"Running deep researcher agent for topic: {topic}")

        # Step 1: Research
        research_content = self.searcher.run(topic)
        # logger.info(f"Searcher content: {research_content.content}")

        logger.info("Analysis started")
        # Step 2: Analysis
        analysis = self.analyst.run(research_content.content)
        # logger.info(f"Analyst analysis: {analysis.content}")

        logger.info("Report Writing Started")
        # Step 3: Report Writing
        report = self.writer.run(analysis.content, stream=True)
        yield from report

def run_research(query: str) -> str:
    agent = DeepResearcherAgent()
    final_report_iterator = agent.run(
        topic=query,
    )
    logger.info("Report Generated")

    # Collect all streaming content into a single string
    full_report = ""
    for chunk in final_report_iterator:
        if chunk.content:
            full_report += chunk.content

    return full_report

if __name__ == "__main__":
    topic = "Extract information about Nebius AI Studio, including its features, capabilities, and applications from available sources."
    response = run_research(topic)
    print(response)

import streamlit as st
from agents import DeepResearcherAgent
import time
import base64

st.set_page_config(
    page_title="Deep Research Agent",
    page_icon="🔎",
)

with open("./assets/scrapegraph.png", "rb") as scrapegraph_file:
    scrapegraph_base64 = base64.b64encode(scrapegraph_file.read()).decode()

    # Create title with embedded images
    title_html = f"""
    <div style="display: flex; justify-content: center; align-items: center; width: 100%; padding: 32px 0 24px 0;">
        <h1 style="margin: 0; padding: 0; font-size: 2.5rem; font-weight: bold;">
            <span style="font-size:2.5rem;">🔎</span> Agentic Deep Searcher with 
            <span style="color: #fb542c;">Agno</span> & 
            <span style="color: #8564ff;">Scrapegraph</span>
            <img src="data:image/png;base64,{scrapegraph_base64}" style="height: 60px; margin-left: 12px; vertical-align: middle;"/>
        </h1>
    </div>
    """
    st.markdown(title_html, unsafe_allow_html=True)

with st.sidebar:

    st.image("./assets/nebius.png", width=150)
    nebius_api_key = st.text_input("Enter your Nebius API key", type="password")
    st.divider()

    st.subheader("Enter Scrapegraph API key")
    scrapegraph_api_key = st.text_input(
        "Enter your Scrapegraph API key", type="password"
    )
    st.divider()

    st.header("About")
    st.markdown(
        """
    This application is powered by a `DeepResearcherAgent` which leverages multiple AI agents for a comprehensive research process:
    - **Searcher**: Finds and extracts information from the web.
    - **Analyst**: Synthesizes and interprets the research findings.
    - **Writer**: Produces a final, polished report.
    """
    )
    st.markdown("---")
    st.markdown(
        "Developed with ❤️ by [Arindam Majumder](https://www.youtube.com/c/Arindam_1729)"
    )

# Chat input at the bottom
user_input = st.chat_input("Ask a question about your documents...")

if user_input:
    try:
        agent = DeepResearcherAgent()
        with st.status("Executing research plan...", expanded=True) as status:
            # PHASE 1: Researching
            phase1_msg = "🧠 **Phase 1: Researching** - Finding and extracting relevant information from the web..."
            status.write(phase1_msg)
            research_content = agent.searcher.run(user_input)

            # PHASE 2: Analyzing
            phase2_msg = "🔬 **Phase 2: Analyzing** - Synthesizing and interpreting the research findings..."
            status.write(phase2_msg)
            analysis = agent.analyst.run(research_content.content)

            # PHASE 3: Writing Report
            phase3_msg = (
                "✍️ **Phase 3: Writing Report** - Producing a final, polished report..."
            )
            status.write(phase3_msg)
            report_iterator = agent.writer.run(analysis.content, stream=True)

        # Move report display outside of status block
        full_report = ""
        report_container = st.empty()
        for chunk in report_iterator:
            if chunk.content:
                full_report += chunk.content
                report_container.markdown(full_report)

    except Exception as e:
        st.error(f"An error occurred: {e}")

import asyncio
from mcp.server.fastmcp import FastMCP
from agents import run_research

mcp = FastMCP("deep_researcher_agent")


@mcp.tool()
def deep_researcher_agent(query: str) -> str:
    """Run Deep Researcher Agent for given user query. Can do both standard and deep web search.

    Args:
        query (str): The research query or question.

    Returns:
        str: The research response from the Deep Researcher Agent.
    """

    return run_research(query)

# Run the server
if __name__ == "__main__":
    mcp.run(transport="stdio")

# add this inside ./.cursor/mcp.json
{
   "mcpServers": {
     "deep_researcher_agent": {
       "command": "python",
       "args": [
         "--directory",
         "/Users/arindammajumder/Developer/Python/awesome-llm-apps/advance_ai_agents/deep_researcher_agent",
         "run",
         "server.py"
       ],
       "env": {
         "NEBIUS_API_KEY": "your_nebius_api_key_here",
         "SGAI_API_KEY": "your_scrapegraph_api_key_here"
       }
     }
   }
 }

uv run streamlit run app.py

uv run python agents.py

{
  "mcpServers": {
    "deep_researcher_agent": {
      "command": "python",
      "args": [
        "--directory",
        "/Your/Path/to/directory/awesome-ai-apps/advance_ai_agents/deep_researcher_agent",
        "run",
        "server.py"
      ],
      "env": {
        "NEBIUS_API_KEY": "your_nebius_api_key_here",
        "SGAI_API_KEY": "your_scrapegraph_api_key_here"
      }
    }
  }
}