| import os |
| import time |
| import re |
| import html |
| import asyncio |
| import ipaddress |
| import socket |
| from typing import Optional, Dict, Any, List, Tuple |
| import fnmatch |
| from urllib.parse import urlsplit |
| from datetime import datetime, timezone |
|
|
| 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 |
|
|
| from analytics import record_request, last_n_days_count_df |
|
|
| |
| |
| |
| SERPER_API_KEY = os.getenv("SERPER_API_KEY") |
| SERPER_SEARCH_ENDPOINT = "https://google.serper.dev/search" |
| SERPER_NEWS_ENDPOINT = "https://google.serper.dev/news" |
| HEADERS = {"X-API-KEY": SERPER_API_KEY or "", "Content-Type": "application/json"} |
|
|
| |
| SERPER_TIMEOUT = httpx.Timeout(connect=5.0, read=10.0, write=5.0, pool=5.0) |
| WEB_TIMEOUT = httpx.Timeout(connect=5.0, read=20.0, write=5.0, pool=5.0) |
|
|
| SERPER_LIMITS = httpx.Limits( |
| max_keepalive_connections=int(os.getenv("SERPER_KEEPALIVE", "32")), |
| max_connections=int(os.getenv("SERPER_MAX_CONNECTIONS", "128")), |
| ) |
| WEB_LIMITS = httpx.Limits( |
| max_keepalive_connections=int(os.getenv("WEB_KEEPALIVE", "128")), |
| max_connections=int(os.getenv("WEB_MAX_CONNECTIONS", "512")), |
| ) |
|
|
| serper_client = httpx.AsyncClient( |
| timeout=SERPER_TIMEOUT, |
| limits=SERPER_LIMITS, |
| http2=True, |
| headers=HEADERS, |
| ) |
|
|
| DEFAULT_USER_AGENT = os.getenv( |
| "FETCH_USER_AGENT", |
| "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 " |
| "(KHTML, like Gecko) Chrome/124.0 Safari/537.36", |
| ) |
|
|
| web_client = httpx.AsyncClient( |
| timeout=WEB_TIMEOUT, |
| limits=WEB_LIMITS, |
| http2=True, |
| follow_redirects=True, |
| headers={"User-Agent": DEFAULT_USER_AGENT}, |
| ) |
|
|
| |
| GLOBAL_RATE = parse(os.getenv("GLOBAL_RATE", "3000/minute")) |
| PER_IP_RATE = parse(os.getenv("PER_IP_RATE", "60/minute")) |
| storage = MemoryStorage() |
| limiter = MovingWindowRateLimiter(storage) |
|
|
| |
| FETCH_MAX_BYTES = max(1024, int(os.getenv("FETCH_MAX_BYTES", "1500000"))) |
| FETCH_CONCURRENCY = max(1, int(os.getenv("FETCH_CONCURRENCY", "64"))) |
| SEARCH_CONCURRENCY = max(1, int(os.getenv("SEARCH_CONCURRENCY", "64"))) |
| EXTRACT_CONCURRENCY = max( |
| 1, |
| int( |
| os.getenv( |
| "EXTRACT_CONCURRENCY", |
| str(max(4, (os.cpu_count() or 2) * 2)), |
| ) |
| ), |
| ) |
|
|
| SEARCH_CACHE_TTL = max(0, int(os.getenv("SEARCH_CACHE_TTL", "30"))) |
| FETCH_CACHE_TTL = max(0, int(os.getenv("FETCH_CACHE_TTL", "300"))) |
|
|
| |
| def _env_flag(name: str, default: bool = False) -> bool: |
| """Parse boolean-like env vars such as 1/true/yes/on.""" |
| v = os.getenv(name) |
| if v is None: |
| return default |
| return str(v).strip().lower() in {"1", "true", "yes", "on", "y"} |
|
|
| |
| FETCH_ALLOW_PRIVATE = _env_flag("FETCH_ALLOW_PRIVATE", False) |
|
|
| |
| |
| FETCH_PRIVATE_ALLOWLIST = [ |
| p for p in re.split(r"[\s,]+", os.getenv("FETCH_PRIVATE_ALLOWLIST", "").strip()) if p |
| ] |
|
|
| _search_cache: Dict[Tuple[str, str, int], Dict[str, Any]] = {} |
| _fetch_cache: Dict[str, Dict[str, Any]] = {} |
| _search_cache_lock: Optional[asyncio.Lock] = None |
| _fetch_cache_lock: Optional[asyncio.Lock] = None |
| _search_sema: Optional[asyncio.Semaphore] = None |
| _fetch_sema: Optional[asyncio.Semaphore] = None |
| _extract_sema: Optional[asyncio.Semaphore] = None |
|
|
|
|
| |
| |
| |
| def _get_cache_lock(name: str) -> asyncio.Lock: |
| global _search_cache_lock, _fetch_cache_lock |
| if name == "search": |
| if _search_cache_lock is None: |
| _search_cache_lock = asyncio.Lock() |
| return _search_cache_lock |
| if name == "fetch": |
| if _fetch_cache_lock is None: |
| _fetch_cache_lock = asyncio.Lock() |
| return _fetch_cache_lock |
| raise ValueError(f"Unknown cache lock: {name}") |
|
|
|
|
| def _get_semaphore(name: str) -> asyncio.Semaphore: |
| global _search_sema, _fetch_sema, _extract_sema |
| if name == "search": |
| if _search_sema is None: |
| _search_sema = asyncio.Semaphore(SEARCH_CONCURRENCY) |
| return _search_sema |
| if name == "fetch": |
| if _fetch_sema is None: |
| _fetch_sema = asyncio.Semaphore(FETCH_CONCURRENCY) |
| return _fetch_sema |
| if name == "extract": |
| if _extract_sema is None: |
| _extract_sema = asyncio.Semaphore(EXTRACT_CONCURRENCY) |
| return _extract_sema |
| raise ValueError(f"Unknown semaphore: {name}") |
|
|
|
|
| async def _cache_get(name: str, cache: Dict[Any, Any], key: Any): |
| lock = _get_cache_lock(name) |
| async with lock: |
| entry = cache.get(key) |
| if not entry: |
| return None |
| if time.time() > entry["expires_at"]: |
| cache.pop(key, None) |
| return None |
| return entry["value"] |
|
|
|
|
| async def _cache_set(name: str, cache: Dict[Any, Any], key: Any, value: Any, ttl: int): |
| if ttl <= 0: |
| return |
| lock = _get_cache_lock(name) |
| async with lock: |
| cache[key] = {"expires_at": time.time() + ttl, "value": value} |
|
|
|
|
| def _client_ip(request: Optional[gr.Request]) -> str: |
| try: |
| if request is None: |
| return "unknown" |
| headers = getattr(request, "headers", None) or {} |
| xff = headers.get("x-forwarded-for") |
| if xff: |
| return xff.split(",")[0].strip() |
| client = getattr(request, "client", None) |
| if client and getattr(client, "host", None): |
| return client.host |
| except Exception: |
| pass |
| return "unknown" |
|
|
|
|
| def _host_matches_allowlist(host: str) -> bool: |
| """Return True if host matches any pattern in FETCH_PRIVATE_ALLOWLIST.""" |
| if not host: |
| return False |
| for pat in FETCH_PRIVATE_ALLOWLIST: |
| |
| if host == pat or fnmatch.fnmatch(host, pat): |
| return True |
| return False |
|
|
|
|
| async def _resolve_addresses(host: str) -> List[str]: |
| def _resolve() -> List[str]: |
| try: |
| return list({ai[4][0] for ai in socket.getaddrinfo(host, None)}) |
| except Exception: |
| return [] |
|
|
| return await asyncio.to_thread(_resolve) |
|
|
|
|
| async def _host_is_public(host: str) -> Tuple[bool, List[str]]: |
| """Return (is_public, resolved_addresses). |
| |
| - If resolution fails, treat as public and let HTTP request decide. |
| - Honors allowlist/env flags via the caller. |
| """ |
| if not host: |
| return False, [] |
|
|
| addresses = await _resolve_addresses(host) |
| if not addresses: |
| return True, [] |
|
|
| for addr in addresses: |
| ip_obj = ipaddress.ip_address(addr) |
| if ( |
| ip_obj.is_private |
| or ip_obj.is_loopback |
| or ip_obj.is_link_local |
| or ip_obj.is_multicast |
| or ip_obj.is_reserved |
| or ip_obj.is_unspecified |
| ): |
| return False, addresses |
| return True, addresses |
|
|
|
|
| async def _check_rate_limits(bucket: str, ip: str) -> Optional[str]: |
| if not await limiter.hit(GLOBAL_RATE, "global"): |
| return f"Global rate limit exceeded. Limit: {GLOBAL_RATE}." |
| if ip != "unknown": |
| if not await limiter.hit(PER_IP_RATE, f"{bucket}:{ip}"): |
| return f"Per-IP rate limit exceeded. Limit: {PER_IP_RATE}." |
| return None |
|
|
|
|
| def _domain_from_url(url: str) -> str: |
| try: |
| netloc = urlsplit(url).netloc |
| return netloc.replace("www.", "") |
| except Exception: |
| return "" |
|
|
|
|
| def _iso_date_or_unknown(date_str: Optional[str]) -> Optional[str]: |
| if not date_str: |
| return None |
| try: |
| return dateparser.parse(date_str, fuzzy=True).strftime("%Y-%m-%d") |
| except Exception: |
| return None |
|
|
|
|
| def _extract_title_from_html(html_text: str) -> Optional[str]: |
| m = re.search(r"<title[^>]*>(.*?)</title>", html_text, re.IGNORECASE | re.DOTALL) |
| if not m: |
| return None |
| title = re.sub(r"\s+", " ", m.group(1)).strip() |
| return html.unescape(title) if title else None |
|
|
|
|
| |
| |
| |
| async def search( |
| query: str, |
| search_type: str = "search", |
| num_results: Optional[int] = 4, |
| request: Optional[gr.Request] = None, |
| ) -> Dict[str, Any]: |
| """Perform a web or news search via Serper and return metadata only.""" |
| start_time = time.time() |
|
|
| if not query or not query.strip(): |
| await record_request("search") |
| return {"error": "Missing 'query'. Please provide a search query string."} |
|
|
| query = query.strip() |
| if num_results is None: |
| num_results = 4 |
| try: |
| num_results = max(1, min(20, int(num_results))) |
| except (TypeError, ValueError): |
| num_results = 4 |
|
|
| if search_type not in ["search", "news"]: |
| search_type = "search" |
|
|
| if not SERPER_API_KEY: |
| await record_request("search") |
| return { |
| "error": "SERPER_API_KEY is not set. Export SERPER_API_KEY and try again." |
| } |
|
|
| ip = _client_ip(request) |
|
|
| try: |
| rl_message = await _check_rate_limits("search", ip) |
| if rl_message: |
| await record_request("search") |
| return {"error": rl_message} |
|
|
| cache_key = (query, search_type, num_results) |
| cached = await _cache_get("search", _search_cache, cache_key) |
| if cached: |
| await record_request("search") |
| return cached |
|
|
| endpoint = ( |
| SERPER_NEWS_ENDPOINT if search_type == "news" else SERPER_SEARCH_ENDPOINT |
| ) |
| payload: Dict[str, Any] = {"q": query, "num": num_results} |
| if search_type == "news": |
| payload["type"] = "news" |
| payload["page"] = 1 |
|
|
| semaphore = _get_semaphore("search") |
| await semaphore.acquire() |
| try: |
| resp = await serper_client.post(endpoint, json=payload) |
| finally: |
| semaphore.release() |
|
|
| if resp.status_code != 200: |
| await record_request("search") |
| return { |
| "error": f"Search API returned status {resp.status_code}. Check your API key and query." |
| } |
|
|
| data = resp.json() |
| raw_results: List[Dict[str, Any]] = ( |
| data.get("news", []) if search_type == "news" else data.get("organic", []) |
| ) |
|
|
| formatted: List[Dict[str, Any]] = [] |
| for idx, item in enumerate(raw_results[:num_results], start=1): |
| entry = { |
| "position": idx, |
| "title": item.get("title"), |
| "link": item.get("link"), |
| "domain": _domain_from_url(item.get("link", "")), |
| "snippet": item.get("snippet") or item.get("description"), |
| } |
| if search_type == "news": |
| entry["source"] = item.get("source") |
| entry["date"] = _iso_date_or_unknown(item.get("date")) |
| formatted.append(entry) |
|
|
| result = { |
| "query": query, |
| "search_type": search_type, |
| "count": len(formatted), |
| "results": formatted, |
| "duration_s": round(time.time() - start_time, 2), |
| } |
|
|
| if not formatted: |
| result["message"] = f"No {search_type} results found." |
|
|
| await _cache_set("search", _search_cache, cache_key, result, SEARCH_CACHE_TTL) |
| await record_request("search") |
|
|
| return result |
|
|
| except Exception as e: |
| await record_request("search") |
| return {"error": f"Search failed: {str(e)}"} |
|
|
|
|
| |
| |
| |
| async def fetch( |
| url: str, |
| timeout: int = 20, |
| request: Optional[gr.Request] = None, |
| ) -> Dict[str, Any]: |
| """Fetch a single URL and extract the main readable content.""" |
| start_time = time.time() |
|
|
| if not url or not isinstance(url, str): |
| await record_request("fetch") |
| return {"error": "Missing 'url'. Please provide a valid URL string."} |
| if not url.lower().startswith(("http://", "https://")): |
| await record_request("fetch") |
| return {"error": "URL must start with http:// or https://."} |
|
|
| try: |
| timeout = max(5, min(60, int(timeout))) |
| except (TypeError, ValueError): |
| timeout = 20 |
|
|
| ip = _client_ip(request) |
|
|
| try: |
| host = urlsplit(url).hostname or "" |
| if not host: |
| await record_request("fetch") |
| return {"error": "Invalid URL; unable to determine host."} |
| rl_message = await _check_rate_limits("fetch", ip) |
| if rl_message: |
| await record_request("fetch") |
| return {"error": rl_message} |
|
|
| cache_key = (url, timeout) |
| cached = await _cache_get("fetch", _fetch_cache, cache_key) |
| if cached: |
| await record_request("fetch") |
| return cached |
|
|
| is_public, addrs = await _host_is_public(host) |
| if not is_public and not (FETCH_ALLOW_PRIVATE or _host_matches_allowlist(host)): |
| await record_request("fetch") |
| detail = f" (resolved: {', '.join(addrs)})" if addrs else "" |
| return { |
| "error": "Refusing to fetch private or local addresses." + detail, |
| "host": host, |
| } |
|
|
| fetch_sema = _get_semaphore("fetch") |
| await fetch_sema.acquire() |
| try: |
| async with web_client.stream("GET", url, timeout=timeout) as resp: |
| status_code = resp.status_code |
| total = 0 |
| chunks: List[bytes] = [] |
| async for chunk in resp.aiter_bytes(): |
| total += len(chunk) |
| if total > FETCH_MAX_BYTES: |
| break |
| chunks.append(chunk) |
| body = b"".join(chunks) |
| final_url_str = str(resp.url) |
| encoding = resp.encoding or "utf-8" |
| finally: |
| fetch_sema.release() |
|
|
| truncated = total > FETCH_MAX_BYTES |
| |
| |
| try: |
| final_host = urlsplit(final_url_str).hostname or "" |
| except Exception: |
| final_host = "" |
| if final_host and not (FETCH_ALLOW_PRIVATE or _host_matches_allowlist(final_host)): |
| final_public, _ = await _host_is_public(final_host) |
| if not final_public: |
| await record_request("fetch") |
| return { |
| "error": "Refusing to fetch private or local addresses after redirect.", |
| "host": final_host, |
| } |
| text = body.decode(encoding, errors="ignore") |
|
|
| extract_sema = _get_semaphore("extract") |
| await extract_sema.acquire() |
| try: |
| content = await asyncio.to_thread( |
| trafilatura.extract, |
| text, |
| include_formatting=False, |
| include_comments=False, |
| ) |
| finally: |
| extract_sema.release() |
|
|
| content = (content or "").strip() |
| title = _extract_title_from_html(text) or "" |
| domain = _domain_from_url(final_url_str) |
| word_count = len(content.split()) if content else 0 |
|
|
| result = { |
| "url": url, |
| "final_url": final_url_str, |
| "domain": domain, |
| "status_code": status_code, |
| "title": title, |
| "fetched_at": datetime.now(timezone.utc).isoformat(), |
| "word_count": word_count, |
| "content": content, |
| "duration_s": round(time.time() - start_time, 2), |
| } |
|
|
| if truncated: |
| result["truncated"] = True |
|
|
| await _cache_set("fetch", _fetch_cache, cache_key, result, FETCH_CACHE_TTL) |
| await record_request("fetch") |
| return result |
|
|
| except httpx.HTTPError as e: |
| await record_request("fetch") |
| return {"error": f"Network error while fetching: {str(e)}"} |
| except Exception as e: |
| await record_request("fetch") |
| return {"error": f"Unexpected error while fetching: {str(e)}"} |
|
|
|
|
| |
| |
| |
| with gr.Blocks(title="Web MCP Server") as demo: |
| gr.HTML( |
| """ |
| <div style="background-color: rgba(59, 130, 246, 0.1); border: 1px solid rgba(59, 130, 246, 0.3); border-radius: 8px; padding: 12px; margin-bottom: 16px; text-align: center;"> |
| <p style="color: rgb(59, 130, 246); margin: 0; font-size: 14px; font-weight: 500;"> |
| 🤝 Community resource — please use responsibly to keep this service available for everyone |
| </p> |
| </div> |
| """ |
| ) |
|
|
| gr.Markdown("# 🔍 Web Search MCP Server") |
| gr.Markdown( |
| "This server provides two composable MCP tools: **search** (metadata only) and **fetch** (single-URL extraction)." |
| ) |
|
|
| with gr.Tabs(): |
| with gr.Tab("App"): |
| with gr.Row(): |
| |
| with gr.Column(scale=3): |
| gr.Markdown("## Search (metadata only)") |
| query_input = gr.Textbox( |
| label="Search Query", |
| placeholder='e.g. "OpenAI news", "climate change 2024", "React hooks useState"', |
| info="Required", |
| ) |
| search_type_input = gr.Radio( |
| choices=["search", "news"], |
| value="search", |
| label="Search Type", |
| info="Choose general web search or news", |
| ) |
| num_results_input = gr.Slider( |
| minimum=1, |
| maximum=20, |
| value=4, |
| step=1, |
| label="Number of Results", |
| info="Optional (default 4)", |
| ) |
| search_button = gr.Button("Run Search", variant="primary") |
| search_output = gr.JSON( |
| label="Search Results (metadata only)", |
| ) |
|
|
| gr.Examples( |
| examples=[ |
| ["OpenAI GPT-5 latest developments", "news", 5], |
| ["React hooks useState", "search", 4], |
| ["Apple Vision Pro reviews", "search", 4], |
| ["Tesla stock price today", "news", 6], |
| ], |
| inputs=[query_input, search_type_input, num_results_input], |
| outputs=search_output, |
| fn=search, |
| cache_examples=False, |
| ) |
|
|
| |
| with gr.Column(scale=2): |
| gr.Markdown("## Fetch (single URL → extracted content)") |
| url_input = gr.Textbox( |
| label="URL", |
| placeholder="https://example.com/article", |
| info="Required: the URL to fetch and extract", |
| ) |
| timeout_input = gr.Slider( |
| minimum=5, |
| maximum=60, |
| value=20, |
| step=1, |
| label="Timeout (seconds)", |
| info="Optional (default 20)", |
| ) |
| fetch_button = gr.Button("Fetch & Extract", variant="primary") |
| fetch_output = gr.JSON(label="Fetched Content (structured)") |
|
|
| gr.Examples( |
| examples=[ |
| ["https://news.ycombinator.com/"], |
| ["https://www.python.org/dev/peps/pep-0008/"], |
| ["https://en.wikipedia.org/wiki/Model_Context_Protocol"], |
| ], |
| inputs=[url_input], |
| outputs=fetch_output, |
| fn=fetch, |
| cache_examples=False, |
| ) |
|
|
| |
| search_button.click( |
| fn=search, |
| inputs=[query_input, search_type_input, num_results_input], |
| outputs=search_output, |
| api_name=False, |
| ) |
| fetch_button.click( |
| fn=fetch, |
| inputs=[url_input, timeout_input], |
| outputs=fetch_output, |
| api_name=False, |
| ) |
|
|
| with gr.Tab("Analytics"): |
| gr.Markdown("## Community Usage Analytics") |
| gr.Markdown("Daily request counts (UTC), split by tool.") |
|
|
| with gr.Row(): |
| with gr.Column(): |
| search_plot = gr.BarPlot( |
| value=last_n_days_count_df("search", 14), |
| x="date", |
| y="count", |
| title="Daily Search Count", |
| tooltip=["date", "count", "full_date"], |
| height=350, |
| x_label_angle=-45, |
| container=False, |
| ) |
| with gr.Column(): |
| fetch_plot = gr.BarPlot( |
| value=last_n_days_count_df("fetch", 14), |
| x="date", |
| y="count", |
| title="Daily Fetch Count", |
| tooltip=["date", "count", "full_date"], |
| height=350, |
| x_label_angle=-45, |
| container=False, |
| ) |
|
|
| |
| demo.load( |
| fn=lambda: ( |
| last_n_days_count_df("search", 14), |
| last_n_days_count_df("fetch", 14), |
| ), |
| outputs=[search_plot, fetch_plot], |
| api_name=False, |
| ) |
|
|
| |
| gr.api(search, api_name="search") |
| gr.api(fetch, api_name="fetch") |
|
|
|
|
| demo.queue( |
| max_size=int(os.getenv("GRADIO_MAX_QUEUE", "256")), |
| default_concurrency_limit=int(os.getenv("GRADIO_CONCURRENCY", "32")), |
| ) |
|
|
|
|
| if __name__ == "__main__": |
| |
| demo.launch(mcp_server=True, show_api=True) |
|
|
