diff --git "a/app.py" "b/app.py" --- "a/app.py" +++ "b/app.py" @@ -1,493 +1,56 @@ -# Purpose: One Space that offers up to seven tools/tabs (all exposed as MCP tools): -# 1) Fetch — convert webpages to clean Markdown format -# 2) Web Search — compact JSONL search output (DuckDuckGo backend) -# 3) Code Interpreter — execute Python code and capture stdout/errors -# 4) Generate Speech — synthesize speech from text using Kokoro-82M with 54 voice options -# 5) Memory Manager — lightweight JSON-based local memory store -# 6) Generate Image - HF serverless inference providers (requires HF_READ_TOKEN) -# 7) Generate Video - HF serverless inference providers (requires HF_READ_TOKEN) - from __future__ import annotations -import re +# Mother app: assemble UI from individual tool modules without changing the user-facing UI. +# Each tool exports a build_interface() function returning a gradio.Interface. + import json -import sys import os -import random -from io import StringIO -from typing import List, Dict, Tuple, Annotated, Literal, Optional - -import gradio as gr -import requests -from bs4 import BeautifulSoup -from markdownify import markdownify as md -from readability import Document -from urllib.parse import urlparse -from ddgs import DDGS -from PIL import Image -from huggingface_hub import InferenceClient -import time -import tempfile -import uuid +import sys import threading -from datetime import datetime - -# Optional imports for Kokoro TTS (loaded lazily) -import numpy as np -try: - import torch # type: ignore -except Exception: # pragma: no cover - optional dependency - torch = None # type: ignore -try: - from kokoro import KModel, KPipeline # type: ignore -except Exception: # pragma: no cover - optional dependency - KModel = None # type: ignore - KPipeline = None # type: ignore - - -# ============================== -# Fetch: Enhanced HTTP + extraction utils -# ============================== - -def _http_get_enhanced(url: str, timeout: int | float = 30, *, skip_rate_limit: bool = False) -> requests.Response: - """ - Download the page with enhanced headers, timeout handling, and better error recovery. - """ - headers = { - "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36", - "Accept-Language": "en-US,en;q=0.9", - "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8", - "Accept-Encoding": "gzip, deflate, br", - "DNT": "1", - "Connection": "keep-alive", - "Upgrade-Insecure-Requests": "1", - } - - # Apply rate limiting unless explicitly skipped - if not skip_rate_limit: - _fetch_rate_limiter.acquire() - - try: - response = requests.get( - url, - headers=headers, - timeout=timeout, # Configurable timeout - allow_redirects=True, - stream=False - ) - response.raise_for_status() - return response - except requests.exceptions.Timeout: - raise requests.exceptions.RequestException("Request timed out. The webpage took too long to respond.") - except requests.exceptions.ConnectionError: - raise requests.exceptions.RequestException("Connection error. Please check the URL and your internet connection.") - except requests.exceptions.HTTPError as e: - if response.status_code == 403: - raise requests.exceptions.RequestException("Access forbidden. The website may be blocking automated requests.") - elif response.status_code == 404: - raise requests.exceptions.RequestException("Page not found. Please check the URL.") - elif response.status_code == 429: - raise requests.exceptions.RequestException("Rate limited. Please try again in a few minutes.") - else: - raise requests.exceptions.RequestException(f"HTTP error {response.status_code}: {str(e)}") - -def _normalize_whitespace(text: str) -> str: - """ - Squeeze extra spaces and blank lines to keep things compact. - (Layman's terms: tidy up the text so it’s not full of weird spacing.) - """ - text = re.sub(r"[ \t\u00A0]+", " ", text) - text = re.sub(r"\n\s*\n\s*\n+", "\n\n", text.strip()) - return text.strip() - - -def _truncate(text: str, max_chars: int) -> Tuple[str, bool]: - """ - Cut text if it gets too long; return the text and whether we trimmed. - (Layman's terms: shorten long text and tell us if we had to cut it.) - """ - if max_chars is None or max_chars <= 0 or len(text) <= max_chars: - return text, False - return text[:max_chars].rstrip() + " …", True - - -def _shorten(text: str, limit: int) -> str: - """ - Hard cap a string with an ellipsis to keep tokens small. - (Layman's terms: force a string to a max length with an ellipsis.) - """ - if limit <= 0 or len(text) <= limit: - return text - return text[: max(0, limit - 1)].rstrip() + "…" - - -def _domain_of(url: str) -> str: - """ - Show a friendly site name like "example.com". - (Layman's terms: pull the website's domain.) - """ - try: - return urlparse(url).netloc or "" - except Exception: - return "" - - -def _meta(soup: BeautifulSoup, name: str) -> str | None: - tag = soup.find("meta", attrs={"name": name}) - return tag.get("content") if tag and tag.has_attr("content") else None - - -def _og(soup: BeautifulSoup, prop: str) -> str | None: - tag = soup.find("meta", attrs={"property": prop}) - return tag.get("content") if tag and tag.has_attr("content") else None - - -def _extract_metadata(soup: BeautifulSoup, final_url: str) -> Dict[str, str]: - """ - Pull the useful bits: title, description, site name, canonical URL, language, etc. - (Layman's terms: gather page basics like title/description/address.) - """ - meta: Dict[str, str] = {} - - # Title preference: > og:title > twitter:title - title_candidates = [ - (soup.title.string if soup.title and soup.title.string else None), - _og(soup, "og:title"), - _meta(soup, "twitter:title"), - ] - meta["title"] = next((t.strip() for t in title_candidates if t and t.strip()), "") - - # Description preference: description > og:description > twitter:description - desc_candidates = [ - _meta(soup, "description"), - _og(soup, "og:description"), - _meta(soup, "twitter:description"), - ] - meta["description"] = next((d.strip() for d in desc_candidates if d and d.strip()), "") - - # Canonical link (helps dedupe) - link_canonical = soup.find("link", rel=lambda v: v and "canonical" in v) - meta["canonical"] = (link_canonical.get("href") or "").strip() if link_canonical else "" - - # Site name + language info if present - meta["site_name"] = (_og(soup, "og:site_name") or "").strip() - html_tag = soup.find("html") - meta["lang"] = (html_tag.get("lang") or "").strip() if html_tag else "" - - # Final URL + domain - meta["fetched_url"] = final_url - meta["domain"] = _domain_of(final_url) - - return meta - - -def _extract_main_text(html: str) -> Tuple[str, BeautifulSoup]: - """ - Use Readability to isolate the main article and turn it into clean text. - Returns (clean_text, soup_of_readable_html). - (Layman's terms: find the real article text and clean it.) - """ - # Simplified article HTML from Readability - doc = Document(html) - readable_html = doc.summary(html_partial=True) - - # Parse simplified HTML - s = BeautifulSoup(readable_html, "lxml") - - # Remove noisy tags - for sel in ["script", "style", "noscript", "iframe", "svg"]: - for tag in s.select(sel): - tag.decompose() - - # Keep paragraphs, list items, and subheadings for structure without bloat - text_parts: List[str] = [] - for p in s.find_all(["p", "li", "h2", "h3", "h4", "blockquote"]): - chunk = p.get_text(" ", strip=True) - if chunk: - text_parts.append(chunk) - - clean_text = _normalize_whitespace("\n\n".join(text_parts)) - return clean_text, s - - -def _extract_links_from_soup(soup: BeautifulSoup, base_url: str) -> str: - """ - Extract all links from the page and return as formatted text. - """ - links = [] - for link in soup.find_all("a", href=True): - href = link.get("href") - text = link.get_text(strip=True) - - # Make relative URLs absolute - if href.startswith("http"): - full_url = href - elif href.startswith("//"): - full_url = "https:" + href - elif href.startswith("/"): - from urllib.parse import urljoin - full_url = urljoin(base_url, href) - else: - from urllib.parse import urljoin - full_url = urljoin(base_url, href) - - if text and href not in ["#", "javascript:void(0)"]: - links.append(f"- [{text}]({full_url})") - - if not links: - return "No links found on this page." - - # Add title if present - title = soup.find("title") - title_text = title.get_text(strip=True) if title else "Links from webpage" - - return f"# {title_text}\n\n" + "\n".join(links) - - -def _fullpage_markdown_from_soup(full_soup: BeautifulSoup, base_url: str, strip_selectors: str = "") -> str: - - # Remove custom selectors first if provided - if strip_selectors: - selectors = [s.strip() for s in strip_selectors.split(",") if s.strip()] - for selector in selectors: - try: - for element in full_soup.select(selector): - element.decompose() - except Exception: - # Invalid CSS selector, skip it - continue - - # Remove unwanted elements globally - for element in full_soup.select("script, style, nav, footer, header, aside"): - element.decompose() - - # Try common main-content containers, then fallback to body - main = ( - full_soup.find("main") - or full_soup.find("article") - or full_soup.find("div", class_=re.compile(r"content|main|post|article", re.I)) - or full_soup.find("body") - ) - - if not main: - return "No main content found on the webpage." - - # Convert selected HTML to Markdown - markdown_text = md(str(main), heading_style="ATX") - - # Clean up the markdown similar to web-scraper - markdown_text = re.sub(r"\n{3,}", "\n\n", markdown_text) - markdown_text = re.sub(r"\[\s*\]\([^)]*\)", "", markdown_text) # empty links - markdown_text = re.sub(r"[ \t]+", " ", markdown_text) - markdown_text = markdown_text.strip() - - # Add title if present - title = full_soup.find("title") - if title and title.get_text(strip=True): - markdown_text = f"# {title.get_text(strip=True)}\n\n{markdown_text}" - - return markdown_text or "No content could be extracted." - - -def _truncate_markdown(markdown: str, max_chars: int) -> Tuple[str, Dict[str, any]]: - """ - Truncate markdown content to a maximum character count while preserving structure. - Tries to break at paragraph boundaries when possible. - - Returns: - Tuple[str, Dict]: (truncated_content, metadata_dict) - metadata_dict contains: truncated, returned_chars, total_chars_estimate, next_cursor - """ - total_chars = len(markdown) - - if total_chars <= max_chars: - return markdown, { - "truncated": False, - "returned_chars": total_chars, - "total_chars_estimate": total_chars, - "next_cursor": None - } - - # Find a good break point near the limit - truncated = markdown[:max_chars] - - # Try to break at the end of a paragraph (double newline) - last_paragraph = truncated.rfind('\n\n') - if last_paragraph > max_chars * 0.7: # If we find a paragraph break in the last 30% - truncated = truncated[:last_paragraph] - cursor_pos = last_paragraph - # Try to break at the end of a sentence - elif '.' in truncated[-100:]: # Look for a period in the last 100 chars - last_period = truncated.rfind('.') - if last_period > max_chars * 0.8: # If we find a period in the last 20% - truncated = truncated[:last_period + 1] - cursor_pos = last_period + 1 - else: - cursor_pos = len(truncated) - else: - cursor_pos = len(truncated) - - metadata = { - "truncated": True, - "returned_chars": len(truncated), - "total_chars_estimate": total_chars, - "next_cursor": cursor_pos - } - - truncated = truncated.rstrip() - - # Add informative truncation notice - truncation_notice = ( - f"\n\n---\n" - f"**Content Truncated:** Showing {metadata['returned_chars']:,} of {metadata['total_chars_estimate']:,} characters " - f"({(metadata['returned_chars']/metadata['total_chars_estimate']*100):.1f}%)\n" - f"**Next cursor:** {metadata['next_cursor']} (use this value with offset parameter for continuation)\n" - f"---" - ) - - return truncated + truncation_notice, metadata - - -def Web_Fetch( # <-- MCP tool #1 (Fetch) - url: Annotated[str, "The absolute URL to fetch (must return HTML)."], - max_chars: Annotated[int, "Maximum characters to return (0 = no limit, full page content)."] = 3000, - strip_selectors: Annotated[str, "CSS selectors to remove (comma-separated, e.g., '.header, .footer, nav')."] = "", - url_scraper: Annotated[bool, "Extract only links from the page instead of content."] = False, - offset: Annotated[int, "Character offset to start from (for pagination, use next_cursor from previous call)."] = 0, -) -> str: - """ - Fetch a web page and return it converted to Markdown format with configurable options. - - This function retrieves a webpage and either converts its main content to clean Markdown - or extracts all links from the page. It automatically removes navigation, footers, - scripts, and other non-content elements, plus any custom selectors you specify. - - Args: - url (str): The absolute URL to fetch (must return HTML). - max_chars (int): Maximum characters to return. Use 0 for no limit (full page). - strip_selectors (str): CSS selectors to remove before processing (comma-separated). - url_scraper (bool): If True, extract only links instead of content. - offset (int): Character offset to start from (for pagination, use next_cursor from previous call). - - Returns: - str: Either the webpage content converted to Markdown or a list of all links, - depending on the url_scraper setting. Content is length-limited by max_chars - and includes detailed truncation metadata when content is truncated. - """ - _log_call_start("Web_Fetch", url=url, max_chars=max_chars, strip_selectors=strip_selectors, url_scraper=url_scraper, offset=offset) - if not url or not url.strip(): - result = "Please enter a valid URL." - _log_call_end("Web_Fetch", _truncate_for_log(result)) - return result - - try: - resp = _http_get_enhanced(url) - resp.raise_for_status() - except requests.exceptions.RequestException as e: - result = f"An error occurred: {e}" - _log_call_end("Web_Fetch", _truncate_for_log(result)) - return result - - final_url = str(resp.url) - ctype = resp.headers.get("Content-Type", "") - if "html" not in ctype.lower(): - result = f"Unsupported content type for extraction: {ctype or 'unknown'}" - _log_call_end("Web_Fetch", _truncate_for_log(result)) - return result - - # Decode to text - resp.encoding = resp.encoding or resp.apparent_encoding - html = resp.text - - # Parse HTML - full_soup = BeautifulSoup(html, "lxml") - - if url_scraper: - # Extract links mode - result = _extract_links_from_soup(full_soup, final_url) - # Apply offset and truncation for link extraction too - if offset > 0: - result = result[offset:] - if max_chars > 0 and len(result) > max_chars: - result, metadata = _truncate_markdown(result, max_chars) - else: - # Convert to markdown mode - full_result = _fullpage_markdown_from_soup(full_soup, final_url, strip_selectors) - - # Apply offset if specified - if offset > 0: - if offset >= len(full_result): - result = f"Offset {offset} exceeds content length ({len(full_result)} characters). Content ends at position {len(full_result)}." - _log_call_end("Web_Fetch", _truncate_for_log(result)) - return result - result = full_result[offset:] - else: - result = full_result - - # Apply max_chars truncation if specified - if max_chars > 0 and len(result) > max_chars: - result, metadata = _truncate_markdown(result, max_chars) - # Adjust metadata to account for offset - if offset > 0: - metadata["total_chars_estimate"] = len(full_result) - metadata["next_cursor"] = offset + metadata["next_cursor"] if metadata["next_cursor"] else None - - _log_call_end("Web_Fetch", f"chars={len(result)}, url_scraper={url_scraper}, offset={offset}") - return result - +import time +from datetime import datetime, timedelta +from typing import Any -# ============================================ -# Web Search (DuckDuckGo backend): Enhanced with error handling & rate limiting -# ============================================ +import gradio as gr -import asyncio -from datetime import datetime, timedelta class RateLimiter: - def __init__(self, requests_per_minute: int = 30): + """Best-effort in-process rate limiter for HTTP-heavy tools.""" + + def __init__(self, requests_per_minute: int = 30) -> None: self.requests_per_minute = requests_per_minute - self.requests = [] + self._requests: list[datetime] = [] + self._lock = threading.Lock() - def acquire(self): - """Synchronous rate limiting for non-async context""" + def acquire(self) -> None: now = datetime.now() - # Remove requests older than 1 minute - self.requests = [ - req for req in self.requests if now - req < timedelta(minutes=1) - ] + with self._lock: + self._requests = [req for req in self._requests if now - req < timedelta(minutes=1)] + if len(self._requests) >= self.requests_per_minute: + wait_time = 60 - (now - self._requests[0]).total_seconds() + if wait_time > 0: + time.sleep(max(1, wait_time)) + self._requests.append(now) - if len(self.requests) >= self.requests_per_minute: - # Wait until we can make another request - wait_time = 60 - (now - self.requests[0]).total_seconds() - if wait_time > 0: - time.sleep(max(1, wait_time)) # At least 1 second wait - - self.requests.append(now) -# Global rate limiters _search_rate_limiter = RateLimiter(requests_per_minute=20) _fetch_rate_limiter = RateLimiter(requests_per_minute=25) -# ============================== -# Logging Helpers (print I/O to terminal) -# ============================== def _truncate_for_log(value: str, limit: int = 500) -> str: - """Truncate long strings for concise terminal logging.""" if len(value) <= limit: return value - return value[:limit - 1] + "…" + return value[: limit - 1] + "…" -def _serialize_input(val): # type: ignore[return-any] - """Best-effort compact serialization of arbitrary input values for logging.""" +def _serialize_input(val: Any) -> Any: try: if isinstance(val, (str, int, float, bool)) or val is None: return val if isinstance(val, (list, tuple)): - return [_serialize_input(v) for v in list(val)[:10]] + (["…"] if len(val) > 10 else []) # type: ignore[index] + return [_serialize_input(v) for v in list(val)[:10]] + (["…"] if len(val) > 10 else []) if isinstance(val, dict): - out = {} + out: dict[str, Any] = {} for i, (k, v) in enumerate(val.items()): if i >= 12: out["…"] = "…" @@ -499,1070 +62,40 @@ def _serialize_input(val): # type: ignore[return-any] return "<unserializable>" -def _log_call_start(func_name: str, **kwargs) -> None: +def _log_call_start(func_name: str, **kwargs: Any) -> None: try: compact = {k: _serialize_input(v) for k, v in kwargs.items()} print(f"[TOOL CALL] {func_name} inputs: {json.dumps(compact, ensure_ascii=False)[:800]}", flush=True) - except Exception as e: # pragma: no cover - logging safety - print(f"[TOOL CALL] {func_name} (failed to log inputs: {e})", flush=True) + except Exception as exc: + print(f"[TOOL CALL] {func_name} (failed to log inputs: {exc})", flush=True) def _log_call_end(func_name: str, output_desc: str) -> None: try: print(f"[TOOL RESULT] {func_name} output: {output_desc}", flush=True) - except Exception as e: # pragma: no cover - print(f"[TOOL RESULT] {func_name} (failed to log output: {e})", flush=True) - - -# ============================== -# Deep Research helpers: slow-host detection -# ============================== - -class SlowHost(Exception): - """Marker exception for slow hosts (timeouts) to trigger requeue.""" - pass - - -def _fetch_page_markdown_fast(url: str, max_chars: int = 3000, timeout: float = 10.0) -> str: - """Fetch a single URL quickly; raise SlowHost on timeout. - - Uses a shorter HTTP timeout to detect slow hosts, then reuses Web_Fetch - logic for conversion to Markdown. Returns empty string on non-timeout errors. - """ - try: - # Bypass global rate limiter here; we want Deep Research to control pacing. - resp = _http_get_enhanced(url, timeout=timeout, skip_rate_limit=True) - resp.raise_for_status() - except requests.exceptions.RequestException as e: - msg = str(e) - if "timed out" in msg.lower(): - raise SlowHost(msg) - return "" - - final_url = str(resp.url) - ctype = resp.headers.get("Content-Type", "") - if "html" not in ctype.lower(): - return "" - - # Decode to text and convert similar to Web_Fetch (lean path) - resp.encoding = resp.encoding or resp.apparent_encoding - html = resp.text - soup = BeautifulSoup(html, "lxml") - # Reuse fullpage conversion with default selectors - md_text = _fullpage_markdown_from_soup(soup, final_url, "") - if max_chars > 0 and len(md_text) > max_chars: - md_text, _ = _truncate_markdown(md_text, max_chars) - return md_text - -def _extract_date_from_snippet(snippet: str) -> str: - """ - Extract publication date from search result snippet using common patterns. - """ - import re - from datetime import datetime - - if not snippet: - return "" - - # Common date patterns - date_patterns = [ - # ISO format: 2023-12-25, 2023/12/25 - r'\b(\d{4}[-/]\d{1,2}[-/]\d{1,2})\b', - # US format: Dec 25, 2023 | December 25, 2023 - r'\b([A-Za-z]{3,9}\s+\d{1,2},?\s+\d{4})\b', - # EU format: 25 Dec 2023 | 25 December 2023 - r'\b(\d{1,2}\s+[A-Za-z]{3,9}\s+\d{4})\b', - # Relative: "2 days ago", "1 week ago", "3 months ago" - r'\b(\d+\s+(?:day|week|month|year)s?\s+ago)\b', - # Common prefixes: "Published: ", "Updated: ", "Posted: " - r'(?:Published|Updated|Posted):\s*([^,\n]+?)(?:[,\n]|$)', - ] - - for pattern in date_patterns: - matches = re.findall(pattern, snippet, re.IGNORECASE) - if matches: - return matches[0].strip() - - return "" - - -def _format_search_result(result: dict, search_type: str, index: int) -> list[str]: - """ - Format a single search result based on the search type. - Returns a list of strings to be joined with newlines. - """ - lines = [] - - if search_type == "text": - title = result.get("title", "").strip() - url = result.get("href", "").strip() - snippet = result.get("body", "").strip() - date = _extract_date_from_snippet(snippet) - - lines.append(f"{index}. {title}") - lines.append(f" URL: {url}") - if snippet: - lines.append(f" Summary: {snippet}") - if date: - lines.append(f" Date: {date}") - - elif search_type == "news": - title = result.get("title", "").strip() - url = result.get("url", "").strip() - body = result.get("body", "").strip() - date = result.get("date", "").strip() - source = result.get("source", "").strip() - - lines.append(f"{index}. {title}") - lines.append(f" URL: {url}") - if source: - lines.append(f" Source: {source}") - if date: - lines.append(f" Date: {date}") - if body: - lines.append(f" Summary: {body}") - - elif search_type == "images": - title = result.get("title", "").strip() - image_url = result.get("image", "").strip() - source_url = result.get("url", "").strip() - source = result.get("source", "").strip() - width = result.get("width", "") - height = result.get("height", "") - - lines.append(f"{index}. {title}") - lines.append(f" Image: {image_url}") - lines.append(f" Source: {source_url}") - if source: - lines.append(f" Publisher: {source}") - if width and height: - lines.append(f" Dimensions: {width}x{height}") - - elif search_type == "videos": - title = result.get("title", "").strip() - description = result.get("description", "").strip() - duration = result.get("duration", "").strip() - published = result.get("published", "").strip() - uploader = result.get("uploader", "").strip() - embed_url = result.get("embed_url", "").strip() - - lines.append(f"{index}. {title}") - if embed_url: - lines.append(f" Video: {embed_url}") - if uploader: - lines.append(f" Uploader: {uploader}") - if duration: - lines.append(f" Duration: {duration}") - if published: - lines.append(f" Published: {published}") - if description: - lines.append(f" Description: {description}") - - elif search_type == "books": - title = result.get("title", "").strip() - url = result.get("url", "").strip() - body = result.get("body", "").strip() - - lines.append(f"{index}. {title}") - lines.append(f" URL: {url}") - if body: - lines.append(f" Description: {body}") - - return lines - - -def Web_Search( # <-- MCP tool #2 (Web Search) - query: Annotated[str, "The search query (supports operators like site:, quotes, OR)."], - max_results: Annotated[int, "Number of results to return (1–20)."] = 5, - page: Annotated[int, "Page number for pagination (1-based, each page contains max_results items)."] = 1, - search_type: Annotated[str, "Type of search: 'text' (web pages), 'news', 'images', 'videos', or 'books'."] = "text", - offset: Annotated[int, "Result offset to start from (overrides page if > 0, for precise continuation)."] = 0, -) -> str: - """ - Run a web search (DuckDuckGo backend) and return formatted results with support for multiple content types. - - Features smart fallback: if 'news' search returns no results, automatically retries with 'text' - search to catch sources like Hacker News that might not appear in news-specific results. - - Args: - query (str): The search query string. Supports operators like site:, quotes for exact matching, - OR for alternatives, and other DuckDuckGo search syntax. - Examples: - - Basic search: "Python programming" - - Site search: "site:example.com" - - Exact phrase: "artificial intelligence" - - Exclude terms: "cats -dogs" - max_results (int): Number of results to return per page (1–20). Default: 5. - page (int): Page number for pagination (1-based). Default: 1. Ignored if offset > 0. - search_type (str): Type of search to perform: - - "text": Web pages (default) - - "news": News articles with dates and sources (with smart fallback to 'text') - - "images": Image results with dimensions and sources - - "videos": Video results with duration and upload info - - "books": Book search results - offset (int): Result offset to start from (0-based). If > 0, overrides page parameter - for precise continuation. Use this to pick up exactly where you left off. - - Returns: - str: Search results formatted appropriately for the search type, with pagination info. - If 'news' search fails, results include a note about automatic fallback to 'text' search. - Includes next_offset information for easy continuation. - """ - _log_call_start("Web_Search", query=query, max_results=max_results, page=page, search_type=search_type, offset=offset) - if not query or not query.strip(): - result = "No search query provided. Please enter a search term." - _log_call_end("Web_Search", _truncate_for_log(result)) - return result - - # Validate parameters - max_results = max(1, min(20, max_results)) - page = max(1, page) - offset = max(0, offset) - valid_types = ["text", "news", "images", "videos", "books"] - if search_type not in valid_types: - search_type = "text" - - # Calculate actual offset: use provided offset if > 0, otherwise calculate from page - if offset > 0: - actual_offset = offset - calculated_page = (offset // max_results) + 1 - else: - actual_offset = (page - 1) * max_results - calculated_page = page - - total_needed = actual_offset + max_results - - # Track if we used fallback - used_fallback = False - original_search_type = search_type - - def _perform_search(stype: str): - """Perform the actual search with the given search type.""" - try: - # Apply rate limiting to avoid being blocked - _search_rate_limiter.acquire() - - # Perform search with timeout handling based on search type - with DDGS() as ddgs: - if stype == "text": - raw_gen = ddgs.text(query, max_results=total_needed + 10) - elif stype == "news": - raw_gen = ddgs.news(query, max_results=total_needed + 10) - elif stype == "images": - raw_gen = ddgs.images(query, max_results=total_needed + 10) - elif stype == "videos": - raw_gen = ddgs.videos(query, max_results=total_needed + 10) - elif stype == "books": - raw_gen = ddgs.books(query, max_results=total_needed + 10) - - # Convert generator to list, handle case where no results are found - try: - return list(raw_gen) - except Exception as inner_e: - # If the generator fails (e.g., no results), return empty list - if "no results" in str(inner_e).lower() or "not found" in str(inner_e).lower(): - return [] - else: - raise inner_e - - except Exception as e: - error_msg = f"Search failed: {str(e)[:200]}" - if "blocked" in str(e).lower() or "rate" in str(e).lower(): - error_msg = "Search temporarily blocked due to rate limiting. Please try again in a few minutes." - elif "timeout" in str(e).lower(): - error_msg = "Search timed out. Please try again with a simpler query." - elif "network" in str(e).lower() or "connection" in str(e).lower(): - error_msg = "Network connection error. Please check your internet connection and try again." - elif "no results" in str(e).lower() or "not found" in str(e).lower(): - # This is expected for some searches, return empty list - return [] - raise Exception(error_msg) - - # Try the primary search - try: - raw = _perform_search(search_type) - except Exception as e: - result = f"Error: {str(e)}" - _log_call_end("Web_Search", _truncate_for_log(result)) - return result - - # Smart fallback: if news search returns empty and we haven't tried text yet, try text search - if not raw and search_type == "news": - try: - raw = _perform_search("text") - if raw: # Only mark as fallback if we actually got results - used_fallback = True - search_type = "text" # Update for result formatting - except Exception: - # If fallback also fails, continue with empty results from original search - pass - - if not raw: - fallback_note = " (also tried 'text' search as fallback)" if original_search_type == "news" and used_fallback else "" - result = f"No {original_search_type} results found for query: {query}{fallback_note}" - _log_call_end("Web_Search", _truncate_for_log(result)) - return result - - # Apply pagination by slicing the results - paginated_results = raw[actual_offset:actual_offset + max_results] - - if not paginated_results: - if actual_offset >= len(raw): - result = f"Offset {actual_offset} exceeds available results ({len(raw)} total). Try offset=0 to start from beginning." - else: - result = f"No {original_search_type} results found on page {calculated_page} for query: {query}. Try page 1 or reduce page number." - _log_call_end("Web_Search", _truncate_for_log(result)) - return result - - # Format results based on search type - total_available = len(raw) - start_num = actual_offset + 1 - end_num = actual_offset + len(paginated_results) - next_offset = actual_offset + len(paginated_results) - - # Create header with fallback notification if applicable - search_label = original_search_type.title() - if used_fallback: - search_label += " → Text (Smart Fallback)" - - # Show both page and offset information for clarity - pagination_info = f"Page {calculated_page}" - if offset > 0: - pagination_info = f"Offset {actual_offset} (≈ {pagination_info})" - - lines = [f"{search_label} search results for: {query}"] - - if used_fallback: - lines.append("📍 Note: News search returned no results, automatically searched general web content instead") - - lines.append(f"{pagination_info} (results {start_num}-{end_num} of ~{total_available}+ available)\n") - - for i, result in enumerate(paginated_results, start_num): - result_lines = _format_search_result(result, search_type, i) - lines.extend(result_lines) - lines.append("") # Empty line between results - - # Add pagination/continuation hints - if total_available > end_num: - lines.append(f"💡 More results available:") - lines.append(f" • Next page: page={calculated_page + 1}") - lines.append(f" • Next offset: offset={next_offset}") - lines.append(f" • Use offset={next_offset} to continue exactly from result {next_offset + 1}") - - result = "\n".join(lines) - search_info = f"type={original_search_type}" - if used_fallback: - search_info += "→text" - _log_call_end("Web_Search", f"{search_info} page={calculated_page} offset={actual_offset} results={len(paginated_results)} chars={len(result)}") - return result - - -# ====================================== -# Code Execution: Python (MCP tool #3) -# ====================================== - -def Code_Interpreter(code: Annotated[str, "Python source code to run; stdout is captured and returned."]) -> str: - """ - Execute arbitrary Python code and return captured stdout or an error message. - - Args: - code (str): Python source code to run; stdout is captured and returned. - - Returns: - str: Combined stdout produced by the code, or the exception text if - execution failed. - """ - _log_call_start("Code_Interpreter", code=_truncate_for_log(code or "", 300)) - if code is None: - result = "No code provided." - _log_call_end("Code_Interpreter", result) - return result - - old_stdout = sys.stdout - redirected_output = sys.stdout = StringIO() - try: - exec(code) - result = redirected_output.getvalue() - except Exception as e: - result = str(e) - finally: - sys.stdout = old_stdout - _log_call_end("Code_Interpreter", _truncate_for_log(result)) - return result - - -# ========================== -# Generate Speech (MCP tool #4) -# ========================== - -_KOKORO_STATE = { - "initialized": False, - "device": "cpu", - "model": None, - "pipelines": {}, -} - - -def get_kokoro_voices(): - """Get comprehensive list of available Kokoro voice IDs (54 total).""" - try: - from huggingface_hub import list_repo_files - # Get voice files from the Kokoro repository - files = list_repo_files('hexgrad/Kokoro-82M') - voice_files = [f for f in files if f.endswith('.pt') and f.startswith('voices/')] - voices = [f.replace('voices/', '').replace('.pt', '') for f in voice_files] - return sorted(voices) if voices else _get_fallback_voices() - except Exception: - return _get_fallback_voices() - - -def _get_fallback_voices(): - """Return comprehensive fallback list of known Kokoro voices (54 total).""" - return [ - # American Female (11 voices) - "af_alloy", "af_aoede", "af_bella", "af_heart", "af_jessica", - "af_kore", "af_nicole", "af_nova", "af_river", "af_sarah", "af_sky", - # American Male (9 voices) - "am_adam", "am_echo", "am_eric", "am_fenrir", "am_liam", - "am_michael", "am_onyx", "am_puck", "am_santa", - # British Female (4 voices) - "bf_alice", "bf_emma", "bf_isabella", "bf_lily", - # British Male (4 voices) - "bm_daniel", "bm_fable", "bm_george", "bm_lewis", - # European Female/Male (3 voices) - "ef_dora", "em_alex", "em_santa", - # French Female (1 voice) - "ff_siwis", - # Hindi Female/Male (4 voices) - "hf_alpha", "hf_beta", "hm_omega", "hm_psi", - # Italian Female/Male (2 voices) - "if_sara", "im_nicola", - # Japanese Female/Male (5 voices) - "jf_alpha", "jf_gongitsune", "jf_nezumi", "jf_tebukuro", "jm_kumo", - # Portuguese Female/Male (3 voices) - "pf_dora", "pm_alex", "pm_santa", - # Chinese Female/Male (8 voices) - "zf_xiaobei", "zf_xiaoni", "zf_xiaoxiao", "zf_xiaoyi", - "zm_yunjian", "zm_yunxi", "zm_yunxia", "zm_yunyang" - ] - - -def _init_kokoro() -> None: - """Lazy-initialize Kokoro model and pipelines on first use. - - Tries CUDA if torch is present and available; falls back to CPU. Keeps a - minimal English pipeline and custom lexicon tweak for the word "kokoro". - """ - if _KOKORO_STATE["initialized"]: - return - - if KModel is None or KPipeline is None: - raise RuntimeError( - "Kokoro is not installed. Please install the 'kokoro' package (>=0.9.4)." - ) - - device = "cpu" - if torch is not None: - try: - if torch.cuda.is_available(): # type: ignore[attr-defined] - device = "cuda" - except Exception: - device = "cpu" - - model = KModel().to(device).eval() - pipelines = {"a": KPipeline(lang_code="a", model=False)} - # Custom pronunciation - try: - pipelines["a"].g2p.lexicon.golds["kokoro"] = "kˈOkəɹO" - except Exception: - pass - - _KOKORO_STATE.update( - { - "initialized": True, - "device": device, - "model": model, - "pipelines": pipelines, - } - ) - - -def List_Kokoro_Voices() -> List[str]: - """ - Get a list of all available Kokoro voice identifiers. - - This MCP tool helps clients discover the 54 available voice options - for the Generate_Speech tool. - - Returns: - List[str]: A list of voice identifiers (e.g., ["af_heart", "am_adam", "bf_alice", ...]) - - Voice naming convention: - - First 2 letters: Language/Region (af=American Female, am=American Male, bf=British Female, etc.) - - Following letters: Voice name (heart, adam, alice, etc.) - - Available categories: - - American Female/Male (20 voices) - - British Female/Male (8 voices) - - European Female/Male (3 voices) - - French Female (1 voice) - - Hindi Female/Male (4 voices) - - Italian Female/Male (2 voices) - - Japanese Female/Male (5 voices) - - Portuguese Female/Male (3 voices) - - Chinese Female/Male (8 voices) - """ - return get_kokoro_voices() - - -def Generate_Speech( # <-- MCP tool #4 (Generate Speech) - text: Annotated[str, "The text to synthesize (English)."], - speed: Annotated[float, "Speech speed multiplier in 0.5–2.0; 1.0 = normal speed."] = 1.25, - voice: Annotated[str, "Voice identifier from 54 available options."] = "af_heart", -) -> Tuple[int, np.ndarray]: - """ - Synthesize speech from text using the Kokoro-82M TTS model. - - This function returns raw audio suitable for a Gradio Audio component and is - also exposed as an MCP tool. It supports 54 different voices across multiple - languages and accents including American, British, European, Hindi, Italian, - Japanese, Portuguese, and Chinese speakers. - - Args: - text (str): The text to synthesize. Works best with English but supports multiple languages. - speed (float): Speech speed multiplier in 0.5–2.0; 1.0 = normal speed. Default: 1.25 (slightly brisk). - voice (str): Voice identifier from 54 available options. Default: 'af_heart'. - - Returns: - A tuple of (sample_rate_hz, audio_waveform) where: - - sample_rate_hz: int sample rate in Hz (24_000) - - audio_waveform: numpy.ndarray float32 mono waveform in range [-1, 1] - """ - _log_call_start("Generate_Speech", text=_truncate_for_log(text, 200), speed=speed, voice=voice) - if not text or not text.strip(): - try: - _log_call_end("Generate_Speech", "error=empty text") - finally: - pass - raise gr.Error("Please provide non-empty text to synthesize.") - - _init_kokoro() - model = _KOKORO_STATE["model"] - pipelines = _KOKORO_STATE["pipelines"] - - pipeline = pipelines.get("a") - if pipeline is None: - raise gr.Error("Kokoro English pipeline not initialized.") - - # Process ALL segments for longer audio generation - audio_segments = [] - pack = pipeline.load_voice(voice) - - try: - # Get all segments first to show progress for long text - segments = list(pipeline(text, voice, speed)) - total_segments = len(segments) - - # Iterate through ALL segments instead of just the first one - for segment_idx, (text_chunk, ps, _) in enumerate(segments): - ref_s = pack[len(ps) - 1] - try: - audio = model(ps, ref_s, float(speed)) - audio_segments.append(audio.detach().cpu().numpy()) - - # For very long text (>10 segments), show progress every few segments - if total_segments > 10 and (segment_idx + 1) % 5 == 0: - print(f"Progress: Generated {segment_idx + 1}/{total_segments} segments...") - - except Exception as e: - raise gr.Error(f"Error generating audio for segment {segment_idx + 1}: {str(e)}") - - if not audio_segments: - raise gr.Error("No audio was generated (empty synthesis result).") - - # Concatenate all segments to create the complete audio - if len(audio_segments) == 1: - final_audio = audio_segments[0] - else: - final_audio = np.concatenate(audio_segments, axis=0) - # For multi-segment audio, provide completion info - duration = len(final_audio) / 24_000 - if total_segments > 1: - print(f"Completed: {total_segments} segments concatenated into {duration:.1f} seconds of audio") - - # Success logging & return - _log_call_end("Generate_Speech", f"samples={final_audio.shape[0]} duration_sec={len(final_audio)/24_000:.2f}") - return 24_000, final_audio - - except gr.Error as e: - _log_call_end("Generate_Speech", f"gr_error={str(e)}") - raise # Re-raise - except Exception as e: - _log_call_end("Generate_Speech", f"error={str(e)[:120]}") - raise gr.Error(f"Error during speech generation: {str(e)}") - - -# ========================== -# JSON Memory System (MCP tools #7–#10 if enabled) -# ========================== - -# Implementation goals (aligned with Gradio MCP docs): -# * Each function has a rich docstring (used for tool description) -# * Type hints + Annotated param docs become the schema -# * Zero external dependencies (pure stdlib JSON file persistence) -# * Safe concurrent access via a process‑local lock -# * Human‑readable & recoverable even if file becomes corrupted - -MEMORY_FILE = os.path.join(os.path.dirname(__file__), "memories.json") -_MEMORY_LOCK = threading.RLock() -_MAX_MEMORIES = 10_000 # soft cap to avoid unbounded growth - - -def _now_iso() -> str: - return datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S") - - -def _load_memories() -> List[Dict[str, str]]: - """Internal helper: load memory list from disk. - - Returns an empty list if the file does not exist or is unreadable. - If the JSON is corrupted, a *.corrupt backup is written once and a - fresh empty list is returned (fail‑open philosophy for tool usage). - """ - if not os.path.exists(MEMORY_FILE): - return [] - try: - with open(MEMORY_FILE, "r", encoding="utf-8") as f: - data = json.load(f) - if isinstance(data, list): - # Filter only dict items containing required keys if present - cleaned: List[Dict[str, str]] = [] - for item in data: - if isinstance(item, dict) and "id" in item and "text" in item: - cleaned.append(item) - return cleaned - return [] - except Exception: - # Backup corrupted file once - try: - backup = MEMORY_FILE + ".corrupt" - if not os.path.exists(backup): - os.replace(MEMORY_FILE, backup) - except Exception: - pass - return [] - - -def _save_memories(memories: List[Dict[str, str]]) -> None: - """Persist memory list atomically to disk (write temp then replace).""" - tmp_path = MEMORY_FILE + ".tmp" - with open(tmp_path, "w", encoding="utf-8") as f: - json.dump(memories, f, ensure_ascii=False, indent=2) - os.replace(tmp_path, MEMORY_FILE) - - -def _mem_save( - text: Annotated[str, "Raw textual content to remember (will be stored verbatim)."], - tags: Annotated[str, "Optional comma-separated tags for lightweight categorization (e.g. 'user, preference')."] = "", -) -> str: - """(Internal) Persist a new memory record. - - Summary: - Adds a memory object to the local JSON store (no external database). - - Stored Fields: - - id (str, UUID4) - - text (str, verbatim user content) - - timestamp (UTC "YYYY-MM-DD HH:MM:SS") - - tags (str, original comma-separated tag string) - - Behavior / Rules: - 1. Whitespace is trimmed; empty text is rejected. - 2. If the most recent existing memory has identical text, the new one is skipped (light dedupe heuristic). - 3. When total entries exceed _MAX_MEMORIES, oldest entries are pruned (soft cap). - 4. Operation is protected by an in‑process reentrant lock only (no cross‑process locking). - - Returns: - str: Human readable confirmation containing the new memory UUID (full or prefix - - Security / Privacy: - Data is plaintext JSON on local disk; do NOT store secrets or regulated data. - """ - text_clean = (text or "").strip() - if not text_clean: - return "Error: memory text is empty." - - with _MEMORY_LOCK: - memories = _load_memories() - if memories and memories[-1].get("text") == text_clean: - return "Skipped: identical to last stored memory." - - mem_id = str(uuid.uuid4()) - entry = { - "id": mem_id, - "text": text_clean, - "timestamp": _now_iso(), - "tags": tags.strip(), - } - memories.append(entry) - if len(memories) > _MAX_MEMORIES: - # Drop oldest overflow - overflow = len(memories) - _MAX_MEMORIES - memories = memories[overflow:] - _save_memories(memories) - return f"Memory saved: {mem_id}" - - -def _mem_list( - limit: Annotated[int, "Maximum number of most recent memories to return (1–200)."] = 20, - include_tags: Annotated[bool, "If true, include tags column in output."] = True, -) -> str: - """(Internal) List most recent memories. - - Parameters: - limit (int): Max rows to return; clamped to [1, 200]. - include_tags (bool): Include tags section when True. - - Output Format (one per line): - <uuid_prefix> [YYYY-MM-DD HH:MM:SS] <text> | tags: <tag list> - (Tag column omitted if empty or include_tags=False.) - - Returns: - str: Joined newline string or a friendly "No memories stored." message. - """ - limit = max(1, min(200, limit)) - with _MEMORY_LOCK: - memories = _load_memories() - if not memories: - return "No memories stored yet." - # Already chronological (append order); display newest first - chosen = memories[-limit:][::-1] - lines: List[str] = [] - for m in chosen: - base = f"{m['id'][:8]} [{m.get('timestamp','?')}] {m.get('text','')}" - if include_tags and m.get("tags"): - base += f" | tags: {m['tags']}" - lines.append(base) - omitted = len(memories) - len(chosen) - if omitted > 0: - lines.append(f"… ({omitted} older memorie{'s' if omitted!=1 else ''} omitted; total={len(memories)})") - return "\n".join(lines) - - -def _parse_search_query(query: str) -> Dict[str, List[str]]: - """Parse a search query into structured components. - - Supports: - - tag:name - search for specific tag - - AND/OR operators (case-insensitive) - - Regular text terms - - Implicit AND between terms when no operator specified - - Examples: - 'tag:work' -> {'tag_terms': ['work'], 'text_terms': [], 'operator': 'and'} - 'tag:work AND tag:project' -> {'tag_terms': ['work', 'project'], 'text_terms': [], 'operator': 'and'} - 'tag:personal OR tag:todo' -> {'tag_terms': ['personal', 'todo'], 'text_terms': [], 'operator': 'or'} - 'meeting tag:work' -> {'tag_terms': ['work'], 'text_terms': ['meeting'], 'operator': 'and'} - 'tag:urgent OR important' -> {'tag_terms': ['urgent'], 'text_terms': ['important'], 'operator': 'or'} - - Returns: - Dict with keys: 'tag_terms', 'text_terms', 'operator' (and/or) - """ - import re - - # Initialize result - result = { - 'tag_terms': [], - 'text_terms': [], - 'operator': 'and' # default - } - - if not query or not query.strip(): - return result - - # Normalize whitespace and detect OR operator - query = re.sub(r'\s+', ' ', query.strip()) - if re.search(r'\bOR\b', query, re.IGNORECASE): - result['operator'] = 'or' - # Split on OR (case-insensitive) - parts = re.split(r'\s+OR\s+', query, flags=re.IGNORECASE) - else: - # Split on AND (case-insensitive) or just whitespace - parts = re.split(r'\s+(?:AND\s+)?', query, flags=re.IGNORECASE) - # Remove empty AND tokens that might have been left - parts = [p for p in parts if p.strip() and p.strip().upper() != 'AND'] - - # Process each part - for part in parts: - part = part.strip() - if not part: - continue - - # Check if it's a tag query - tag_match = re.match(r'^tag:(.+)$', part, re.IGNORECASE) - if tag_match: - tag_name = tag_match.group(1).strip() - if tag_name: - result['tag_terms'].append(tag_name.lower()) - else: - # Regular text term - result['text_terms'].append(part.lower()) - - return result - - -def _match_memory_with_query(memory: Dict[str, str], parsed_query: Dict[str, List[str]]) -> bool: - """Check if a memory matches the parsed search query.""" - tag_terms = parsed_query['tag_terms'] - text_terms = parsed_query['text_terms'] - operator = parsed_query['operator'] - - # If no terms, no match - if not tag_terms and not text_terms: - return False - - # Get memory content (case-insensitive) - memory_text = memory.get('text', '').lower() - memory_tags = memory.get('tags', '').lower() - - # Split memory tags into individual tags - memory_tag_list = [tag.strip() for tag in memory_tags.split(',') if tag.strip()] - - # Check tag matches - tag_matches = [] - for tag_term in tag_terms: - # Check if tag_term matches any of the memory's tags - tag_matches.append(any(tag_term in tag for tag in memory_tag_list)) - - # Check text matches - text_matches = [] - combined_text = memory_text + ' ' + memory_tags # For backward compatibility - for text_term in text_terms: - text_matches.append(text_term in combined_text) - - # Combine all matches - all_matches = tag_matches + text_matches - - if not all_matches: - return False - - # Apply operator logic - if operator == 'or': - return any(all_matches) - else: # 'and' - return all(all_matches) - - -def _mem_search( - query: Annotated[str, "Advanced search with tag:name syntax, AND/OR operators, and text terms."], - limit: Annotated[int, "Maximum number of matches (1–200)."] = 20, -) -> str: - """(Internal) Enhanced search with tag queries and boolean operators. - - Search Syntax: - - tag:name - search for specific tag - - AND/OR operators (case-insensitive, default is AND) - - Regular text terms search in text content and tags - - Examples: - * 'tag:work' - memories with 'work' tag - * 'tag:work AND tag:project' - memories with both tags - * 'tag:personal OR tag:todo' - memories with either tag - * 'meeting tag:work' - memories with "meeting" in text and 'work' tag - * 'tag:urgent OR important' - memories with 'urgent' tag OR "important" anywhere - - Parameters: - query (str): Enhanced query string with tag: syntax and AND/OR operators. - limit (int): Max rows to return; clamped to [1, 200]. - - Returns: - str: Formatted lines identical to _mem_list output or "No matches". - """ - q = (query or "").strip() - if not q: - return "Error: empty query." - - # Parse the enhanced query - parsed_query = _parse_search_query(q) - if not parsed_query['tag_terms'] and not parsed_query['text_terms']: - return "Error: no valid search terms found." - - limit = max(1, min(200, limit)) - with _MEMORY_LOCK: - memories = _load_memories() - - # Search with enhanced logic - matches: List[Dict[str, str]] = [] - total_matches = 0 - for m in reversed(memories): # newest first - if _match_memory_with_query(m, parsed_query): - total_matches += 1 - if len(matches) < limit: - matches.append(m) - if not matches: - return f"No matches for: {query}" - lines = [ - f"{m['id'][:8]} [{m.get('timestamp','?')}] {m.get('text','')}" + (f" | tags: {m['tags']}" if m.get('tags') else "") - for m in matches - ] - omitted = total_matches - len(matches) - if omitted > 0: - lines.append(f"… ({omitted} additional match{'es' if omitted!=1 else ''} omitted; total_matches={total_matches})") - return "\n".join(lines) - - -def _mem_delete( - memory_id: Annotated[str, "Full UUID or a unique prefix (>=4 chars) of the memory id to delete."], -) -> str: - """(Internal) Delete one memory by UUID or unique prefix. - - Parameters: - memory_id (str): Full UUID4 (preferred) OR a unique prefix (>=4 chars). If prefix is ambiguous, no deletion occurs. - - Returns: - str: One of: success message, ambiguity notice, or not-found message. - - Safety: - Ambiguous prefixes are rejected to prevent accidental mass deletion. - """ - key = (memory_id or "").strip().lower() - if len(key) < 4: - return "Error: supply at least 4 characters of the id." - with _MEMORY_LOCK: - memories = _load_memories() - matched = [m for m in memories if m["id"].lower().startswith(key)] - if not matched: - return "Memory not found." - if len(matched) > 1 and key != matched[0]["id"].lower(): - # ambiguous prefix - sample = ", ".join(m["id"][:8] for m in matched[:5]) - more = "…" if len(matched) > 5 else "" - return f"Ambiguous prefix (matches {len(matched)} ids: {sample}{more}). Provide more characters." - # Unique match - target_id = matched[0]["id"] - memories = [m for m in memories if m["id"] != target_id] - _save_memories(memories) - return f"Deleted memory: {target_id}" - - -# ====================== -# UI: four-tab interface -# ====================== - -# --- Fetch tab (compact controllable extraction) --- -fetch_interface = gr.Interface( - fn=Web_Fetch, - inputs=[ - gr.Textbox(label="URL", placeholder="https://example.com/article"), - gr.Slider( - minimum=0, - maximum=20000, - value=3000, - step=100, - label="Max Characters", - info="0 = no limit (full page), default 3000" - ), - gr.Textbox( - label="Strip Selectors", - placeholder=".header, .footer, nav, .sidebar", - value="", - info="CSS selectors to remove (comma-separated)" - ), - gr.Checkbox( - label="URL Scraper", - value=False, - info="Extract only links instead of content" - ), - gr.Slider( - minimum=0, - maximum=100000, - value=0, - step=100, - label="Offset", - info="Character offset to start from (use next_cursor from previous call for pagination)" - ), - ], - outputs=gr.Markdown(label="Extracted Content"), - title="Web Fetch", - description=( - "<div style=\"text-align:center\">Convert any webpage to clean Markdown format with precision controls, or extract all links. Supports custom element removal, length limits, and pagination with offset.</div>" - ), - api_description=( - "Fetch a web page and return it converted to Markdown format or extract links with configurable options. " - "Includes enhanced truncation with detailed metadata and pagination support via offset parameter. " - "Parameters: url (str - absolute URL), max_chars (int - 0=no limit, default 3000), " - "strip_selectors (str - CSS selectors to remove, comma-separated), " - "url_scraper (bool - extract only links instead of content, default False), " - "offset (int - character offset for pagination, use next_cursor from previous call). " - "When content is truncated, returns detailed metadata including truncated status, character counts, " - "and next_cursor for continuation. When url_scraper=True, returns formatted list of all links found on the page." - ), - flagging_mode="never", -) - -# --- Web Search tab (readable output only) --- -concise_interface = gr.Interface( - fn=Web_Search, - inputs=[ - gr.Textbox(label="Query", placeholder="topic OR site:example.com"), - gr.Slider(minimum=1, maximum=20, value=5, step=1, label="Max results"), - gr.Slider(minimum=1, maximum=10, value=1, step=1, label="Page", info="Page number for pagination (ignored if offset > 0)"), - gr.Radio( - label="Search Type", - choices=["text", "news", "images", "videos", "books"], - value="text", - info="Type of content to search for" - ), - gr.Slider( - minimum=0, - maximum=1000, - value=0, - step=1, - label="Offset", - info="Result offset to start from (overrides page if > 0, use next_offset from previous search)" - ), - ], - outputs=gr.Textbox(label="Search Results", interactive=False), - title="Web Search", - description=( - "<div style=\"text-align:center\">Multi-type web search with readable output format, date detection, and flexible pagination. Supports text, news, images, videos, and books. Features smart fallback for news searches and precise offset control.</div>" - ), - api_description=( - "Run a web search (DuckDuckGo backend) with support for multiple content types and return formatted results. " - "Features smart fallback: if 'news' search returns no results, automatically retries with 'text' search " - "to catch sources like Hacker News that might not appear in news-specific results. " - "Supports advanced search operators: site: for specific domains, quotes for exact phrases, " - "OR for alternatives, and - to exclude terms. Examples: 'Python programming', 'site:example.com', " - "'\"artificial intelligence\"', 'cats -dogs', 'Python OR JavaScript'. " - "Parameters: query (str), max_results (int, 1-20), page (int, 1-based pagination), " - "search_type (str: text/news/images/videos/books), offset (int, result offset for precise continuation). " - "If offset > 0, it overrides the page parameter. Returns appropriately formatted results with metadata, " - "pagination hints, and next_offset information for each content type." - ), - flagging_mode="never", - submit_btn="Search", -) - -## - -# --- Code Interpreter tab (Python) --- -code_interface = gr.Interface( - fn=Code_Interpreter, - inputs=gr.Code(label="Python Code", language="python"), - outputs=gr.Textbox(label="Output"), - title="Code Interpreter", - description=( - "<div style=\"text-align:center\">Execute Python code and see the output.</div>" - ), - api_description=( - "Execute arbitrary Python code and return captured stdout or an error message. " - "Supports any valid Python code including imports, variables, functions, loops, and calculations. " - "Examples: 'print(2+2)', 'import math; print(math.sqrt(16))', 'for i in range(3): print(i)'. " - "Parameters: code (str - Python source code to execute). " - "Returns: Combined stdout output or exception text if execution fails." - ), - flagging_mode="never", -) - + except Exception as exc: + print(f"[TOOL RESULT] {func_name} (failed to log output: {exc})", flush=True) + +# Ensure Tools modules can import 'app' when this file is executed as a script +# (their code does `from app import ...`). +sys.modules.setdefault("app", sys.modules[__name__]) + +# Import per-tool interface builders from the Tools package +from Modules.Web_Fetch import build_interface as build_fetch_interface +from Modules.Web_Search import build_interface as build_search_interface +from Modules.Code_Interpreter import build_interface as build_code_interface +from Modules.Memory_Manager import build_interface as build_memory_interface +from Modules.Generate_Speech import build_interface as build_speech_interface +from Modules.Generate_Image import build_interface as build_image_interface +from Modules.Generate_Video import build_interface as build_video_interface +from Modules.Deep_Research import build_interface as build_research_interface + +# Optional environment flags used to conditionally show API schemas (unchanged behavior) +HF_IMAGE_TOKEN = bool(os.getenv("HF_READ_TOKEN")) +HF_VIDEO_TOKEN = bool(os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")) +HF_TEXTGEN_TOKEN = bool(os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")) + +# CSS copied from prior app.py to preserve exact look-and-feel CSS_STYLES = """ /* Style only the top-level app title to avoid affecting headings elsewhere */ .app-title { @@ -1574,7 +107,7 @@ CSS_STYLES = """ /* Place bold tools list on line 2, normal auth note on line 3 (below title) */ .app-title::before { grid-row: 2; - content: "Web Fetch | Web Search | Code Interpreter | Memory Manager | Generate Speech | Generate Image | Generate Video | Deep Research"; + content: "Web Fetch | Web Search | Code Interpreter | Memory Manager | Generate Speech | Generate Image | Generate Video | Deep Research"; display: block; font-size: 1rem; font-weight: 700; @@ -1752,1013 +285,24 @@ CSS_STYLES = """ } """ -# --- Generate Speech tab (text to speech) --- -available_voices = get_kokoro_voices() -kokoro_interface = gr.Interface( - fn=Generate_Speech, - inputs=[ - gr.Textbox(label="Text", placeholder="Type text to synthesize…", lines=4), - gr.Slider(minimum=0.5, maximum=2.0, value=1.25, step=0.1, label="Speed"), - gr.Dropdown( - label="Voice", - choices=available_voices, - value="af_heart", - info="Select from 54 available voices across multiple languages and accents" - ), - ], - outputs=gr.Audio(label="Audio", type="numpy", format="wav", show_download_button=True), - title="Generate Speech", - description=( - "<div style=\"text-align:center\">Generate speech with Kokoro-82M. Supports multiple languages and accents. Runs on CPU or CUDA if available.</div>" - ), - api_description=( - "Synthesize speech from text using Kokoro-82M TTS model. Returns (sample_rate, waveform) suitable for playback. " - "Parameters: text (str), speed (float 0.5–2.0, default 1.25x), voice (str, default 'af_heart'). " - "Voice Legend: af=American female, am=American male, bf=British female, bm=British male, ef=European female, em=European male, hf=Hindi female, hm=Hindi male, if=Italian female, im=Italian male, jf=Japanese female, jm=Japanese male, pf=Portuguese female, pm=Portuguese male, zf=Chinese female, zm=Chinese male, ff=French female. " - "All Voices: af_alloy, af_aoede, af_bella, af_heart, af_jessica, af_kore, af_nicole, af_nova, af_river, af_sarah, af_sky, am_adam, am_echo, am_eric, am_fenrir, am_liam, am_michael, am_onyx, am_puck, am_santa, bf_alice, bf_emma, bf_isabella, bf_lily, bm_daniel, bm_fable, bm_george, bm_lewis, ef_dora, em_alex, em_santa, ff_siwis, hf_alpha, hf_beta, hm_omega, hm_psi, if_sara, im_nicola, jf_alpha, jf_gongitsune, jf_nezumi, jf_tebukuro, jm_kumo, pf_dora, pm_alex, pm_santa, zf_xiaobei, zf_xiaoni, zf_xiaoxiao, zf_xiaoyi, zm_yunjian, zm_yunxi, zm_yunxia, zm_yunyang. " - "Return the generated media to the user in this format `![Alt text](URL)`" - ), - flagging_mode="never", -) - -def Memory_Manager( - action: Annotated[Literal["save","list","search","delete"], "Action to perform: save | list | search | delete"], - text: Annotated[Optional[str], "Text content (Save only)"] = None, - tags: Annotated[Optional[str], "Comma-separated tags (Save only)"] = None, - query: Annotated[Optional[str], "Enhanced search with tag:name syntax, AND/OR operators (Search only)"] = None, - limit: Annotated[int, "Max results (List/Search only)"] = 20, - memory_id: Annotated[Optional[str], "Full UUID or unique prefix (Delete only)"] = None, - include_tags: Annotated[bool, "Include tags (List/Search only)"] = True, -) -> str: - """Manage lightweight local JSON “memories” (save | list | search | delete) in one MCP tool. - - Overview: - This tool provides simple, local, append‑only style persistence for short text memories - with optional tags. Data is stored in a plaintext JSON file ("memories.json") beside the - application; no external database or network access is required. - - Supported Actions: - - save : Store a new memory (requires 'text'; optional 'tags'). - - list : Return the most recent memories (respects 'limit' + 'include_tags'). - - search : Enhanced AND match with tag: queries, boolean operators, and text terms (uses 'query', 'limit'). - - delete : Remove one memory by full UUID or unique prefix (uses 'memory_id'). - - Parameter Usage by Action: - action=save -> text (required), tags (optional) - action=list -> limit, include_tags - action=search -> query (required), limit, include_tags - action=delete -> memory_id (required) - - Parameters: - action (Literal[save|list|search|delete]): Operation selector (case-insensitive). - text (str): Raw memory content; leading/trailing whitespace trimmed (save only). - tags (str): Optional comma-separated tags; stored verbatim (save only). - query (str): Enhanced search query supporting: - - tag:name - search for specific tag - - AND/OR operators (case-insensitive, default is AND) - - Regular text terms search in text content and tags - - Examples: 'tag:work', 'tag:work AND tag:project', 'meeting tag:work', 'tag:urgent OR important' - limit (int): Maximum rows for list/search (clamped internally to 1–200). - memory_id (str): Full UUID or unique prefix (>=4 chars) (delete only). - include_tags (bool): When True, show tag column in list/search output. - - Storage Format (per entry): - {"id": "<uuid4>", "text": "<original text>", "timestamp": "YYYY-MM-DD HH:MM:SS", "tags": "tag1, tag2"} - - Lifecycle & Constraints: - - A soft cap of {_MAX_MEMORIES} entries is enforced by pruning oldest records on save. - - A light duplicate guard skips saving if the newest existing entry has identical text. - - All operations are protected by a thread‑local reentrant lock (NOT multi‑process safe). - - Returns: - str: Human‑readable status / result lines (never raw JSON) suitable for direct model consumption. - - Error Modes: - - Invalid action -> error string. - - Missing required field for the chosen action -> explanatory message. - - Ambiguous or unknown memory_id on delete -> clarification message. - - Security & Privacy: - Plaintext JSON; do not store secrets, credentials, or regulated personal data. - """ - act = (action or "").lower().strip() - - # Normalize None -> "" for internal helpers - text = text or "" - tags = tags or "" - query = query or "" - memory_id = memory_id or "" - - if act == "save": - if not text.strip(): - return "Error: 'text' is required when action=save." - return _mem_save(text=text, tags=tags) - if act == "list": - return _mem_list(limit=limit, include_tags=include_tags) - if act == "search": - if not query.strip(): - return "Error: 'query' is required when action=search." - return _mem_search(query=query, limit=limit) - if act == "delete": - if not memory_id.strip(): - return "Error: 'memory_id' is required when action=delete." - return _mem_delete(memory_id=memory_id) - return "Error: invalid action (use save|list|search|delete)." - -memory_interface = gr.Interface( - fn=Memory_Manager, - inputs=[ - gr.Dropdown(label="Action", choices=["save","list","search","delete"], value="list"), - gr.Textbox(label="Text", lines=3, placeholder="Memory text (save)"), - gr.Textbox(label="Tags", placeholder="tag1, tag2"), - gr.Textbox(label="Query", placeholder="tag:work AND tag:project OR meeting"), - gr.Slider(1, 200, value=20, step=1, label="Limit"), - gr.Textbox(label="Memory ID / Prefix", placeholder="UUID or prefix (delete)"), - gr.Checkbox(value=True, label="Include Tags"), - ], - outputs=gr.Textbox(label="Result", lines=14), - title="Memory Manager", - description=( - "<div style=\"text-align:center\">Lightweight local JSON memory store (no external DB). Choose an Action, fill only the relevant fields, and run.</div>" - ), - api_description=( - "Manage short text memories with optional tags. Actions: save(text,tags), list(limit,include_tags), " - "search(query,limit,include_tags), delete(memory_id). Enhanced search supports tag:name queries and AND/OR operators. " - "Examples: 'tag:work', 'tag:work AND tag:project', 'meeting tag:work', 'tag:urgent OR important'. " - "Action parameter is always required. Use Memory_Manager whenever you are given information worth remembering about the user, " - "and search for memories when relevant." - ), - flagging_mode="never", -) - -# ========================== -# Generate Image (Serverless) -# ========================== - -HF_API_TOKEN = os.getenv("HF_READ_TOKEN") - - -def Generate_Image( # <-- MCP tool #5 (Generate Image) - prompt: Annotated[str, "Text description of the image to generate."], - model_id: Annotated[str, "Hugging Face model id in the form 'creator/model-name' (e.g., black-forest-labs/FLUX.1-Krea-dev)."] = "black-forest-labs/FLUX.1-Krea-dev", - negative_prompt: Annotated[str, "What should NOT appear in the image." ] = ( - "(deformed, distorted, disfigured), poorly drawn, bad anatomy, wrong anatomy, extra limb, " - "missing limb, floating limbs, (mutated hands and fingers), disconnected limbs, mutation, " - "mutated, ugly, disgusting, blurry, amputation, misspellings, typos" - ), - steps: Annotated[int, "Number of denoising steps (1–100). Higher = slower, potentially higher quality."] = 35, - cfg_scale: Annotated[float, "Classifier-free guidance scale (1–20). Higher = follow the prompt more closely."] = 7.0, - sampler: Annotated[str, "Sampling method label (UI only). Common options: 'DPM++ 2M Karras', 'DPM++ SDE Karras', 'Euler', 'Euler a', 'Heun', 'DDIM'."] = "DPM++ 2M Karras", - seed: Annotated[int, "Random seed for reproducibility. Use -1 for a random seed per call."] = -1, - width: Annotated[int, "Output width in pixels (64–1216, multiple of 32 recommended)."] = 1024, - height: Annotated[int, "Output height in pixels (64–1216, multiple of 32 recommended)."] = 1024, -) -> Image.Image: - """ - Generate a single image from a text prompt using a Hugging Face model via serverless inference. - - Args: - prompt (str): Text description of the image to generate. - model_id (str): The Hugging Face model id (creator/model-name). Defaults to "black-forest-labs/FLUX.1-Krea-dev". - negative_prompt (str): What should NOT appear in the image. - steps (int): Number of denoising steps (1–100). Higher can improve quality. - cfg_scale (float): Guidance scale (1–20). Higher = follow the prompt more closely. - sampler (str): Sampling method label for UI; not all providers expose this control. - seed (int): Random seed. Use -1 to randomize on each call. - width (int): Output width in pixels (64–1216; multiples of 32 recommended). - height (int): Output height in pixels (64–1216; multiples of 32 recommended). - - Returns: - PIL.Image.Image: The generated image. - - Error modes: - - Raises gr.Error with a user-friendly message on auth/model/load errors. - """ - _log_call_start("Generate_Image", prompt=_truncate_for_log(prompt, 200), model_id=model_id, steps=steps, cfg_scale=cfg_scale, seed=seed, size=f"{width}x{height}") - if not prompt or not prompt.strip(): - _log_call_end("Generate_Image", "error=empty prompt") - raise gr.Error("Please provide a non-empty prompt.") - - # Slightly enhance prompt for quality (kept consistent with Serverless space) - enhanced_prompt = f"{prompt} | ultra detail, ultra elaboration, ultra quality, perfect." - - # Try multiple providers for resilience - providers = ["auto", "replicate", "fal-ai"] - last_error: Exception | None = None - - for provider in providers: - try: - client = InferenceClient(api_key=HF_API_TOKEN, provider=provider) - image = client.text_to_image( - prompt=enhanced_prompt, - negative_prompt=negative_prompt, - model=model_id, - width=width, - height=height, - num_inference_steps=steps, - guidance_scale=cfg_scale, - seed=seed if seed != -1 else random.randint(1, 1_000_000_000), - ) - _log_call_end("Generate_Image", f"provider={provider} size={image.size}") - return image - except Exception as e: # try next provider, transform last one to friendly error - last_error = e - continue - - # If we reach here, all providers failed - msg = str(last_error) if last_error else "Unknown error" - if "404" in msg: - raise gr.Error(f"Model not found or unavailable: {model_id}. Check the id and your HF token access.") - if "503" in msg: - raise gr.Error("The model is warming up. Please try again shortly.") - if "401" in msg or "403" in msg: - raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.") - # Map common provider auth messages to the same friendly guidance - low = msg.lower() - if ("api_key" in low) or ("hf auth login" in low) or ("unauthorized" in low) or ("forbidden" in low): - raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.") - _log_call_end("Generate_Image", f"error={_truncate_for_log(msg, 200)}") - raise gr.Error(f"Image generation failed: {msg}") - - -image_generation_interface = gr.Interface( - fn=Generate_Image, - inputs=[ - gr.Textbox(label="Prompt", placeholder="Enter a prompt", lines=2), - gr.Textbox( - label="Model", - value="black-forest-labs/FLUX.1-Krea-dev", - placeholder="creator/model-name", - info="<a href=\"https://huggingface.co/models?pipeline_tag=text-to-image&inference_provider=nebius,cerebras,novita,fireworks-ai,together,fal-ai,groq,featherless-ai,nscale,hyperbolic,sambanova,cohere,replicate,scaleway,publicai,hf-inference&sort=trending\" target=\"_blank\" rel=\"noopener noreferrer\">Browse models</a>", - ), - gr.Textbox( - label="Negative Prompt", - value=( - "(deformed, distorted, disfigured), poorly drawn, bad anatomy, wrong anatomy, extra limb, " - "missing limb, floating limbs, (mutated hands and fingers), disconnected limbs, mutation, " - "mutated, ugly, disgusting, blurry, amputation, misspellings, typos" - ), - lines=2, - ), - gr.Slider(minimum=1, maximum=100, value=35, step=1, label="Steps"), - gr.Slider(minimum=1.0, maximum=20.0, value=7.0, step=0.1, label="CFG Scale"), - gr.Radio(label="Sampler", value="DPM++ 2M Karras", choices=[ - "DPM++ 2M Karras", "DPM++ SDE Karras", "Euler", "Euler a", "Heun", "DDIM" - ]), - gr.Slider(minimum=-1, maximum=1_000_000_000, value=-1, step=1, label="Seed (-1 = random)"), - gr.Slider(minimum=64, maximum=1216, value=1024, step=32, label="Width"), - gr.Slider(minimum=64, maximum=1216, value=1024, step=32, label="Height"), - ], - outputs=gr.Image(label="Generated Image"), - title="Generate Image", - description=( - "<div style=\"text-align:center\">Generate images via Hugging Face serverless inference. " - "Default model is FLUX.1-Krea-dev.</div>" - ), - api_description=( - "Generate a single image from a text prompt using a Hugging Face model via serverless inference. " - "Supports creative prompts like 'a serene mountain landscape at sunset', 'portrait of a wise owl', " - "'futuristic city with flying cars'. Default model: FLUX.1-Krea-dev. " - "Parameters: prompt (str), model_id (str, creator/model-name), negative_prompt (str), steps (int, 1–100), " - "cfg_scale (float, 1–20), sampler (str), seed (int, -1=random), width/height (int, 64–1216). " - "Returns a PIL.Image. Return the generated media to the user in this format `![Alt text](URL)`" - ), - flagging_mode="never", - # Only expose to MCP when HF token is provided; UI tab is always visible - show_api=bool(os.getenv("HF_READ_TOKEN")), -) - -# ========================== -# Generate Video (Serverless) -# ========================== - -def _write_video_tmp(data_iter_or_bytes: object, suffix: str = ".mp4") -> str: - """Write video bytes or iterable of bytes to a system temporary file and return its path. - - This avoids polluting the project directory. The file is created in the OS temp - location; Gradio will handle serving & offering the download button. - """ - fd, fname = tempfile.mkstemp(suffix=suffix) - try: - with os.fdopen(fd, "wb") as f: - if isinstance(data_iter_or_bytes, (bytes, bytearray)): - f.write(data_iter_or_bytes) # type: ignore[arg-type] - elif hasattr(data_iter_or_bytes, "read"): - f.write(data_iter_or_bytes.read()) # type: ignore[call-arg] - elif hasattr(data_iter_or_bytes, "content"): - f.write(data_iter_or_bytes.content) # type: ignore[attr-defined] - elif hasattr(data_iter_or_bytes, "__iter__") and not isinstance(data_iter_or_bytes, (str, dict)): - for chunk in data_iter_or_bytes: # type: ignore[assignment] - if chunk: - f.write(chunk) - else: - raise gr.Error("Unsupported video data type returned by provider.") - except Exception: - # Clean up if writing failed - try: - os.remove(fname) - except Exception: - pass - raise - return fname - - -HF_VIDEO_TOKEN = os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN") - - -def Generate_Video( # <-- MCP tool #6 (Generate Video) - prompt: Annotated[str, "Text description of the video to generate (e.g., 'a red fox running through a snowy forest at sunrise')."], - model_id: Annotated[str, "Hugging Face model id in the form 'creator/model-name'. Defaults to Wan-AI/Wan2.2-T2V-A14B."] = "Wan-AI/Wan2.2-T2V-A14B", - negative_prompt: Annotated[str, "What should NOT appear in the video."] = "", - steps: Annotated[int, "Number of denoising steps (1–100). Higher can improve quality but is slower."] = 25, - cfg_scale: Annotated[float, "Guidance scale (1–20). Higher = follow the prompt more closely, lower = more creative."] = 3.5, - seed: Annotated[int, "Random seed for reproducibility. Use -1 for a random seed per call."] = -1, - width: Annotated[int, "Output width in pixels (multiples of 8 recommended)."] = 768, - height: Annotated[int, "Output height in pixels (multiples of 8 recommended)."] = 768, - fps: Annotated[int, "Frames per second of the output video (e.g., 24)."] = 24, - duration: Annotated[float, "Target duration in seconds (provider/model dependent, commonly 2–6s)."] = 4.0, -) -> str: - """ - Generate a short video from a text prompt using a Hugging Face model via serverless inference. - - Args: - prompt (str): Text description of the video to generate. - model_id (str): The Hugging Face model id (creator/model-name). Defaults to "Wan-AI/Wan2.2-T2V-A14B". - negative_prompt (str): What should NOT appear in the video. - steps (int): Number of denoising steps (1–100). Higher can improve quality but is slower. - cfg_scale (float): Guidance scale (1–20). Higher = follow the prompt more closely. - seed (int): Random seed. Use -1 to randomize on each call. - width (int): Output width in pixels. - height (int): Output height in pixels. - fps (int): Frames per second. - duration (float): Target duration in seconds. - - Returns: - str: Path to an MP4 file on disk (Gradio will serve this file; MCP converts it to a file URL). - - Error modes: - - Raises gr.Error with a user-friendly message on auth/model/load errors or unsupported parameters. - """ - _log_call_start("Generate_Video", prompt=_truncate_for_log(prompt, 160), model_id=model_id, steps=steps, cfg_scale=cfg_scale, fps=fps, duration=duration, size=f"{width}x{height}") - if not prompt or not prompt.strip(): - _log_call_end("Generate_Video", "error=empty prompt") - raise gr.Error("Please provide a non-empty prompt.") - - if not HF_VIDEO_TOKEN: - # Still attempt without a token (public models), but warn earlier if it fails. - pass - - providers = ["auto", "replicate", "fal-ai"] - last_error: Exception | None = None - - # Build a common parameters dict. Providers may ignore unsupported keys. - parameters = { - "negative_prompt": negative_prompt or None, - "num_inference_steps": steps, - "guidance_scale": cfg_scale, - "seed": seed if seed != -1 else random.randint(1, 1_000_000_000), - "width": width, - "height": height, - "fps": fps, - # Some providers/models expect num_frames instead of duration; we pass both-friendly value - # when supported; they may be ignored by the backend. - "duration": duration, - } - - for provider in providers: - try: - client = InferenceClient(api_key=HF_VIDEO_TOKEN, provider=provider) - # Use the documented text_to_video API with correct parameters - if hasattr(client, "text_to_video"): - # Calculate num_frames from duration and fps if both provided - num_frames = int(duration * fps) if duration and fps else None - - # Build extra_body for provider-specific parameters - extra_body = {} - if width: - extra_body["width"] = width - if height: - extra_body["height"] = height - if fps: - extra_body["fps"] = fps - if duration: - extra_body["duration"] = duration - - result = client.text_to_video( - prompt=prompt, - model=model_id, - guidance_scale=cfg_scale, - negative_prompt=[negative_prompt] if negative_prompt else None, - num_frames=num_frames, - num_inference_steps=steps, - seed=parameters["seed"], - extra_body=extra_body if extra_body else None, - ) - else: - # Generic POST fallback for older versions - result = client.post( - model=model_id, - json={ - "inputs": prompt, - "parameters": {k: v for k, v in parameters.items() if v is not None}, - }, - ) - - # Save output to an .mp4 - path = _write_video_tmp(result, suffix=".mp4") - try: - size = os.path.getsize(path) - except Exception: - size = -1 - _log_call_end("Generate_Video", f"provider={provider} path={os.path.basename(path)} bytes={size}") - return path - except Exception as e: - last_error = e - continue - - msg = str(last_error) if last_error else "Unknown error" - if "404" in msg: - raise gr.Error(f"Model not found or unavailable: {model_id}. Check the id and HF token access.") - if "503" in msg: - raise gr.Error("The model is warming up. Please try again shortly.") - if "401" in msg or "403" in msg: - raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.") - # Map common provider auth messages to the same friendly guidance - low = msg.lower() - if ("api_key" in low) or ("hf auth login" in low) or ("unauthorized" in low) or ("forbidden" in low): - raise gr.Error("Please duplicate the space and provide a `HF_READ_TOKEN` to enable Image and Video Generation.") - _log_call_end("Generate_Video", f"error={_truncate_for_log(msg, 200)}") - raise gr.Error(f"Video generation failed: {msg}") - - -video_generation_interface = gr.Interface( - fn=Generate_Video, - inputs=[ - gr.Textbox(label="Prompt", placeholder="Enter a prompt for the video", lines=2), - gr.Textbox( - label="Model", - value="Wan-AI/Wan2.2-T2V-A14B", - placeholder="creator/model-name", - info="<a href=\"https://huggingface.co/models?pipeline_tag=text-to-video&inference_provider=nebius,cerebras,novita,fireworks-ai,together,fal-ai,groq,featherless-ai,nscale,hyperbolic,sambanova,cohere,replicate,scaleway,publicai,hf-inference&sort=trending\" target=\"_blank\" rel=\"noopener noreferrer\">Browse models</a>", - ), - gr.Textbox(label="Negative Prompt", value="", lines=2), - gr.Slider(minimum=1, maximum=100, value=25, step=1, label="Steps"), - gr.Slider(minimum=1.0, maximum=20.0, value=3.5, step=0.1, label="CFG Scale"), - gr.Slider(minimum=-1, maximum=1_000_000_000, value=-1, step=1, label="Seed (-1 = random)"), - gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Width"), - gr.Slider(minimum=64, maximum=1920, value=768, step=8, label="Height"), - gr.Slider(minimum=4, maximum=60, value=24, step=1, label="FPS"), - gr.Slider(minimum=1.0, maximum=10.0, value=4.0, step=0.5, label="Duration (s)"), - ], - outputs=gr.Video(label="Generated Video", show_download_button=True, format="mp4"), - title="Generate Video", - description=( - "<div style=\"text-align:center\">Generate short videos via Hugging Face serverless inference. " - "Default model is Wan2.2-T2V-A14B.</div>" - ), - api_description=( - "Generate a short video from a text prompt using a Hugging Face model via serverless inference. " - "Create dynamic scenes like 'a red fox running through a snowy forest at sunrise', 'waves crashing on a rocky shore', " - "'time-lapse of clouds moving across a blue sky'. Default model: Wan2.2-T2V-A14B (2-6 second videos). " - "Parameters: prompt (str), model_id (str), negative_prompt (str), steps (int), cfg_scale (float), seed (int), " - "width/height (int), fps (int), duration (float in seconds). Returns MP4 file path. " - "Return the generated media to the user in this format `![Alt text](URL)`" - ), - flagging_mode="never", - # Only expose to MCP when HF token is provided; UI tab is always visible - show_api=bool(os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN")), -) - -# ========================== -# Deep Research (Search + Fetch + LLM) -# ========================== - -HF_TEXTGEN_TOKEN = os.getenv("HF_READ_TOKEN") or os.getenv("HF_TOKEN") - - -def _normalize_query(q: str) -> str: - """Normalize fancy quotes and stray punctuation in queries. - - - Replace curly quotes with straight quotes - - Collapse multiple quotes/spaces - - Strip leading/trailing quotes - """ - if not q: - return "" - repl = { - "“": '"', - "”": '"', - "‘": "'", - "’": "'", - "`": "'", - } - for k, v in repl.items(): - q = q.replace(k, v) - # Remove duplicated quotes and excessive spaces - q = re.sub(r'\s+', ' ', q) - q = re.sub(r'"\s+"', ' ', q) - q = q.strip().strip('"').strip() - return q - - -def _search_urls_only(query: str, max_results: int) -> list[str]: - """Return a list of result URLs using DuckDuckGo search with rate limiting. - - Uses ddgs to fetch web results only (no news/images/videos). Falls back to empty list on error. - """ - if not query or not query.strip() or max_results <= 0: - return [] - urls: list[str] = [] - try: - _search_rate_limiter.acquire() - with DDGS() as ddgs: - for item in ddgs.text(query, region="wt-wt", safesearch="moderate", max_results=max_results): - url = (item.get("href") or item.get("url") or "").strip() - if url: - urls.append(url) - except Exception: - pass - # De-duplicate while preserving order - seen = set() - deduped = [] - for u in urls: - if u not in seen: - seen.add(u) - deduped.append(u) - return deduped - - -def _fetch_page_markdown(url: str, max_chars: int = 3000) -> str: - """Fetch a single URL and return cleaned Markdown using existing Web_Fetch. - - Returns empty string on error. - """ - try: - # Intentionally skip global fetch rate limiting for Deep Research speed. - return Web_Fetch(url=url, max_chars=max_chars, strip_selectors="", url_scraper=False, offset=0) # type: ignore[misc] - except Exception: - return "" - - -def _truncate_join(parts: list[str], max_chars: int) -> tuple[str, bool]: - out = [] - total = 0 - truncated = False - for p in parts: - if not p: - continue - if total + len(p) > max_chars: - out.append(p[: max(0, max_chars - total)]) - truncated = True - break - out.append(p) - total += len(p) - return ("\n\n".join(out), truncated) - - -def _build_research_prompt( - summary: str, - queries: list[str], - url_list: list[str], - pages_map: dict[str, str], -) -> str: - researcher_instructions = ( - "You are Nymbot, a helpful deep research assistant. You will be asked a Query from a user and you will create a long, comprehensive, well-structured research report in response to the user's Query.\n\n" - "You have been provided with User Question, Search Queries, and numerous webpages that the searches yielded.\n\n" - "<report_format>\n" - "Write a well-formatted report in the structure of a scientific report to a broad audience. The report must be readable and have a nice flow of Markdown headers and paragraphs of text. Do NOT use bullet points or lists which break up the natural flow. The report must be exhaustive for comprehensive topics.\n" - "For any given user query, first determine the major themes or areas that need investigation, then structure these as main sections, and develop detailed subsections that explore various facets of each theme. Each section and subsection requires paragraphs of texts that need to all connect into one narrative flow.\n" - "</report_format>\n\n" - "<document_structure>\n" - "- Always begin with a clear title using a single # header\n" - "- Organize content into major sections using ## headers\n" - "- Further divide into subsections using ### headers\n" - "- Use #### headers sparingly for special subsections\n" - "- Never skip header levels\n" - "- Write multiple paragraphs per section or subsection\n" - "- Each paragraph must contain at least 4-5 sentences, present novel insights and analysis grounded in source material, connect ideas to original query, and build upon previous paragraphs to create a narrative flow\n" - "- Never use lists, instead always use text or tables\n\n" - "Mandatory Section Flow:\n" - "1. Title (# level)\n - Before writing the main report, start with one detailed paragraph summarizing key findings\n" - "2. Main Body Sections (## level)\n - Each major topic gets its own section (## level). There MUST BE at least 5 sections.\n - Use ### subsections for detailed analysis\n - Every section or subsection needs at least one paragraph of narrative before moving to the next section\n - Do NOT have a section titled \"Main Body Sections\" and instead pick informative section names that convey the theme of the section\n" - "3. Conclusion (## level)\n - Synthesis of findings\n - Potential recommendations or next steps\n" - "</document_structure>\n\n" - "<planning_rules>\n" - "- Always break it down into multiple steps\n" - "- Assess the different sources and whether they are useful for any steps needed to answer the query\n" - "- Create the best report that weighs all the evidence from the sources\n" - "- Remember that the current date is: Wednesday, April 23, 2025, 11:50 AM EDT\n" - "- Make sure that your final report addresses all parts of the query\n" - "- Communicate a brief high-level plan in the introduction; do not reveal chain-of-thought.\n" - "- When referencing sources during analysis, you should still refer to them by index with brackets and follow <citations>\n" - "- As a final step, review your planned report structure and ensure it completely answers the query.\n" - "</planning_rules>\n\n" - ) - - # Build sources block limited to a reasonable size to avoid overrun - # Cap combined sources to ~180k characters - sources_blocks: list[str] = [] - indexed_urls: list[str] = [] - for idx, u in enumerate(url_list, start=1): - txt = pages_map.get(u, "").strip() - if not txt: - continue - indexed_urls.append(f"[{idx}] {u}") - # Prefix each source with its index and URL for citation - sources_blocks.append(f"[Source {idx}] URL: {u}\n\n{txt}") - - # Cap combined sources aggressively to stay within provider limits - sources_joined, truncated = _truncate_join(sources_blocks, max_chars=100_000) - - prompt = [] - prompt.append(researcher_instructions) - prompt.append("<user_query_summary>\n" + (summary or "") + "\n</user_query_summary>\n") - # Include populated queries only - populated = [q for q in queries if q and q.strip()] - if populated: - prompt.append("<search_queries>\n" + "\n".join(f"- {q.strip()}" for q in populated) + "\n</search_queries>\n") - if indexed_urls: - prompt.append("<sources_list>\n" + "\n".join(indexed_urls) + "\n</sources_list>\n") - prompt.append("<fetched_documents>\n" + sources_joined + ("\n\n[NOTE] Sources truncated due to context limits." if truncated else "") + "\n</fetched_documents>") - return "\n\n".join(prompt) - - -def _write_report_tmp(text: str) -> str: - # Create a unique temp directory and write a deterministic filename inside it. - tmp_dir = tempfile.mkdtemp(prefix="deep_research_") - path = os.path.join(tmp_dir, "research_report.txt") - with open(path, "w", encoding="utf-8") as f: - f.write(text) - return path - - -def Deep_Research( - summary: Annotated[str, "Summarization of research topic (one or more sentences)."], - query1: Annotated[str, "DDG Search Query 1"], - max1: Annotated[int, "Max results for Query 1 (1-50)"] = 10, - query2: Annotated[str, "DDG Search Query 2"] = "", - max2: Annotated[int, "Max results for Query 2 (1-50)"] = 10, - query3: Annotated[str, "DDG Search Query 3"] = "", - max3: Annotated[int, "Max results for Query 3 (1-50)"] = 10, - query4: Annotated[str, "DDG Search Query 4"] = "", - max4: Annotated[int, "Max results for Query 4 (1-50)"] = 10, - query5: Annotated[str, "DDG Search Query 5"] = "", - max5: Annotated[int, "Max results for Query 5 (1-50)"] = 10, -) -> tuple[str, str, str]: - """ - Run deep research by searching, fetching pages, and generating a comprehensive report via a large LLM provider. - - Pipeline: - 1) Perform up to 5 DuckDuckGo searches (URLs only). If total requested > 50, each query is limited to 10. - 2) Fetch all discovered URLs (up to 50) as cleaned Markdown (max 3000 chars per page). - 3) Call Hugging Face Inference Providers (Cerebras) with model `Qwen/Qwen3-235B-A22B-Instruct-2507` to write a research report. - - Args: - summary (str): A brief description of the overall research topic or user question. - This is shown to the researcher model and used to frame the report. - query1 (str): DuckDuckGo search query #1. Required if you want any results. - Example: "site:nature.com CRISPR ethical implications". - max1 (int): Maximum number of URLs to take from query #1 (1–50). - If the combined total requested across all queries exceeds 50, each query will be capped to 10. - query2 (str): DuckDuckGo search query #2. Optional; leave empty to skip. - max2 (int): Maximum number of URLs to take from query #2 (1–50). - query3 (str): DuckDuckGo search query #3. Optional; leave empty to skip. - max3 (int): Maximum number of URLs to take from query #3 (1–50). - query4 (str): DuckDuckGo search query #4. Optional; leave empty to skip. - max4 (int): Maximum number of URLs to take from query #4 (1–50). - query5 (str): DuckDuckGo search query #5. Optional; leave empty to skip. - max5 (int): Maximum number of URLs to take from query #5 (1–50). - - Returns: - - Markdown research report - - Newline-separated list of fetched URLs - - Path to a downloadable .txt file containing the full report - - Raises: - gr.Error: If a required Hugging Face token is not provided or if the researcher - model call fails after retries. - - Notes: - - Total URLs across queries are capped at 50. - - Each fetched page is truncated to ~3000 characters before prompting the model. - - The function is optimized to complete within typical MCP time budgets. - """ - _log_call_start( - "Deep_Research", - summary=_truncate_for_log(summary or "", 200), - queries=[q for q in [query1, query2, query3, query4, query5] if q], - ) - - # Validate token - if not HF_TEXTGEN_TOKEN: - _log_call_end("Deep_Research", "error=missing HF token") - raise gr.Error("Please provide a `HF_READ_TOKEN` to enable Deep Research.") - - # Normalize caps per spec and sanitize queries - queries = [ - _normalize_query(query1 or ""), - _normalize_query(query2 or ""), - _normalize_query(query3 or ""), - _normalize_query(query4 or ""), - _normalize_query(query5 or ""), - ] - reqs = [max(1, min(50, int(max1))), max(1, min(50, int(max2))), max(1, min(50, int(max3))), max(1, min(50, int(max4))), max(1, min(50, int(max5)))] - total_requested = sum(reqs) - if total_requested > 50: - # Enforce rule: each query fetches 10 results when over 50 total requested - reqs = [10, 10, 10, 10, 10] - - # Overall deadline to avoid MCP 60s timeout (reserve ~5s for prompt+inference) - start_ts = time.time() - budget_seconds = 55.0 - deadline = start_ts + budget_seconds - - def time_left() -> float: - return max(0.0, deadline - time.time()) - - # 1) Run searches (parallelize queries to reduce latency) and stop if budget exceeded - all_urls: list[str] = [] - from concurrent.futures import ThreadPoolExecutor, as_completed - tasks = [] - with ThreadPoolExecutor(max_workers=min(5, sum(1 for q in queries if q.strip())) or 1) as executor: - for q, n in zip(queries, reqs): - if not q.strip(): - continue - tasks.append(executor.submit(_search_urls_only, q.strip(), n)) - for fut in as_completed(tasks): - try: - urls = fut.result() or [] - except Exception: - urls = [] - for u in urls: - if u not in all_urls: - all_urls.append(u) - if len(all_urls) >= 50: - break - if time_left() <= 0.5: - # Out of budget for searching; stop early - break - # Don't block on leftover tasks; cancel/shutdown immediately - # Python futures don't support true cancel if running, but we can just avoid waiting - # and let executor context exit cleanly. - if len(all_urls) > 50: - all_urls = all_urls[:50] - - # Filter obviously irrelevant/shopping/dictionary/forum domains that often appear due to phrase tokenization - blacklist = { - "homedepot.com", - "tractorsupply.com", - "mcmaster.com", - "mrchain.com", - "answers.com", - "city-data.com", - "dictionary.cambridge.org", - } - def _domain(u: str) -> str: - try: - return urlparse(u).netloc.lower() - except Exception: - return "" - all_urls = [u for u in all_urls if _domain(u) not in blacklist] - - # Skip known large/non-HTML file types to avoid wasted fetch time - skip_exts = ( - ".pdf", ".ppt", ".pptx", ".doc", ".docx", ".xls", ".xlsx", - ".zip", ".gz", ".tgz", ".bz2", ".7z", ".rar" - ) - def _skip_url(u: str) -> bool: - try: - path = urlparse(u).path.lower() - except Exception: - return False - return any(path.endswith(ext) for ext in skip_exts) - all_urls = [u for u in all_urls if not _skip_url(u)] - - # 2) Fetch pages (markdown, 3000 chars) with slow-host requeue (3s delay), respecting deadline - pages: dict[str, str] = {} - if all_urls: - from concurrent.futures import ThreadPoolExecutor, Future - from collections import deque - - queue = deque(all_urls) - attempts: dict[str, int] = {u: 0 for u in all_urls} - max_attempts = 2 # fewer retries to honor budget - max_workers = min(12, max(4, len(all_urls))) - - in_flight: dict[Future, str] = {} - - def schedule_next(executor: ThreadPoolExecutor) -> None: - while queue and len(in_flight) < max_workers: - u = queue.popleft() - # Skip if already fetched or exceeded attempts - if u in pages: - continue - if attempts[u] >= max_attempts: - continue - attempts[u] += 1 - # Adaptive per-attempt timeout based on time remaining; min 2s, max 10s - tl = time_left() - per_timeout = 10.0 if tl > 15 else (5.0 if tl > 8 else 2.0) - fut = executor.submit(_fetch_page_markdown_fast, u, 3000, per_timeout) - in_flight[fut] = u - - delayed: list[tuple[float, str]] = [] # (ready_time, url) - - with ThreadPoolExecutor(max_workers=max_workers) as executor: - schedule_next(executor) - - while (in_flight or queue) and time_left() > 0.2: - # Move any delayed items whose time has arrived back into the queue - now = time.time() - if delayed: - ready, not_ready = [], [] - for t, u in delayed: - (ready if t <= now else not_ready).append((t, u)) - delayed = not_ready - for _, u in ready: - queue.append(u) - # Try to schedule newly ready URLs - if ready: - schedule_next(executor) - - done: list[Future] = [] - # Poll completed futures without blocking too long - for fut in list(in_flight.keys()): - if fut.done(): - done.append(fut) - - if not done: - # If nothing to do but we have delayed items pending, sleep until next due time (capped) - if not queue and delayed: - sleep_for = max(0.02, min(0.25, max(0.0, min(t for t, _ in delayed) - time.time()))) - time.sleep(sleep_for) - else: - # brief sleep to avoid busy spin - time.sleep(0.05) - else: - for fut in done: - u = in_flight.pop(fut) - try: - md = fut.result() - if md and not md.startswith("Unsupported content type") and not md.startswith("An error occurred"): - pages[u] = md - try: - print(f"[FETCH OK] {u} (chars={len(md)})", flush=True) - except Exception: - pass - else: - # If empty due to non-timeout error, don't retry further - pass - except SlowHost: - # Requeue to the back after 3 seconds - # But only if we have enough time left for a retry window - if time_left() > 5.0: - delayed.append((time.time() + 3.0, u)) - except Exception: - # Non-timeout error; skip - pass - # After handling done items, try to schedule more - schedule_next(executor) - - # If budget is nearly up and no pages were fetched, fall back to using the unique URL list in prompt (no content) - # The prompt builder will include sources list even if pages_map is empty; LLM can still reason over URLs indirectly. - - # Build final prompt - prompt = _build_research_prompt(summary=summary or "", queries=[q for q in queries if q.strip()], url_list=list(pages.keys()), pages_map=pages) - - # 3) Call the Researcher model via Cerebras provider with robust fallbacks - messages = [ - {"role": "system", "content": "You are Nymbot, an expert deep research assistant."}, - {"role": "user", "content": prompt}, - ] - try: - prompt_chars = len(prompt) - except Exception: - prompt_chars = -1 - print(f"[PIPELINE] Fetch complete: pages={len(pages)}, unique_urls={len(pages.keys())}, prompt_chars={prompt_chars}", flush=True) - print("[PIPELINE] Starting inference (provider=cerebras, model=Qwen/Qwen3-235B-A22B-Thinking-2507)", flush=True) - def _run_inference(provider: str, max_tokens: int, temp: float, top_p: float): - client = InferenceClient(provider=provider, api_key=HF_TEXTGEN_TOKEN) - return client.chat.completions.create( - model="Qwen/Qwen3-235B-A22B-Thinking-2507", - messages=messages, - max_tokens=max_tokens, - temperature=temp, - top_p=top_p, - ) - try: - # Attempt 1: Cerebras, full prompt - print("[LLM] Attempt 1: provider=cerebras, max_tokens=32768", flush=True) - completion = _run_inference("cerebras", max_tokens=32768, temp=0.3, top_p=0.95) - except Exception as e1: - print(f"[LLM] Attempt 1 failed: {str(e1)[:200]}", flush=True) - # Attempt 2: Cerebras, trimmed prompt and lower max_tokens - try: - prompt2 = _build_research_prompt(summary=summary or "", queries=[q for q in queries if q.strip()], url_list=list(pages.keys())[:30], pages_map={k: pages[k] for k in list(pages.keys())[:30]}) - messages = [ - {"role": "system", "content": "You are Nymbot, an expert deep research assistant."}, - {"role": "user", "content": prompt2}, - ] - print("[LLM] Attempt 2: provider=cerebras (trimmed), max_tokens=16384", flush=True) - completion = _run_inference("cerebras", max_tokens=16384, temp=0.7, top_p=0.95) - except Exception as e2: - print(f"[LLM] Attempt 2 failed: {str(e2)[:200]}", flush=True) - # Attempt 3: provider auto-fallback with trimmed prompt - try: - print("[LLM] Attempt 3: provider=auto, max_tokens=8192", flush=True) - completion = _run_inference("auto", max_tokens=8192, temp=0.7, top_p=0.95) - except Exception as e3: - _log_call_end("Deep_Research", f"error={_truncate_for_log(str(e3), 260)}") - raise gr.Error(f"Researcher model call failed: {e3}") - raw = completion.choices[0].message.content or "" - # 1) Strip any internal <think>...</think> blocks produced by the Thinking model - try: - no_think = re.sub(r"<think>[\s\S]*?<\\/think>", "", raw, flags=re.IGNORECASE) - no_think = re.sub(r"<\\/?think>", "", no_think, flags=re.IGNORECASE) - except Exception: - no_think = raw - - # 2) Remove planning / meta-analysis paragraphs that are part of the model's visible thinking trace. - # Heuristics: paragraphs (double-newline separated) containing phrases like "let me", "first,", "now i'll", - # "i will", "i'll", "let's", "now let me", or starting with "first" (case-insensitive). - try: - paragraphs = [p for p in re.split(r"\n\s*\n", no_think) if p.strip()] - keep: list[str] = [] - removed = 0 - planning_re = re.compile(r"\b(let me|now i(?:'ll| will)?|first,|i will now|i will|i'll|let's|now let me|i need to|i will now|now i'll|now i will)\b", re.IGNORECASE) - for p in paragraphs: - # If the paragraph looks like explicit planning/analysis, drop it - if planning_re.search(p): - removed += 1 - continue - keep.append(p) - report = "\n\n".join(keep).strip() - # If we removed everything, fall back to the no_think version - if not report: - report = no_think.strip() - except Exception: - report = no_think - - # 3) Final whitespace normalization - report = re.sub(r"\n\s*\n\s*\n+", "\n\n", report) - # Emit a short postprocess log - try: - print(f"[POSTPROCESS] removed_planning_paragraphs={removed}, raw_chars={len(raw)}, final_chars={len(report)}", flush=True) - except Exception: - pass - - # Build outputs - links_text = "\n".join([f"[{i+1}] {u}" for i, u in enumerate(pages.keys())]) - file_path = _write_report_tmp(report) - elapsed = time.time() - start_ts - # Print explicit timing and include in structured log output - print(f"[TIMING] Deep_Research elapsed: {elapsed:.2f}s", flush=True) - _log_call_end("Deep_Research", f"urls={len(pages)} file={os.path.basename(file_path)} duration={elapsed:.2f}s") - return report, links_text, file_path - - -deep_research_interface = gr.Interface( - fn=Deep_Research, - inputs=[ - gr.Textbox(label="Summarization of research topic", lines=3, placeholder="Briefly summarize the research topic or user question"), - gr.Textbox(label="DDG Search Query 1"), - gr.Slider(1, 50, value=10, step=1, label="Max results (Q1)"), - gr.Textbox(label="DDG Search Query 2", value=""), - gr.Slider(1, 50, value=10, step=1, label="Max results (Q2)"), - gr.Textbox(label="DDG Search Query 3", value=""), - gr.Slider(1, 50, value=10, step=1, label="Max results (Q3)"), - gr.Textbox(label="DDG Search Query 4", value=""), - gr.Slider(1, 50, value=10, step=1, label="Max results (Q4)"), - gr.Textbox(label="DDG Search Query 5", value=""), - gr.Slider(1, 50, value=10, step=1, label="Max results (Q5)"), - ], - outputs=[ - gr.Markdown(label="Research Report"), - gr.Textbox(label="Fetched Links", lines=8), - gr.File(label="Download Research Report", file_count="single"), - ], - title="Deep Research", - description=( - "<div style=\"text-align:center\">Perform multi-query web research: search with DuckDuckGo, fetch up to 50 pages in parallel, " - "and generate a comprehensive report using a large LLM via Hugging Face Inference Providers (Cerebras). Requires HF_READ_TOKEN.</div>" - ), - api_description=( - "Runs 1–5 DDG searches (URLs only), caps total results to 50 (when exceeding, each query returns 10). " - "Fetches all URLs (3000 chars each) and calls the Researcher to write a research report. " - "Returns the report (Markdown), the list of sources, and a downloadable text file path. " - "Provide the user with one-paragraph summary of the research report and the txt file in this format `![research_report.txt](URL)`" - ), - flagging_mode="never", - show_api=bool(HF_TEXTGEN_TOKEN), -) +# Build each tab interface using modular builders +fetch_interface = build_fetch_interface() +concise_interface = build_search_interface() +code_interface = build_code_interface() +memory_interface = build_memory_interface() +kokoro_interface = build_speech_interface() +image_generation_interface = build_image_interface() +video_generation_interface = build_video_interface() +deep_research_interface = build_research_interface() _interfaces = [ fetch_interface, concise_interface, code_interface, - memory_interface, # Always visible in UI + memory_interface, kokoro_interface, - image_generation_interface, # Always visible in UI - video_generation_interface, # Always visible in UI + image_generation_interface, + video_generation_interface, deep_research_interface, ] _tab_names = [ @@ -2773,10 +317,9 @@ _tab_names = [ ] with gr.Blocks(title="Nymbo/Tools MCP", theme="Nymbo/Nymbo_Theme", css=CSS_STYLES) as demo: - # Page title (scoped styling via .app-title to avoid affecting other headings) + # Title and information panel unchanged to preserve UI gr.HTML("<h1 class='app-title'>Nymbo/Tools MCP</h1>") - # Collapsed Information accordion (appears below subtitle and above tabs) with gr.Accordion("Information", open=False): gr.HTML( """ @@ -2823,15 +366,14 @@ with gr.Blocks(title="Nymbo/Tools MCP", theme="Nymbo/Nymbo_Theme", css=CSS_STYLE <br/> <code>https://mcp.nymbo.net/gradio_api/mcp/</code> </p> - <p>Example client configuration:</p> - <pre><code class="language-json">{ + <p>Example client configuration:</p> + <pre><code class="language-json">{ "mcpServers": { "nymbo-tools": { "url": "https://mcp.nymbo.net/gradio_api/mcp/" } } }</code></pre> - </div> </section> @@ -2870,9 +412,8 @@ with gr.Blocks(title="Nymbo/Tools MCP", theme="Nymbo/Nymbo_Theme", css=CSS_STYLE """ ) - # Existing tool tabs gr.TabbedInterface(interface_list=_interfaces, tab_names=_tab_names) -# Launch the UI and expose all functions as MCP tools in one server if __name__ == "__main__": + # Launch with MCP server enabled as before demo.launch(mcp_server=True) \ No newline at end of file