app.py

"""
Web Search MCP Server - Feed LLMs with fresh sources
====================================================

Prerequisites
-------------
$ pip install "gradio[mcp]" httpx trafilatura python-dateutil limits

Environment
-----------
export SERPER_API_KEY="YOUR-KEY-HERE"

Usage
-----
python app_mcp.py
Then connect to: http://localhost:7860/gradio_api/mcp/sse
"""

import os
import asyncio
from typing import Optional
import httpx
import trafilatura
import gradio as gr
from dateutil import parser as dateparser
from limits import parse
from limits.aio.storage import MemoryStorage
from limits.aio.strategies import MovingWindowRateLimiter

# Configuration
SERPER_API_KEY = os.getenv("SERPER_API_KEY")
SERPER_ENDPOINT = "https://google.serper.dev/news"
HEADERS = {"X-API-KEY": SERPER_API_KEY, "Content-Type": "application/json"}

# Rate limiting
storage = MemoryStorage()
limiter = MovingWindowRateLimiter(storage)
rate_limit = parse("200/hour")


async def search_web(query: str, num_results: Optional[int] = 4) -> str:
    """
    Search the web for recent news and information, returning extracted content.

    This tool searches for recent news articles related to your query and extracts
    the main content from each article, providing you with fresh, relevant information
    from the web.

    Args:
        query (str): The search query. This is REQUIRED. Examples: "apple inc earnings",
                    "climate change 2024", "AI developments"
        num_results (int): Number of results to fetch. This is OPTIONAL. Default is 4.
                          Range: 1-20. More results = more context but longer response time.

    Returns:
        str: Formatted text containing extracted article content with metadata (title,
             source, date, URL, and main text) for each result, separated by dividers.
             Returns error message if API key is missing or search fails.

    Examples:
        - search_web("OpenAI news", 5) - Get 5 recent news articles about OpenAI
        - search_web("python 3.13 features") - Get 4 articles about Python 3.13
        - search_web("stock market today", 10) - Get 10 articles about today's market
    """
    if not SERPER_API_KEY:
        return "Error: SERPER_API_KEY environment variable is not set. Please set it to use this tool."

    # Validate and constrain num_results
    if num_results is None:
        num_results = 4
    num_results = max(1, min(20, num_results))

    try:
        # Check rate limit
        if not await limiter.hit(rate_limit, "global"):
            return "Error: Rate limit exceeded. Please try again later (limit: 200 requests per hour)."

        # Search for news
        payload = {"q": query, "type": "news", "num": num_results, "page": 1}
        async with httpx.AsyncClient(timeout=15) as client:
            resp = await client.post(SERPER_ENDPOINT, headers=HEADERS, json=payload)

        if resp.status_code != 200:
            return f"Error: Search API returned status {resp.status_code}. Please check your API key and try again."

        news_items = resp.json().get("news", [])
        if not news_items:
            return (
                f"No results found for query: '{query}'. Try a different search term."
            )

        # Fetch HTML content concurrently
        urls = [n["link"] for n in news_items]
        async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client:
            tasks = [client.get(u) for u in urls]
            responses = await asyncio.gather(*tasks, return_exceptions=True)

        # Extract and format content
        chunks = []
        successful_extractions = 0

        for meta, response in zip(news_items, responses):
            if isinstance(response, Exception):
                continue

            # Extract main text content
            body = trafilatura.extract(
                response.text, include_formatting=False, include_comments=False
            )

            if not body:
                continue

            successful_extractions += 1

            # Parse and format date
            try:
                date_iso = dateparser.parse(meta.get("date", ""), fuzzy=True).strftime(
                    "%Y-%m-%d"
                )
            except Exception:
                date_iso = meta.get("date", "Unknown")

            # Format the chunk
            chunk = (
                f"## {meta['title']}\n"
                f"**Source:** {meta['source']}   "
                f"**Date:** {date_iso}\n"
                f"**URL:** {meta['link']}\n\n"
                f"{body.strip()}\n"
            )
            chunks.append(chunk)

        if not chunks:
            return f"Found {len(news_items)} results for '{query}', but couldn't extract readable content from any of them. The websites might be blocking automated access."

        result = "\n---\n".join(chunks)
        summary = f"Successfully extracted content from {successful_extractions} out of {len(news_items)} search results for query: '{query}'\n\n---\n\n"

        return summary + result

    except Exception as e:
        return f"Error occurred while searching: {str(e)}. Please try again or check your query."


# Create Gradio interface
with gr.Blocks(title="Web Search MCP Server") as demo:
    gr.Markdown(
        """
        # 🔍 Web Search MCP Server
        
        This MCP server provides web search capabilities to LLMs. It searches for recent news 
        and extracts the main content from articles.
        
        **Note:** This interface is primarily designed for MCP tool usage by LLMs, but you can 
        also test it manually below.
        """
    )

    with gr.Row():
        query_input = gr.Textbox(
            label="Search Query",
            placeholder='e.g. "OpenAI news", "climate change 2024", "AI developments"',
            info="Required: Enter your search query",
        )
        num_results_input = gr.Slider(
            minimum=1,
            maximum=20,
            value=4,
            step=1,
            label="Number of Results",
            info="Optional: How many articles to fetch (default: 4)",
        )

    output = gr.Textbox(
        label="Extracted Content",
        lines=25,
        max_lines=50,
        info="The extracted article content will appear here",
    )

    search_button = gr.Button("Search", variant="primary")

    # Add examples
    gr.Examples(
        examples=[
            ["OpenAI GPT-5 news", 5],
            ["climate change 2024", 4],
            ["artificial intelligence breakthroughs", 8],
            ["stock market today", 6],
            ["python programming updates", 4],
        ],
        inputs=[query_input, num_results_input],
        outputs=output,
        fn=search_web,
        cache_examples=False,
    )

    search_button.click(
        fn=search_web, inputs=[query_input, num_results_input], outputs=output
    )


if __name__ == "__main__":
    # Launch with MCP server enabled
    # The MCP endpoint will be available at: http://localhost:7860/gradio_api/mcp/sse
    demo.launch(mcp_server=True, show_api=True)