#!/usr/bin/env python3 """ Standalone Web Tools Module This module provides generic web tools that work with multiple backend providers. Backend is selected during ``hermes tools`` setup (web.backend in config.yaml). When available, Hermes can route Firecrawl calls through a Nous-hosted tool-gateway for Nous Subscribers only. Available tools: - web_search_tool: Search the web for information - web_extract_tool: Extract content from specific web pages - web_crawl_tool: Crawl websites with specific instructions Backend compatibility: - Exa: https://exa.ai (search, extract) - Firecrawl: https://docs.firecrawl.dev/introduction (search, extract, crawl; direct or derived firecrawl-gateway. for Nous Subscribers) - Parallel: https://docs.parallel.ai (search, extract) - Tavily: https://tavily.com (search, extract, crawl) LLM Processing: - Uses OpenRouter API with Gemini 3 Flash Preview for intelligent content extraction - Extracts key excerpts and creates markdown summaries to reduce token usage Debug Mode: - Set WEB_TOOLS_DEBUG=true to enable detailed logging - Creates web_tools_debug_UUID.json in ./logs directory - Captures all tool calls, results, and compression metrics Usage: from web_tools import web_search_tool, web_extract_tool, web_crawl_tool # Search the web results = web_search_tool("Python machine learning libraries", limit=3) # Extract content from URLs content = web_extract_tool(["https://example.com"], format="markdown") # Crawl a website crawl_data = web_crawl_tool("example.com", "Find contact information") """ import json import logging import os import re import asyncio from typing import List, Dict, Any, Optional, TYPE_CHECKING import httpx # noqa: F401 — kept at module top so tests can patch tools.web_tools.httpx # After the web-provider plugin migration (PR #25182), the Firecrawl SDK # proxy, client construction, and response-shape normalizers all live in # plugins.web.firecrawl.provider. We re-export the names that external # code, integration tests, and unit-test patches reach for so the public # surface stays stable. if TYPE_CHECKING: from firecrawl import Firecrawl # noqa: F401 — type hints only from plugins.web.firecrawl.provider import ( Firecrawl, _FirecrawlProxy, _FIRECRAWL_CLS_CACHE, _extract_scrape_payload, _extract_web_search_results, _firecrawl_backend_help_suffix, _get_direct_firecrawl_config, _get_firecrawl_client, _get_firecrawl_gateway_url, _has_direct_firecrawl_config, _is_tool_gateway_ready, _load_firecrawl_cls, _normalize_result_list, _raise_web_backend_configuration_error, _to_plain_object, check_firecrawl_api_key, ) # Tavily helpers re-exported for backward-compat with existing unit tests # (tests/tools/test_web_tools_tavily.py imports these names directly). from plugins.web.tavily.provider import ( # noqa: F401 — backward-compat names _normalize_tavily_documents, _normalize_tavily_search_results, _tavily_request, ) # Parallel + Exa clients re-exported for backward-compat with existing # unit tests (tests/tools/test_web_tools_config.py imports _get_parallel_client # / _get_async_parallel_client / _get_exa_client directly). from plugins.web.parallel.provider import ( # noqa: F401 — backward-compat names _get_async_parallel_client, _get_parallel_client, ) from plugins.web.exa.provider import _get_exa_client # noqa: F401 # Module-level cache slots for the per-vendor clients. The plugins read/write # these via tools.web_tools so unit tests that reset # ``tools.web_tools.__client = None`` between cases keep working. _firecrawl_client: Optional[Any] = None _firecrawl_client_config: Optional[Any] = None _parallel_client: Optional[Any] = None _async_parallel_client: Optional[Any] = None _exa_client: Optional[Any] = None from agent.auxiliary_client import ( async_call_llm, extract_content_or_reasoning, get_async_text_auxiliary_client, ) from tools.debug_helpers import DebugSession # Imported solely so unit tests can monkeypatch these names on # tools.web_tools (the firecrawl plugin reads them via its own import chain). from tools.managed_tool_gateway import ( # noqa: F401 — backward-compat names for tests build_vendor_gateway_url, read_nous_access_token as _read_nous_access_token, resolve_managed_tool_gateway, ) from tools.tool_backend_helpers import managed_nous_tools_enabled, prefers_gateway # noqa: F401 from tools.url_safety import is_safe_url from tools.website_policy import check_website_access import sys logger = logging.getLogger(__name__) # ─── Backend Selection ──────────────────────────────────────────────────────── def _has_env(name: str) -> bool: val = os.getenv(name) return bool(val and val.strip()) def _load_web_config() -> dict: """Load the ``web:`` section from ~/.hermes/config.yaml.""" try: from hermes_cli.config import load_config return load_config().get("web", {}) except (ImportError, Exception): return {} def _get_backend() -> str: """Determine which web backend to use (shared fallback). Reads ``web.backend`` from config.yaml (set by ``hermes tools``). Falls back to whichever API key is present for users who configured keys manually without running setup. """ configured = (_load_web_config().get("backend") or "").lower().strip() if configured in {"parallel", "firecrawl", "tavily", "exa", "searxng", "brave-free", "ddgs"}: return configured # Fallback for manual / legacy config — pick the highest-priority # available backend. Firecrawl also counts as available when the managed # tool gateway is configured for Nous subscribers. # Free-tier backends (searxng / brave-free / ddgs) trail the paid ones so # existing paid setups are unaffected. backend_candidates = ( ("firecrawl", _has_env("FIRECRAWL_API_KEY") or _has_env("FIRECRAWL_API_URL") or _is_tool_gateway_ready()), ("parallel", _has_env("PARALLEL_API_KEY")), ("tavily", _has_env("TAVILY_API_KEY")), ("exa", _has_env("EXA_API_KEY")), ("searxng", _has_env("SEARXNG_URL")), ("brave-free", _has_env("BRAVE_SEARCH_API_KEY")), ("ddgs", _ddgs_package_importable()), ) for backend, available in backend_candidates: if available: return backend return "firecrawl" # default (backward compat) def _get_search_backend() -> str: """Determine which backend to use for web_search specifically. Selection priority: 1. ``web.search_backend`` (per-capability override) 2. ``web.backend`` (shared fallback — existing behavior) 3. Auto-detect from env vars This enables using different providers for search vs extract (e.g. SearXNG for search + Firecrawl for extract). """ return _get_capability_backend("search") def _get_extract_backend() -> str: """Determine which backend to use for web_extract specifically. Selection priority: 1. ``web.extract_backend`` (per-capability override) 2. ``web.backend`` (shared fallback — existing behavior) 3. Auto-detect from env vars """ return _get_capability_backend("extract") def _get_capability_backend(capability: str) -> str: """Shared helper for per-capability backend selection. Reads ``web.{capability}_backend`` from config; if set and available, uses it. Otherwise falls through to the shared ``_get_backend()``. """ cfg = _load_web_config() specific = (cfg.get(f"{capability}_backend") or "").lower().strip() if specific and _is_backend_available(specific): return specific return _get_backend() def _is_backend_available(backend: str) -> bool: """Return True when the selected backend is currently usable.""" if backend == "exa": return _has_env("EXA_API_KEY") if backend == "parallel": return _has_env("PARALLEL_API_KEY") if backend == "firecrawl": return check_firecrawl_api_key() if backend == "tavily": return _has_env("TAVILY_API_KEY") if backend == "searxng": return _has_env("SEARXNG_URL") if backend == "brave-free": return _has_env("BRAVE_SEARCH_API_KEY") if backend == "ddgs": return _ddgs_package_importable() return False def _ddgs_package_importable() -> bool: """Return True when the ``ddgs`` Python package can be imported. ddgs is the only backend whose availability is driven by a package presence rather than an env var / config entry. Wrapped in a helper so auto-detect and ``_is_backend_available`` share the same check (and tests can monkeypatch a single symbol). """ try: import ddgs # noqa: F401 return True except ImportError: return False # ─── Firecrawl Client ──────────────────────────────────────────────────────── # ─── Firecrawl Client ──────────────────────────────────────────────────────── # After PR #25182, the firecrawl client, lazy SDK proxy, dual-auth config # resolution, response normalizers, and check_firecrawl_api_key() all live # in plugins.web.firecrawl.provider and are re-exported at the top of this # module so external callers (integration tests, tool-registry gating) and # unit tests that patch tools.web_tools. continue to work. def _web_requires_env() -> list[str]: """Return tool metadata env vars for the currently enabled web backends. The gateway env vars are always reported — they're metadata strings used by the tool registry to light up the tool when the variable is set. Gating them on ``managed_nous_tools_enabled()`` only saved string noise in the metadata list, but cost a synchronous HTTP refresh against the Nous portal on every CLI startup (invoked at tool-registration time). The behavioral contract is: if the env var is set, the tool sees it; if not, it doesn't. Not-logged-in users simply don't have the vars set, so the extra entries are harmless. """ return [ "EXA_API_KEY", "PARALLEL_API_KEY", "TAVILY_API_KEY", "FIRECRAWL_API_KEY", "FIRECRAWL_API_URL", "FIRECRAWL_GATEWAY_URL", "TOOL_GATEWAY_DOMAIN", "TOOL_GATEWAY_SCHEME", "TOOL_GATEWAY_USER_TOKEN", ] # ─── Parallel / Tavily / Firecrawl helpers — moved into plugins ────────────── # After PR #25182, the per-vendor client construction, request helpers, and # response normalizers all live in plugins.web..provider: # - parallel: plugins/web/parallel/provider.py # - tavily: plugins/web/tavily/provider.py # - firecrawl: plugins/web/firecrawl/provider.py # The names from the firecrawl plugin (Firecrawl proxy, _get_firecrawl_client, # _to_plain_object, _normalize_result_list, _extract_web_search_results, # _extract_scrape_payload, _is_tool_gateway_ready, etc.) are re-exported at # the top of this module for backward-compat with integration tests and # unit-test patches. DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION = 5000 def _is_nous_auxiliary_client(client: Any) -> bool: """Return True when the resolved auxiliary backend is Nous Portal.""" from urllib.parse import urlparse base_url = str(getattr(client, "base_url", "") or "") host = (urlparse(base_url).hostname or "").lower() return host == "nousresearch.com" or host.endswith(".nousresearch.com") def _resolve_web_extract_auxiliary(model: Optional[str] = None) -> tuple[Optional[Any], Optional[str], Dict[str, Any]]: """Resolve the current web-extract auxiliary client, model, and extra body.""" client, default_model = get_async_text_auxiliary_client("web_extract") configured_model = os.getenv("AUXILIARY_WEB_EXTRACT_MODEL", "").strip() effective_model = model or configured_model or default_model extra_body: Dict[str, Any] = {} if client is not None and _is_nous_auxiliary_client(client): from agent.auxiliary_client import get_auxiliary_extra_body from agent.portal_tags import nous_portal_tags extra_body = get_auxiliary_extra_body() or {"tags": nous_portal_tags()} return client, effective_model, extra_body def _get_default_summarizer_model() -> Optional[str]: """Return the current default model for web extraction summarization.""" _, model, _ = _resolve_web_extract_auxiliary() return model _debug = DebugSession("web_tools", env_var="WEB_TOOLS_DEBUG") async def process_content_with_llm( content: str, url: str = "", title: str = "", model: Optional[str] = None, min_length: int = DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION ) -> Optional[str]: """ Process web content using LLM to create intelligent summaries with key excerpts. This function uses Gemini 3 Flash Preview (or specified model) via OpenRouter API to intelligently extract key information and create markdown summaries, significantly reducing token usage while preserving all important information. For very large content (>500k chars), uses chunked processing with synthesis. For extremely large content (>2M chars), refuses to process entirely. Args: content (str): The raw content to process url (str): The source URL (for context, optional) title (str): The page title (for context, optional) model (str): The model to use for processing (default: google/gemini-3-flash-preview) min_length (int): Minimum content length to trigger processing (default: 5000) Returns: Optional[str]: Processed markdown content, or None if content too short or processing fails """ # Size thresholds MAX_CONTENT_SIZE = 2_000_000 # 2M chars - refuse entirely above this CHUNK_THRESHOLD = 500_000 # 500k chars - use chunked processing above this CHUNK_SIZE = 100_000 # 100k chars per chunk MAX_OUTPUT_SIZE = 5000 # Hard cap on final output size try: content_len = len(content) # Refuse if content is absurdly large if content_len > MAX_CONTENT_SIZE: size_mb = content_len / 1_000_000 logger.warning("Content too large (%.1fMB > 2MB limit). Refusing to process.", size_mb) return f"[Content too large to process: {size_mb:.1f}MB. Try using web_crawl with specific extraction instructions, or search for a more focused source.]" # Skip processing if content is too short if content_len < min_length: logger.debug("Content too short (%d < %d chars), skipping LLM processing", content_len, min_length) return None # Create context information context_info = [] if title: context_info.append(f"Title: {title}") if url: context_info.append(f"Source: {url}") context_str = "\n".join(context_info) + "\n\n" if context_info else "" # Check if we need chunked processing if content_len > CHUNK_THRESHOLD: logger.info("Content large (%d chars). Using chunked processing...", content_len) return await _process_large_content_chunked( content, context_str, model, CHUNK_SIZE, MAX_OUTPUT_SIZE ) # Standard single-pass processing for normal content logger.info("Processing content with LLM (%d characters)", content_len) processed_content = await _call_summarizer_llm(content, context_str, model) if processed_content: # Enforce output cap if len(processed_content) > MAX_OUTPUT_SIZE: processed_content = processed_content[:MAX_OUTPUT_SIZE] + "\n\n[... summary truncated for context management ...]" # Log compression metrics processed_length = len(processed_content) compression_ratio = processed_length / content_len if content_len > 0 else 1.0 logger.info("Content processed: %d -> %d chars (%.1f%%)", content_len, processed_length, compression_ratio * 100) return processed_content except Exception as e: logger.warning( "web_extract LLM summarization failed (%s). " "Tip: increase auxiliary.web_extract.timeout in config.yaml " "or switch to a faster auxiliary model.", str(e)[:120], ) # Fall back to truncated raw content instead of returning a useless # error message. The first ~5000 chars are almost always more useful # to the model than "[Failed to process content: ...]". truncated = content[:MAX_OUTPUT_SIZE] if len(content) > MAX_OUTPUT_SIZE: truncated += ( f"\n\n[Content truncated — showing first {MAX_OUTPUT_SIZE:,} of " f"{len(content):,} chars. LLM summarization timed out. " f"To fix: increase auxiliary.web_extract.timeout in config.yaml, " f"or use a faster auxiliary model. Use browser_navigate for the full page.]" ) return truncated async def _call_summarizer_llm( content: str, context_str: str, model: Optional[str], max_tokens: int = 20000, is_chunk: bool = False, chunk_info: str = "" ) -> Optional[str]: """ Make a single LLM call to summarize content. Args: content: The content to summarize context_str: Context information (title, URL) model: Model to use max_tokens: Maximum output tokens is_chunk: Whether this is a chunk of a larger document chunk_info: Information about chunk position (e.g., "Chunk 2/5") Returns: Summarized content or None on failure """ if is_chunk: # Chunk-specific prompt - aware that this is partial content system_prompt = """You are an expert content analyst processing a SECTION of a larger document. Your job is to extract and summarize the key information from THIS SECTION ONLY. Important guidelines for chunk processing: 1. Do NOT write introductions or conclusions - this is a partial document 2. Focus on extracting ALL key facts, figures, data points, and insights from this section 3. Preserve important quotes, code snippets, and specific details verbatim 4. Use bullet points and structured formatting for easy synthesis later 5. Note any references to other sections (e.g., "as mentioned earlier", "see below") without trying to resolve them Your output will be combined with summaries of other sections, so focus on thorough extraction rather than narrative flow.""" user_prompt = f"""Extract key information from this SECTION of a larger document: {context_str}{chunk_info} SECTION CONTENT: {content} Extract all important information from this section in a structured format. Focus on facts, data, insights, and key details. Do not add introductions or conclusions.""" else: # Standard full-document prompt system_prompt = """You are an expert content analyst. Your job is to process web content and create a comprehensive yet concise summary that preserves all important information while dramatically reducing bulk. Create a well-structured markdown summary that includes: 1. Key excerpts (quotes, code snippets, important facts) in their original format 2. Comprehensive summary of all other important information 3. Proper markdown formatting with headers, bullets, and emphasis Your goal is to preserve ALL important information while reducing length. Never lose key facts, figures, insights, or actionable information. Make it scannable and well-organized.""" user_prompt = f"""Please process this web content and create a comprehensive markdown summary: {context_str}CONTENT TO PROCESS: {content} Create a markdown summary that captures all key information in a well-organized, scannable format. Include important quotes and code snippets in their original formatting. Focus on actionable information, specific details, and unique insights.""" # Call the LLM with retry logic — keep retries low since summarization # is a nice-to-have; the caller falls back to truncated content on failure. max_retries = 2 retry_delay = 2 last_error = None for attempt in range(max_retries): try: aux_client, effective_model, extra_body = _resolve_web_extract_auxiliary(model) if aux_client is None or not effective_model: logger.warning("No auxiliary model available for web content processing") return None call_kwargs = { "task": "web_extract", "model": effective_model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], "temperature": 0.1, "max_tokens": max_tokens, # No explicit timeout — async_call_llm reads auxiliary.web_extract.timeout # from config.yaml. Fresh configs ship with 360s; if the key is absent # the runtime default is 30s (_DEFAULT_AUX_TIMEOUT in # agent/auxiliary_client.py). Users with slow local models should set # or increase auxiliary.web_extract.timeout in config.yaml. } if extra_body: call_kwargs["extra_body"] = extra_body response = await async_call_llm(**call_kwargs) content = extract_content_or_reasoning(response) if content: return content # Reasoning-only / empty response — let the retry loop handle it logger.warning("LLM returned empty content (attempt %d/%d), retrying", attempt + 1, max_retries) if attempt < max_retries - 1: await asyncio.sleep(retry_delay) retry_delay = min(retry_delay * 2, 60) continue return content # Return whatever we got after exhausting retries except RuntimeError: logger.warning("No auxiliary model available for web content processing") return None except Exception as api_error: last_error = api_error if attempt < max_retries - 1: logger.warning("LLM API call failed (attempt %d/%d): %s", attempt + 1, max_retries, str(api_error)[:100]) logger.warning("Retrying in %ds...", retry_delay) await asyncio.sleep(retry_delay) retry_delay = min(retry_delay * 2, 60) else: raise last_error return None async def _process_large_content_chunked( content: str, context_str: str, model: Optional[str], chunk_size: int, max_output_size: int ) -> Optional[str]: """ Process large content by chunking, summarizing each chunk in parallel, then synthesizing the summaries. Args: content: The large content to process context_str: Context information model: Model to use chunk_size: Size of each chunk in characters max_output_size: Maximum final output size Returns: Synthesized summary or None on failure """ # Split content into chunks chunks = [] for i in range(0, len(content), chunk_size): chunk = content[i:i + chunk_size] chunks.append(chunk) logger.info("Split into %d chunks of ~%d chars each", len(chunks), chunk_size) # Summarize each chunk in parallel async def summarize_chunk(chunk_idx: int, chunk_content: str) -> tuple[int, Optional[str]]: """Summarize a single chunk.""" try: chunk_info = f"[Processing chunk {chunk_idx + 1} of {len(chunks)}]" summary = await _call_summarizer_llm( chunk_content, context_str, model, max_tokens=10000, is_chunk=True, chunk_info=chunk_info ) if summary: logger.info("Chunk %d/%d summarized: %d -> %d chars", chunk_idx + 1, len(chunks), len(chunk_content), len(summary)) return chunk_idx, summary except Exception as e: logger.warning("Chunk %d/%d failed: %s", chunk_idx + 1, len(chunks), str(e)[:50]) return chunk_idx, None # Run all chunk summarizations in parallel tasks = [summarize_chunk(i, chunk) for i, chunk in enumerate(chunks)] results = await asyncio.gather(*tasks) # Collect successful summaries in order summaries = [] for chunk_idx, summary in sorted(results, key=lambda x: x[0]): if summary: summaries.append(f"## Section {chunk_idx + 1}\n{summary}") if not summaries: logger.debug("All chunk summarizations failed") return "[Failed to process large content: all chunk summarizations failed]" logger.info("Got %d/%d chunk summaries", len(summaries), len(chunks)) # If only one chunk succeeded, just return it (with cap) if len(summaries) == 1: result = summaries[0] if len(result) > max_output_size: result = result[:max_output_size] + "\n\n[... truncated ...]" return result # Synthesize the summaries into a final summary logger.info("Synthesizing %d summaries...", len(summaries)) combined_summaries = "\n\n---\n\n".join(summaries) synthesis_prompt = f"""You have been given summaries of different sections of a large document. Synthesize these into ONE cohesive, comprehensive summary that: 1. Removes redundancy between sections 2. Preserves all key facts, figures, and actionable information 3. Is well-organized with clear structure 4. Is under {max_output_size} characters {context_str}SECTION SUMMARIES: {combined_summaries} Create a single, unified markdown summary.""" try: aux_client, effective_model, extra_body = _resolve_web_extract_auxiliary(model) if aux_client is None or not effective_model: logger.warning("No auxiliary model for synthesis, concatenating summaries") fallback = "\n\n".join(summaries) if len(fallback) > max_output_size: fallback = fallback[:max_output_size] + "\n\n[... truncated ...]" return fallback call_kwargs = { "task": "web_extract", "model": effective_model, "messages": [ {"role": "system", "content": "You synthesize multiple summaries into one cohesive, comprehensive summary. Be thorough but concise."}, {"role": "user", "content": synthesis_prompt}, ], "temperature": 0.1, "max_tokens": 20000, } if extra_body: call_kwargs["extra_body"] = extra_body response = await async_call_llm(**call_kwargs) final_summary = extract_content_or_reasoning(response) # Retry once on empty content (reasoning-only response) if not final_summary: logger.warning("Synthesis LLM returned empty content, retrying once") response = await async_call_llm(**call_kwargs) final_summary = extract_content_or_reasoning(response) # If still None after retry, fall back to concatenated summaries if not final_summary: logger.warning("Synthesis failed after retry — concatenating chunk summaries") fallback = "\n\n".join(summaries) if len(fallback) > max_output_size: fallback = fallback[:max_output_size] + "\n\n[... truncated ...]" return fallback # Enforce hard cap if len(final_summary) > max_output_size: final_summary = final_summary[:max_output_size] + "\n\n[... summary truncated for context management ...]" original_len = len(content) final_len = len(final_summary) compression = final_len / original_len if original_len > 0 else 1.0 logger.info("Synthesis complete: %d -> %d chars (%.2f%%)", original_len, final_len, compression * 100) return final_summary except Exception as e: logger.warning("Synthesis failed: %s", str(e)[:100]) # Fall back to concatenated summaries with truncation fallback = "\n\n".join(summaries) if len(fallback) > max_output_size: fallback = fallback[:max_output_size] + "\n\n[... truncated due to synthesis failure ...]" return fallback def clean_base64_images(text: str) -> str: """ Remove base64 encoded images from text to reduce token count and clutter. This function finds and removes base64 encoded images in various formats: - (data:image/png;base64,...) - (data:image/jpeg;base64,...) - (data:image/svg+xml;base64,...) - data:image/[type];base64,... (without parentheses) Args: text: The text content to clean Returns: Cleaned text with base64 images replaced with placeholders """ # Pattern to match base64 encoded images wrapped in parentheses # Matches: (data:image/[type];base64,[base64-string]) base64_with_parens_pattern = r'\(data:image/[^;]+;base64,[A-Za-z0-9+/=]+\)' # Pattern to match base64 encoded images without parentheses # Matches: data:image/[type];base64,[base64-string] base64_pattern = r'data:image/[^;]+;base64,[A-Za-z0-9+/=]+' # Replace parentheses-wrapped images first cleaned_text = re.sub(base64_with_parens_pattern, '[BASE64_IMAGE_REMOVED]', text) # Then replace any remaining non-parentheses images cleaned_text = re.sub(base64_pattern, '[BASE64_IMAGE_REMOVED]', cleaned_text) return cleaned_text # ─── Exa / Parallel inline helpers — moved into plugins ────────────────────── # After PR #25182, the exa client + search/extract and parallel client + # search/extract helpers all live in their respective plugins: # - plugins/web/exa/provider.py # - plugins/web/parallel/provider.py # Both plugins register through agent.web_search_registry and the # dispatchers in this file resolve them via get_active_*_provider(). def web_search_tool(query: str, limit: int = 5) -> str: """ Search the web for information using available search API backend. This function provides a generic interface for web search that can work with multiple backends (Parallel or Firecrawl). Note: This function returns search result metadata only (URLs, titles, descriptions). Use web_extract_tool to get full content from specific URLs. Args: query (str): The search query to look up limit (int): Maximum number of results to return (default: 5) Returns: str: JSON string containing search results with the following structure: { "success": bool, "data": { "web": [ { "title": str, "url": str, "description": str, "position": int }, ... ] } } Raises: Exception: If search fails or API key is not set """ try: limit = int(limit) except (TypeError, ValueError): limit = 5 limit = min(max(limit, 1), 100) debug_call_data = { "parameters": { "query": query, "limit": limit }, "error": None, "results_count": 0, "original_response_size": 0, "final_response_size": 0 } try: from tools.interrupt import is_interrupted if is_interrupted(): return tool_error("Interrupted", success=False) # Dispatch through the web search registry. All 7 providers # (brave-free, ddgs, searxng, exa, parallel, tavily, firecrawl) # now live as plugins; the dispatcher is just a registry lookup + # delegation. Sync only — every provider's search() is sync. from agent.web_search_registry import ( get_active_search_provider, get_provider as _wsp_get_provider, ) backend = _get_search_backend() provider = _wsp_get_provider(backend) if backend else None if provider is None or not provider.supports_search(): # Fall back to availability-walked active provider when the # configured backend isn't a registered search provider (typo, # uninstalled plugin, or capability mismatch). provider = get_active_search_provider() if provider is None: response_data = { "success": False, "error": ( "No web search provider configured. " "Run `hermes tools` to set one up." ), } else: logger.info( "Web search via %s: '%s' (limit: %d)", provider.name, query, limit, ) response_data = provider.search(query, limit) debug_call_data["results_count"] = len(response_data.get("data", {}).get("web", [])) result_json = json.dumps(response_data, indent=2, ensure_ascii=False) debug_call_data["final_response_size"] = len(result_json) _debug.log_call("web_search_tool", debug_call_data) _debug.save() return result_json except Exception as e: error_msg = f"Error searching web: {str(e)}" logger.debug("%s", error_msg) debug_call_data["error"] = error_msg _debug.log_call("web_search_tool", debug_call_data) _debug.save() return tool_error(error_msg) async def web_extract_tool( urls: List[str], format: str = None, use_llm_processing: bool = True, model: Optional[str] = None, min_length: int = DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION ) -> str: """ Extract content from specific web pages using available extraction API backend. This function provides a generic interface for web content extraction that can work with multiple backends. Currently uses Firecrawl. Args: urls (List[str]): List of URLs to extract content from format (str): Desired output format ("markdown" or "html", optional) use_llm_processing (bool): Whether to process content with LLM for summarization (default: True) model (Optional[str]): The model to use for LLM processing (defaults to current auxiliary backend model) min_length (int): Minimum content length to trigger LLM processing (default: 5000) Security: URLs are checked for embedded secrets before fetching. Returns: str: JSON string containing extracted content. If LLM processing is enabled and successful, the 'content' field will contain the processed markdown summary instead of raw content. Raises: Exception: If extraction fails or API key is not set """ # Block URLs containing embedded secrets (exfiltration prevention). # URL-decode first so percent-encoded secrets (%73k- = sk-) are caught. from agent.redact import _PREFIX_RE from urllib.parse import unquote for _url in urls: if _PREFIX_RE.search(_url) or _PREFIX_RE.search(unquote(_url)): return json.dumps({ "success": False, "error": "Blocked: URL contains what appears to be an API key or token. " "Secrets must not be sent in URLs.", }) debug_call_data = { "parameters": { "urls": urls, "format": format, "use_llm_processing": use_llm_processing, "model": model, "min_length": min_length }, "error": None, "pages_extracted": 0, "pages_processed_with_llm": 0, "original_response_size": 0, "final_response_size": 0, "compression_metrics": [], "processing_applied": [] } try: logger.info("Extracting content from %d URL(s)", len(urls)) # ── SSRF protection — filter out private/internal URLs before any backend ── safe_urls = [] ssrf_blocked: List[Dict[str, Any]] = [] for url in urls: if not is_safe_url(url): ssrf_blocked.append({ "url": url, "title": "", "content": "", "error": "Blocked: URL targets a private or internal network address", }) else: safe_urls.append(url) # Dispatch only safe URLs to the configured backend if not safe_urls: results = [] else: backend = _get_extract_backend() # All seven providers (brave-free, ddgs, searxng, exa, parallel, # tavily, firecrawl) now live as plugins. The dispatcher is a # registry lookup + delegation. Some providers' extract() is # async (parallel, firecrawl), others sync (exa, tavily) — we # detect coroutine functions and await; sync functions run # inline (the policy gate, SSRF re-check, etc. live inside the # provider itself for the firecrawl per-URL loop). from agent.web_search_registry import ( get_active_extract_provider, get_provider as _wsp_get_provider, ) provider = _wsp_get_provider(backend) if backend else None if provider is None or not provider.supports_extract(): # When the configured name IS registered but doesn't support # extract (search-only providers like brave-free / ddgs / # searxng), surface that as a typed "search-only" error # rather than silently switching backends. When the name # isn't registered at all (typo / uninstalled plugin), fall # through to the active-provider walk. if provider is not None and not provider.supports_extract(): return json.dumps( { "success": False, "error": ( f"{provider.display_name} is a search-only " "backend and cannot extract URL content. " "Set web.extract_backend to firecrawl, " "tavily, exa, or parallel." ), }, ensure_ascii=False, ) provider = get_active_extract_provider() if provider is None: return json.dumps( { "success": False, "error": ( "No web extract provider configured. " "Set web.extract_backend to firecrawl, " "tavily, exa, or parallel." ), }, ensure_ascii=False, ) logger.info( "Web extract via %s: %d URL(s)", provider.name, len(safe_urls) ) # Async-or-sync dispatch: parallel + firecrawl have async # extract(); exa + tavily are sync. import inspect if inspect.iscoroutinefunction(provider.extract): results = await provider.extract(safe_urls, format=format) else: # Run sync extract() in a thread so we don't block the # event loop on network I/O. results = await asyncio.to_thread( provider.extract, safe_urls, format=format ) # Merge any SSRF-blocked results back in if ssrf_blocked: results = ssrf_blocked + results response = {"results": results} pages_extracted = len(response.get('results', [])) logger.info("Extracted content from %d pages", pages_extracted) debug_call_data["pages_extracted"] = pages_extracted debug_call_data["original_response_size"] = len(json.dumps(response)) effective_model = model or _get_default_summarizer_model() auxiliary_available = check_auxiliary_model() # Process each result with LLM if enabled if use_llm_processing and auxiliary_available: logger.info("Processing extracted content with LLM (parallel)...") debug_call_data["processing_applied"].append("llm_processing") # Prepare tasks for parallel processing async def process_single_result(result): """Process a single result with LLM and return updated result with metrics.""" url = result.get('url', 'Unknown URL') title = result.get('title', '') raw_content = result.get('raw_content', '') or result.get('content', '') if not raw_content: return result, None, "no_content" original_size = len(raw_content) # Process content with LLM processed = await process_content_with_llm( raw_content, url, title, effective_model, min_length ) if processed: processed_size = len(processed) compression_ratio = processed_size / original_size if original_size > 0 else 1.0 # Update result with processed content result['content'] = processed result['raw_content'] = raw_content metrics = { "url": url, "original_size": original_size, "processed_size": processed_size, "compression_ratio": compression_ratio, "model_used": effective_model } return result, metrics, "processed" else: metrics = { "url": url, "original_size": original_size, "processed_size": original_size, "compression_ratio": 1.0, "model_used": None, "reason": "content_too_short" } return result, metrics, "too_short" # Run all LLM processing in parallel results_list = response.get('results', []) tasks = [process_single_result(result) for result in results_list] processed_results = await asyncio.gather(*tasks) # Collect metrics and print results for result, metrics, status in processed_results: url = result.get('url', 'Unknown URL') if status == "processed": debug_call_data["compression_metrics"].append(metrics) debug_call_data["pages_processed_with_llm"] += 1 logger.info("%s (processed)", url) elif status == "too_short": debug_call_data["compression_metrics"].append(metrics) logger.info("%s (no processing - content too short)", url) else: logger.warning("%s (no content to process)", url) else: if use_llm_processing and not auxiliary_available: logger.warning("LLM processing requested but no auxiliary model available, returning raw content") debug_call_data["processing_applied"].append("llm_processing_unavailable") # Print summary of extracted pages for debugging (original behavior) for result in response.get('results', []): url = result.get('url', 'Unknown URL') content_length = len(result.get('raw_content', '')) logger.info("%s (%d characters)", url, content_length) # Trim output to minimal fields per entry: title, content, error trimmed_results = [ { "url": r.get("url", ""), "title": r.get("title", ""), "content": r.get("content", ""), "error": r.get("error"), **({ "blocked_by_policy": r["blocked_by_policy"]} if "blocked_by_policy" in r else {}), } for r in response.get("results", []) ] trimmed_response = {"results": trimmed_results} if trimmed_response.get("results") == []: result_json = tool_error("Content was inaccessible or not found") cleaned_result = clean_base64_images(result_json) else: result_json = json.dumps(trimmed_response, indent=2, ensure_ascii=False) cleaned_result = clean_base64_images(result_json) debug_call_data["final_response_size"] = len(cleaned_result) debug_call_data["processing_applied"].append("base64_image_removal") # Log debug information _debug.log_call("web_extract_tool", debug_call_data) _debug.save() return cleaned_result except Exception as e: error_msg = f"Error extracting content: {str(e)}" logger.debug("%s", error_msg) debug_call_data["error"] = error_msg _debug.log_call("web_extract_tool", debug_call_data) _debug.save() return tool_error(error_msg) async def web_crawl_tool( url: str, instructions: str = None, depth: str = "basic", use_llm_processing: bool = True, model: Optional[str] = None, min_length: int = DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION ) -> str: """ Crawl a website with specific instructions using available crawling API backend. This function provides a generic interface for web crawling that can work with multiple backends. Currently uses Firecrawl. Args: url (str): The base URL to crawl (can include or exclude https://) instructions (str): Instructions for what to crawl/extract using LLM intelligence (optional) depth (str): Depth of extraction ("basic" or "advanced", default: "basic") use_llm_processing (bool): Whether to process content with LLM for summarization (default: True) model (Optional[str]): The model to use for LLM processing (defaults to current auxiliary backend model) min_length (int): Minimum content length to trigger LLM processing (default: 5000) Returns: str: JSON string containing crawled content. If LLM processing is enabled and successful, the 'content' field will contain the processed markdown summary instead of raw content. Each page is processed individually. Raises: Exception: If crawling fails or API key is not set """ debug_call_data = { "parameters": { "url": url, "instructions": instructions, "depth": depth, "use_llm_processing": use_llm_processing, "model": model, "min_length": min_length }, "error": None, "pages_crawled": 0, "pages_processed_with_llm": 0, "original_response_size": 0, "final_response_size": 0, "compression_metrics": [], "processing_applied": [] } try: effective_model = model or _get_default_summarizer_model() auxiliary_available = check_auxiliary_model() backend = _get_backend() # Tavily (and any future plugin advertising supports_crawl=True) # dispatches through agent.web_search_registry. The crawl response # shape — {"results": [{"url", "title", "content", ...}]} — is then # post-processed by the shared LLM-summarization path below. from agent.web_search_registry import ( get_active_crawl_provider, get_provider as _wsp_get_provider, ) crawl_provider = _wsp_get_provider(backend) if backend else None if crawl_provider is not None and not crawl_provider.supports_crawl(): # When the configured provider is search-only AND cannot # extract URLs either (brave-free / ddgs / searxng), surface a # typed "search-only" error rather than silently switching to # a different crawl backend. When the provider supports extract # but not crawl (e.g. firecrawl), fall through to the legacy # firecrawl-via-extract path below. if not crawl_provider.supports_extract(): return json.dumps( { "success": False, "error": ( f"{crawl_provider.display_name} is a search-only " "backend and cannot crawl URLs. " "Set FIRECRAWL_API_KEY for crawling, or use " "web_search instead." ), }, ensure_ascii=False, ) crawl_provider = None # let legacy firecrawl path handle it if crawl_provider is None: crawl_provider = get_active_crawl_provider() # Mirror main's upstream availability gate: when the resolved # provider is configured-but-unavailable (e.g. firecrawl without # FIRECRAWL_API_KEY), short-circuit BEFORE we dispatch so the # error envelope matches the legacy top-level shape # ``{"success": False, "error": "..."}`` rather than burying the # configuration message inside a per-page ``results[]`` entry. if crawl_provider is not None and not crawl_provider.is_available(): return json.dumps( { "success": False, "error": ( "web_crawl requires Firecrawl. Set FIRECRAWL_API_KEY, " f"FIRECRAWL_API_URL{_firecrawl_backend_help_suffix()}, " "or use web_search + web_extract instead." ), }, ensure_ascii=False, ) if crawl_provider is not None: # Ensure URL has protocol if not url.startswith(('http://', 'https://')): url = f'https://{url}' # SSRF protection — block private/internal addresses if not is_safe_url(url): return json.dumps({"results": [{"url": url, "title": "", "content": "", "error": "Blocked: URL targets a private or internal network address"}]}, ensure_ascii=False) # Website policy check blocked = check_website_access(url) if blocked: logger.info("Blocked web_crawl for %s by rule %s", blocked["host"], blocked["rule"]) return json.dumps({"results": [{"url": url, "title": "", "content": "", "error": blocked["message"], "blocked_by_policy": {"host": blocked["host"], "rule": blocked["rule"], "source": blocked["source"]}}]}, ensure_ascii=False) from tools.interrupt import is_interrupted as _is_int if _is_int(): return tool_error("Interrupted", success=False) logger.info("Web crawl via %s: %s", crawl_provider.name, url) # Async-or-sync dispatch — Tavily's crawl is sync, but a future # async-crawl provider works transparently. import inspect crawl_kwargs = {"depth": depth, "limit": 20} if instructions: crawl_kwargs["instructions"] = instructions if inspect.iscoroutinefunction(crawl_provider.crawl): response = await crawl_provider.crawl(url, **crawl_kwargs) else: response = await asyncio.to_thread( crawl_provider.crawl, url, **crawl_kwargs ) # Provider returns {"results": [...]} matching what the shared # LLM post-processing below expects. if not isinstance(response, dict): response = {"results": []} response.setdefault("results", []) # Fall through to the shared LLM processing and trimming below # (skip the Firecrawl-specific crawl logic) pages_crawled = len(response.get('results', [])) logger.info("Crawled %d pages", pages_crawled) debug_call_data["pages_crawled"] = pages_crawled debug_call_data["original_response_size"] = len(json.dumps(response)) # Process each result with LLM if enabled if use_llm_processing and auxiliary_available: logger.info("Processing crawled content with LLM (parallel)...") debug_call_data["processing_applied"].append("llm_processing") async def _process_tavily_crawl(result): page_url = result.get('url', 'Unknown URL') title = result.get('title', '') content = result.get('content', '') if not content: return result, None, "no_content" original_size = len(content) processed = await process_content_with_llm(content, page_url, title, effective_model, min_length) if processed: result['raw_content'] = content result['content'] = processed metrics = {"url": page_url, "original_size": original_size, "processed_size": len(processed), "compression_ratio": len(processed) / original_size if original_size else 1.0, "model_used": effective_model} return result, metrics, "processed" metrics = {"url": page_url, "original_size": original_size, "processed_size": original_size, "compression_ratio": 1.0, "model_used": None, "reason": "content_too_short"} return result, metrics, "too_short" tasks = [_process_tavily_crawl(r) for r in response.get('results', [])] processed_results = await asyncio.gather(*tasks) for result, metrics, status in processed_results: if status == "processed": debug_call_data["compression_metrics"].append(metrics) debug_call_data["pages_processed_with_llm"] += 1 if use_llm_processing and not auxiliary_available: logger.warning("LLM processing requested but no auxiliary model available, returning raw content") debug_call_data["processing_applied"].append("llm_processing_unavailable") trimmed_results = [{"url": r.get("url", ""), "title": r.get("title", ""), "content": r.get("content", ""), "error": r.get("error"), **({ "blocked_by_policy": r["blocked_by_policy"]} if "blocked_by_policy" in r else {})} for r in response.get("results", [])] result_json = json.dumps({"results": trimmed_results}, indent=2, ensure_ascii=False) cleaned_result = clean_base64_images(result_json) debug_call_data["final_response_size"] = len(cleaned_result) _debug.log_call("web_crawl_tool", debug_call_data) _debug.save() return cleaned_result # No registered provider supports crawl AND no crawl-capable plugin # is available. Surface a typed error pointing the user at the two # crawl-capable providers (Firecrawl + Tavily). return json.dumps( { "success": False, "error": ( "web_crawl has no available backend. " "Set FIRECRAWL_API_KEY (or FIRECRAWL_API_URL for " f"self-hosted){_firecrawl_backend_help_suffix()}, " "or set TAVILY_API_KEY for Tavily. " "Alternatively use web_search + web_extract instead." ), }, ensure_ascii=False, ) except Exception as e: error_msg = f"Error crawling website: {str(e)}" logger.debug("%s", error_msg) debug_call_data["error"] = error_msg _debug.log_call("web_crawl_tool", debug_call_data) _debug.save() return tool_error(error_msg) # Convenience function to check Firecrawl credentials def check_web_api_key() -> bool: """Check whether the configured web backend is available.""" configured = _load_web_config().get("backend", "").lower().strip() if configured in {"exa", "parallel", "firecrawl", "tavily", "searxng", "brave-free", "ddgs"}: return _is_backend_available(configured) return any( _is_backend_available(backend) for backend in ("exa", "parallel", "firecrawl", "tavily", "searxng", "brave-free", "ddgs") ) def check_auxiliary_model() -> bool: """Check if an auxiliary text model is available for LLM content processing.""" client, _, _ = _resolve_web_extract_auxiliary() return client is not None if __name__ == "__main__": """ Simple test/demo when run directly """ print("🌐 Standalone Web Tools Module") print("=" * 40) # Check if API keys are available web_available = check_web_api_key() tool_gateway_available = _is_tool_gateway_ready() firecrawl_key_available = bool(os.getenv("FIRECRAWL_API_KEY", "").strip()) firecrawl_url_available = bool(os.getenv("FIRECRAWL_API_URL", "").strip()) nous_available = check_auxiliary_model() default_summarizer_model = _get_default_summarizer_model() if web_available: backend = _get_backend() print(f"✅ Web backend: {backend}") if backend == "exa": print(" Using Exa API (https://exa.ai)") elif backend == "parallel": print(" Using Parallel API (https://parallel.ai)") elif backend == "tavily": print(" Using Tavily API (https://tavily.com)") elif backend == "searxng": print(f" Using SearXNG (search only): {os.getenv('SEARXNG_URL', '').strip()}") elif backend == "brave-free": print(" Using Brave Search free tier (search only)") elif backend == "ddgs": print(" Using DuckDuckGo via ddgs package (search only)") elif firecrawl_url_available: print(f" Using self-hosted Firecrawl: {os.getenv('FIRECRAWL_API_URL').strip().rstrip('/')}") elif firecrawl_key_available: print(" Using direct Firecrawl cloud API") elif tool_gateway_available: print(f" Using Firecrawl tool-gateway: {_get_firecrawl_gateway_url()}") else: print(" Firecrawl backend selected but not configured") else: print("❌ No web search backend configured") print( "Set EXA_API_KEY, PARALLEL_API_KEY, TAVILY_API_KEY, FIRECRAWL_API_KEY, FIRECRAWL_API_URL" f"{_firecrawl_backend_help_suffix()}" ) if not nous_available: print("❌ No auxiliary model available for LLM content processing") print("Set OPENROUTER_API_KEY, configure Nous Portal, or set OPENAI_BASE_URL + OPENAI_API_KEY") print("⚠️ Without an auxiliary model, LLM content processing will be disabled") else: print(f"✅ Auxiliary model available: {default_summarizer_model}") if not web_available: sys.exit(1) print("🛠️ Web tools ready for use!") if nous_available: print(f"🧠 LLM content processing available with {default_summarizer_model}") print(f" Default min length for processing: {DEFAULT_MIN_LENGTH_FOR_SUMMARIZATION} chars") # Show debug mode status if _debug.active: print(f"🐛 Debug mode ENABLED - Session ID: {_debug.session_id}") print(f" Debug logs will be saved to: {_debug.log_dir}/web_tools_debug_{_debug.session_id}.json") else: print("🐛 Debug mode disabled (set WEB_TOOLS_DEBUG=true to enable)") print("\nBasic usage:") print(" from web_tools import web_search_tool, web_extract_tool, web_crawl_tool") print(" import asyncio") print("") print(" # Search (synchronous)") print(" results = web_search_tool('Python tutorials')") print("") print(" # Extract and crawl (asynchronous)") print(" async def main():") print(" content = await web_extract_tool(['https://example.com'])") print(" crawl_data = await web_crawl_tool('example.com', 'Find docs')") print(" asyncio.run(main())") if nous_available: print("\nLLM-enhanced usage:") print(" # Content automatically processed for pages >5000 chars (default)") print(" content = await web_extract_tool(['https://python.org/about/'])") print("") print(" # Customize processing parameters") print(" crawl_data = await web_crawl_tool(") print(" 'docs.python.org',") print(" 'Find key concepts',") print(" model='google/gemini-3-flash-preview',") print(" min_length=3000") print(" )") print("") print(" # Disable LLM processing") print(" raw_content = await web_extract_tool(['https://example.com'], use_llm_processing=False)") print("\nDebug mode:") print(" # Enable debug logging") print(" export WEB_TOOLS_DEBUG=true") print(" # Debug logs capture:") print(" # - All tool calls with parameters") print(" # - Original API responses") print(" # - LLM compression metrics") print(" # - Final processed results") print(" # Logs saved to: ./logs/web_tools_debug_UUID.json") print("\n📝 Run 'python test_web_tools_llm.py' to test LLM processing capabilities") # --------------------------------------------------------------------------- # Registry # --------------------------------------------------------------------------- from tools.registry import registry, tool_error WEB_SEARCH_SCHEMA = { "name": "web_search", "description": "Search the web for information. Returns up to 5 results by default with titles, URLs, and descriptions. The query is passed through to the configured backend, so operators such as site:domain, filetype:pdf, intitle:word, -term, and \"exact phrase\" may work when the backend supports them.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "The search query to look up on the web. You may include backend-supported operators such as site:example.com, filetype:pdf, intitle:word, -term, or \"exact phrase\"." }, "limit": { "type": "integer", "description": "Maximum number of results to return. Defaults to 5.", "minimum": 1, "maximum": 100, "default": 5 } }, "required": ["query"] } } WEB_EXTRACT_SCHEMA = { "name": "web_extract", "description": "Extract content from web page URLs. Returns page content in markdown format. Also works with PDF URLs (arxiv papers, documents, etc.) — pass the PDF link directly and it converts to markdown text. Pages under 5000 chars return full markdown; larger pages are LLM-summarized and capped at ~5000 chars per page. Pages over 2M chars are refused. If a URL fails or times out, use the browser tool to access it instead.", "parameters": { "type": "object", "properties": { "urls": { "type": "array", "items": {"type": "string"}, "description": "List of URLs to extract content from (max 5 URLs per call)", "maxItems": 5 } }, "required": ["urls"] } } registry.register( name="web_search", toolset="web", schema=WEB_SEARCH_SCHEMA, handler=lambda args, **kw: web_search_tool(args.get("query", ""), limit=args.get("limit", 5)), check_fn=check_web_api_key, requires_env=_web_requires_env(), emoji="🔍", max_result_size_chars=100_000, ) registry.register( name="web_extract", toolset="web", schema=WEB_EXTRACT_SCHEMA, handler=lambda args, **kw: web_extract_tool( args.get("urls", [])[:5] if isinstance(args.get("urls"), list) else [], "markdown"), check_fn=check_web_api_key, requires_env=_web_requires_env(), is_async=True, emoji="📄", max_result_size_chars=100_000, )