import os import asyncio import base64 import io from typing import Optional, List, Dict, Any from PIL import Image import uvicorn from fastapi import FastAPI import gradio as gr import spaces # Hugging Face ZeroGPU module import torch from transformers import pipeline from playwright.async_api import async_playwright from mcp.server.fastmcp import FastMCP # --------------------------------------------------------- # 1. ZeroGPU Model Initialization & GPU-Decorated Functions # --------------------------------------------------------- print("Initializing ZeroGPU Pipeline...") try: research_pipe = pipeline("text-generation", model="gpt2", device="cuda" if torch.cuda.is_available() else "cpu") except Exception as e: print(f"Pipeline loading note: {e}") research_pipe = None @spaces.GPU def run_zerogpu_analysis(prompt: str) -> str: """ZeroGPU function decorated with @spaces.GPU.""" if not research_pipe: return f"ZeroGPU offline. Input snippet: {prompt[:150]}..." formatted_input = f"Hybrid Research Context:\n{prompt}\n\nSynthesis:" result = research_pipe(formatted_input, max_new_tokens=200, do_sample=True, temperature=0.7) return result[0]["generated_text"] # --------------------------------------------------------- # 2. Playwright Core Engine # --------------------------------------------------------- async def get_browser_page(p): browser = await p.chromium.launch( headless=True, args=["--no-sandbox", "--disable-setuid-sandbox", "--disable-dev-shm-usage"] ) context = await browser.new_context( user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" ) page = await context.new_page() return browser, page # --------------------------------------------------------- # 3. FastMCP Server Tools # --------------------------------------------------------- mcp = FastMCP("Hybrid Research Playwright Server") @mcp.tool() async def search_web(query: str, max_results: int = 5) -> str: """Search the web via DuckDuckGo to discover candidate research URLs.""" async with async_playwright() as p: browser, page = await get_browser_page(p) try: search_url = f"https://html.duckduckgo.com/html/?q={query.replace(' ', '+')}" await page.goto(search_url, wait_until="domcontentloaded", timeout=20000) results = [] links = await page.query_selector_all(".result__body") for link in links[:max_results]: title_el = await link.query_selector(".result__title") snippet_el = await link.query_selector(".result__snippet") url_el = await link.query_selector(".result__url") title = await title_el.inner_text() if title_el else "No title" snippet = await snippet_el.inner_text() if snippet_el else "No snippet" url = (await url_el.get_attribute("href")) if url_el else "" results.append(f"- **{title.strip()}**\n URL: {url.strip()}\n Snippet: {snippet.strip()}") await browser.close() return f"### Search Results for '{query}':\n\n" + "\n\n".join(results) except Exception as e: await browser.close() return f"Search error: {str(e)}" @mcp.tool() async def fetch_webpage_text(url: str, selector: Optional[str] = None) -> str: """Fetch readable text content from a webpage.""" async with async_playwright() as p: browser, page = await get_browser_page(p) try: await page.goto(url, wait_until="networkidle", timeout=30000) title = await page.title() if selector: elements = await page.query_selector_all(selector) content = "\n".join([await el.inner_text() for el in elements]) else: content = await page.inner_text("body") await browser.close() return f"Title: {title}\nURL: {url}\n\nContent:\n{content[:4000]}" except Exception as e: await browser.close() return f"Error fetching {url}: {str(e)}" @mcp.tool() async def extract_tables_and_data(url: str) -> str: """Extract HTML tables as Markdown formatted tables.""" async with async_playwright() as p: browser, page = await get_browser_page(p) try: await page.goto(url, wait_until="networkidle", timeout=25000) tables = await page.query_selector_all("table") extracted_tables = [] for i, table in enumerate(tables[:3]): rows = await table.query_selector_all("tr") table_md = [] for row in rows: cols = await row.query_selector_all("th, td") col_texts = [f" { (await col.inner_text()).strip() } " for col in cols] if col_texts: table_md.append("|" + "|".join(col_texts) + "|") if table_md: if len(table_md) > 1: num_cols = table_md[0].count("|") - 1 sep = "|" + "|".join(["---"] * num_cols) + "|" table_md.insert(1, sep) extracted_tables.append(f"**Table {i+1}:**\n" + "\n".join(table_md)) await browser.close() if not extracted_tables: return f"No standard HTML tables found on {url}." return f"### Tables extracted from {url}:\n\n" + "\n\n".join(extracted_tables) except Exception as e: await browser.close() return f"Table extraction error: {str(e)}" @mcp.tool() async def visual_chart_analysis_gpu(url: str, question: str) -> str: """Run ZeroGPU visual/content reasoning on page layout or text.""" async with async_playwright() as p: browser, page = await get_browser_page(p) try: await page.goto(url, wait_until="networkidle", timeout=30000) title = await page.title() text_content = await page.inner_text("body") await browser.close() prompt = f"Query: {question}\nTitle: {title}\nURL: {url}\n\nContent:\n{text_content[:1500]}" analysis = run_zerogpu_analysis(prompt) return f"### ZeroGPU Analysis ({url}):\n\n{analysis}" except Exception as e: await browser.close() return f"Visual analysis error: {str(e)}" # --------------------------------------------------------- # 4. Gradio Interface # --------------------------------------------------------- def build_gradio_ui(): with gr.Blocks(title="🎭 Playwright MCP Server on ZeroGPU") as demo: gr.Markdown( """ # 🎭 Hybrid Research MCP Server (ZeroGPU + Docker) ### 📡 External MCP SSE Connection Point MCP clients to: `https://swroop1242-pywright-mcp-server.hf.space/sse` """ ) with gr.Tab("Web Search"): q_in = gr.Textbox(label="Search Query", value="ZeroGPU Hugging Face") s_btn = gr.Button("Search Web") s_out = gr.Markdown() s_btn.click(fn=lambda q: asyncio.run(search_web(q)), inputs=q_in, outputs=s_out) with gr.Tab("ZeroGPU Research"): url_in = gr.Textbox(label="Target URL", value="https://news.ycombinator.com") query_in = gr.Textbox(label="Research Question", value="Summarize top headlines") r_btn = gr.Button("Run ZeroGPU Analysis", variant="primary") r_out = gr.Markdown() r_btn.click(fn=lambda u, q: asyncio.run(visual_chart_analysis_gpu(u, q)), inputs=[url_in, query_in], outputs=r_out) return demo # --------------------------------------------------------- # 5. Unified FastAPI Application Setup # --------------------------------------------------------- app = FastAPI() # Mount FastMCP SSE sub-application sse_subapp = mcp.sse_app() app.mount("/sse", sse_subapp) app.mount("/messages", sse_subapp) # Mount Gradio UI demo_ui = build_gradio_ui() app = gr.mount_gradio_app(app, demo_ui, path="/") if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=7860)