Spaces:
Runtime error
Runtime error
| 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 | |
| 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") | |
| 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)}" | |
| 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)}" | |
| 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)}" | |
| 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) |