WET - Web Extended Toolkit

Instructions

Implementation Reference

• src/wet_mcp/server.py:824-968 (handler)

  The config tool handler function. Implements actions: status (shows server configuration and status), set (updates runtime settings like log_level, tool_timeout, etc.), cache_clear (clears web cache), and docs_reindex (forces re-index of a library). Returns JSON responses for all actions.

  async def config(
      action: str,
      key: str | None = None,
      value: str | None = None,
  ) -> str:
      """Server configuration and management.
  
      Actions:
      - status: Show current config and status
      - set: Update runtime setting (key + value required)
      - cache_clear: Clear web cache
      - docs_reindex: Force re-index a library (key = library name)
      """
      match action:
          case "status":
              from wet_mcp.embedder import get_backend
              from wet_mcp.reranker import get_reranker
  
              embed_backend = get_backend()
              reranker = get_reranker()
  
              status = {
                  "database": {
                      "path": str(settings.get_db_path()),
                      "docs_indexed": (_docs_db.stats() if _docs_db else {}),
                  },
                  "embedding": {
                      "backend": (
                          type(embed_backend).__name__ if embed_backend else None
                      ),
                      "dims": _embedding_dims,
                      "available": embed_backend is not None,
                  },
                  "reranker": {
                      "available": reranker is not None,
                      "backend": (type(reranker).__name__ if reranker else None),
                  },
                  "cache": {
                      "enabled": settings.wet_cache,
                      "path": (
                          str(settings.get_cache_db_path())
                          if settings.wet_cache
                          else None
                      ),
                  },
                  "sync": {
                      "enabled": settings.sync_enabled,
                      "remote": settings.sync_remote,
                      "folder": settings.sync_folder,
                      "interval": settings.sync_interval,
                  },
                  "settings": {
                      "log_level": settings.log_level,
                      "tool_timeout": settings.tool_timeout,
                  },
              }
              return json.dumps(status, indent=2, default=str)
  
          case "set":
              if not key or value is None:
                  return json.dumps({"error": "key and value are required for set"})
              valid_keys = {
                  "log_level",
                  "tool_timeout",
                  "wet_cache",
                  "sync_enabled",
                  "sync_remote",
                  "sync_folder",
                  "sync_interval",
              }
              if key not in valid_keys:
                  return json.dumps(
                      {
                          "error": f"Invalid key: {key}",
                          "valid_keys": sorted(valid_keys),
                      }
                  )
              if key == "log_level":
                  settings.log_level = value.upper()
                  logger.remove()
                  logger.add(sys.stderr, level=settings.log_level)
              elif key == "tool_timeout":
                  settings.tool_timeout = int(value)
              elif key == "wet_cache":
                  settings.wet_cache = value.lower() in (
                      "true",
                      "1",
                      "yes",
                  )
              elif key == "sync_enabled":
                  settings.sync_enabled = value.lower() in (
                      "true",
                      "1",
                      "yes",
                  )
              elif key == "sync_interval":
                  settings.sync_interval = int(value)
              else:
                  setattr(settings, key, value)
              return json.dumps(
                  {
                      "status": "updated",
                      "key": key,
                      "value": getattr(settings, key),
                  },
                  default=str,
              )
  
          case "cache_clear":
              if _web_cache:
                  _web_cache.clear()
                  return json.dumps({"status": "cache cleared"})
              return json.dumps({"error": "Cache is not enabled"})
  
          case "docs_reindex":
              if not key:
                  return json.dumps({"error": "key (library name) is required"})
              if not _docs_db:
                  return json.dumps({"error": "Docs database not initialized"})
              lib = _docs_db.get_library(key)
              if lib:
                  ver = _docs_db.get_best_version(lib["id"])
                  if ver:
                      _docs_db.clear_version_chunks(ver["id"])
                  return json.dumps(
                      {
                          "status": "cleared",
                          "library": key,
                          "hint": ("Next docs search will re-index"),
                      }
                  )
              return json.dumps({"error": f"Library '{key}' not found in index"})
  
          case _:
              return json.dumps(
                  {
                      "error": f"Unknown action: {action}",
                      "valid_actions": [
                          "status",
                          "set",
                          "cache_clear",
                          "docs_reindex",
                      ],
                  }
              )

• src/wet_mcp/server.py:810-823 (registration)

  Registration of the config tool using @mcp.tool() decorator. Defines tool metadata including description, title, and ToolAnnotations (readOnlyHint=False, destructiveHint=False, idempotentHint=True, openWorldHint=False).

  @mcp.tool(
      description=(
          "Server config and management. Actions: "
          "status|set|cache_clear|docs_reindex. "
          "Use help tool with tool_name='config' for full docs."
      ),
      annotations=ToolAnnotations(
          title="Config",
          readOnlyHint=False,
          destructiveHint=False,
          idempotentHint=True,
          openWorldHint=False,
      ),
  )

• src/wet_mcp/config.py:52-418 (schema)

  The Settings class defines all configuration parameters that the config tool can read and modify. Includes settings for SearXNG, crawler, tool timeout, embedding/reranking backends, cache, docs storage, sync, and logging. Provides methods to resolve and validate settings.

  class Settings(BaseSettings):
      """WET MCP Server configuration.
  
      Environment variables:
      - SEARXNG_URL: SearXNG instance URL (default: http://localhost:8080)
      - API_KEYS: Provider API keys, supports multiple providers
          Format: "ENV_VAR:key,ENV_VAR:key,..."
          Or file path: "@path/to/keys"
          Example: "GOOGLE_API_KEY:AIza...,COHERE_API_KEY:..."
          Embedding providers: Google, OpenAI, Cohere
          Reranking providers: Cohere (auto-detected)
      - LITELLM_PROXY_URL: LiteLLM Proxy base URL (e.g. http://10.0.0.20:4000)
      - LITELLM_PROXY_KEY: API key for the LiteLLM Proxy
      - EMBEDDING_API_BASE: Custom embedding endpoint URL
      - EMBEDDING_API_KEY: API key for custom embedding endpoint
      - RERANK_API_BASE: Custom rerank endpoint URL
      - RERANK_API_KEY: API key for custom rerank endpoint
      - LLM_API_BASE: Custom LLM chat completion endpoint URL
      - LLM_API_KEY: API key for custom LLM endpoint
      - EMBEDDING_MODEL: LiteLLM embedding model (auto-detected if not set)
      - EMBEDDING_DIMS: Embedding dimensions (0 = auto-detect, default 768)
      - EMBEDDING_BACKEND: "litellm" | "local" (auto: API_KEYS -> litellm, else local)
          Local: GGUF if GPU + llama-cpp-python, else ONNX
      - RERANK_ENABLED: Enable reranking (default: true)
      - RERANK_BACKEND: "litellm" | "local" (auto: Cohere key -> litellm, else local)
      - RERANK_MODEL: LiteLLM rerank model (auto-detected from API_KEYS if Cohere)
      - RERANK_TOP_N: Return top N results after reranking (default: 10)
      - SYNC_ENABLED: Enable rclone sync (default: false)
      - SYNC_PROVIDER: rclone provider type (default: "drive" for Google Drive)
      - SYNC_REMOTE: Rclone remote name (default: "gdrive")
      - SYNC_FOLDER: Remote folder name (default: "wet-mcp")
      - SYNC_INTERVAL: Auto-sync interval in seconds (default: 300)
  
      LiteLLM Mode Detection (resolve_litellm_mode):
      - "proxy": LITELLM_PROXY_URL is set → all calls routed through proxy
      - "sdk": API_KEYS or custom endpoints set → direct LiteLLM SDK calls
      - "local": no keys/proxy → local ONNX models only
      """
  
      # SearXNG
      searxng_url: str = "http://localhost:41592"
      searxng_timeout: int = 30
  
      # Crawler
      crawler_headless: bool = True
      crawler_timeout: int = 60
  
      # SearXNG Management
      wet_auto_searxng: bool = True
      wet_searxng_port: int = 41592
  
      # Tool execution timeout (seconds, 0 = no timeout)
      tool_timeout: int = 120
  
      # Media
      download_dir: str = "~/.wet-mcp/downloads"
  
      # Media Analysis (LiteLLM)
      api_keys: SecretStr | None = None  # ENV_VAR:key,ENV_VAR:key (multiple providers)
  
      # LiteLLM Proxy (selfhosted gateway)
      litellm_proxy_url: str = ""  # e.g. http://10.0.0.20:4000
      litellm_proxy_key: SecretStr | None = None
  
      # Custom endpoints (e.g. modalcom-ai-workers on Modal.com)
      embedding_api_base: str = ""
      embedding_api_key: SecretStr | None = None
      rerank_api_base: str = ""
      rerank_api_key: SecretStr | None = None
      llm_api_base: str = ""
      llm_api_key: SecretStr | None = None
  
      llm_models: str = "gemini/gemini-3-flash-preview"  # provider/model (fallback chain)
      llm_temperature: float | None = None
  
      # Cache (web operations)
      wet_cache: bool = True  # Enable/disable web cache
      cache_dir: str = ""  # Cache database directory, default: ~/.wet-mcp
  
      # Docs storage
      docs_db_path: str = ""  # Default: ~/.wet-mcp/docs.db
  
      # Embedding
      embedding_model: str = ""  # LiteLLM format, auto-detect if empty
      embedding_dims: int = 0  # 0 = use server default (768)
      embedding_backend: str = (
          ""  # "litellm" | "local" | "" (auto: API_KEYS->litellm, else local)
      )
  
      # Reranking
      rerank_enabled: bool = (
          True  # Enable reranking (always available via local fallback)
      )
      rerank_backend: str = (
          ""  # "litellm" | "local" | "" (auto: Cohere->litellm, else local)
      )
      rerank_model: str = (
          ""  # LiteLLM rerank model (e.g., "cohere/rerank-multilingual-v3.0")
      )
      rerank_top_n: int = 10  # Return top N after reranking
  
      # Docs sync (rclone)
      sync_enabled: bool = False
      sync_provider: str = "drive"  # rclone provider type (drive, dropbox, s3, etc.)
      sync_remote: str = "gdrive"  # rclone remote name
      sync_folder: str = "wet-mcp"  # remote folder
      sync_interval: int = 300  # seconds, 0 = manual only
  
      # Logging
      log_level: str = "INFO"
  
      model_config = {"env_prefix": "", "case_sensitive": False}
  
      # --- Path helpers (aligned with mnemo-mcp) ---
  
      def get_data_dir(self) -> Path:
          """Get data directory.
  
          Uses CACHE_DIR if set, otherwise ~/.wet-mcp/.
          """
          if self.cache_dir:
              return Path(self.cache_dir).expanduser()
          return _default_data_dir()
  
      def get_db_path(self) -> Path:
          """Get resolved docs database path."""
          if self.docs_db_path:
              return Path(self.docs_db_path).expanduser()
          return self.get_data_dir() / "docs.db"
  
      def get_cache_db_path(self) -> Path:
          """Get resolved web cache database path."""
          return self.get_data_dir() / "cache.db"
  
      # --- API key management ---
  
      # LiteLLM uses different env vars for embeddings vs completions
      _ENV_ALIASES: dict[str, str] = {
          "GOOGLE_API_KEY": "GEMINI_API_KEY",
      }
  
      def setup_api_keys(self) -> dict[str, list[str]]:
          """Parse API_KEYS and set env vars for LiteLLM.
  
          Format: "GOOGLE_API_KEY:AIza...,OPENAI_API_KEY:sk-..."
          Or file: "@path/to/keys_file"
  
          Also sets aliases (e.g., GOOGLE_API_KEY -> GEMINI_API_KEY)
          because LiteLLM embedding uses GEMINI_API_KEY for gemini/ models.
  
          Returns:
              Dict mapping env var name to list of API keys.
          """
          if not self.api_keys:
              return {}
  
          keys_by_env: dict[str, list[str]] = {}
          api_keys_str = self.api_keys.get_secret_value()
  
          # Handle file-based keys
          if api_keys_str.startswith("@"):
              path_str = api_keys_str[1:]
              path = Path(path_str).expanduser()
              if not path.exists():
                  raise FileNotFoundError(f"API keys file not found: {path}")
  
              try:
                  # Read content and normalize newlines to commas
                  content = path.read_text(encoding="utf-8").strip()
                  api_keys_str = content.replace("\n", ",")
              except Exception as e:
                  raise ValueError(f"Failed to read API keys file: {e}") from e
  
          for pair in api_keys_str.split(","):
              pair = pair.strip()
              if ":" not in pair:
                  continue
  
              env_var, key = pair.split(":", 1)
              env_var = env_var.strip()
              key = key.strip()
  
              if not key:
                  continue
  
              keys_by_env.setdefault(env_var, []).append(key)
  
          # Set first key of each env var (LiteLLM reads from env)
          for env_var, keys in keys_by_env.items():
              if keys:
                  os.environ[env_var] = keys[0]
                  # Set alias if defined (e.g., GOOGLE_API_KEY -> GEMINI_API_KEY)
                  alias = self._ENV_ALIASES.get(env_var)
                  if alias and alias not in os.environ:
                      os.environ[alias] = keys[0]
  
          return keys_by_env
  
      # --- Embedding resolution ---
  
      def resolve_embedding_model(self) -> str | None:
          """Return explicit EMBEDDING_MODEL or None for auto-detect."""
          if self.embedding_model:
              return self.embedding_model
          return None
  
      def resolve_embedding_dims(self) -> int:
          """Return explicit EMBEDDING_DIMS or 0 for auto-detect."""
          return self.embedding_dims
  
      def resolve_local_embedding_model(self) -> str:
          """Resolve local embedding model: GGUF if GPU + llama-cpp, else ONNX."""
          return _resolve_local_model(
              "n24q02m/Qwen3-Embedding-0.6B-ONNX",
              "n24q02m/Qwen3-Embedding-0.6B-GGUF",
          )
  
      def resolve_embedding_backend(self) -> str:
          """Resolve embedding backend: 'local' or 'litellm'.
  
          Always returns a valid backend (never empty).
  
          Auto-detect order:
          1. Explicit EMBEDDING_BACKEND setting
          2. 'litellm' if in proxy or sdk mode
          3. 'local' (qwen3-embed built-in, always available)
          """
          if self.embedding_backend:
              return self.embedding_backend
          mode = self.resolve_litellm_mode()
          if mode in ("proxy", "sdk"):
              return "litellm"
          return "local"
  
      # --- Reranking resolution ---
  
      def resolve_local_rerank_model(self) -> str:
          """Resolve local rerank model: GGUF if GPU + llama-cpp, else ONNX."""
          return _resolve_local_model(
              "n24q02m/Qwen3-Reranker-0.6B-ONNX",
              "n24q02m/Qwen3-Reranker-0.6B-GGUF",
          )
  
      def resolve_rerank_backend(self) -> str:
          """Resolve reranking backend: 'local', 'litellm', or ''.
  
          Returns '' only if reranking is explicitly disabled.
          Always returns a valid backend otherwise.
  
          Auto-detect order:
          1. Explicit RERANK_BACKEND setting
          2. 'litellm' if proxy mode (proxy handles routing)
          3. 'litellm' if RERANK_MODEL is set
          4. 'litellm' if API_KEYS contains a rerank-capable provider (Cohere)
          5. 'local' (qwen3-embed built-in, always available)
          """
          if not self.rerank_enabled:
              return ""
          if self.rerank_backend:
              return self.rerank_backend
          # Proxy mode: always litellm (proxy handles routing)
          if self.litellm_proxy_url:
              return "litellm"
          if self.rerank_model:
              return "litellm"
  
          # Check env vars first (populated by setup_api_keys)
          for provider_key in _RERANK_PROVIDERS:
              if os.environ.get(provider_key):
                  return "litellm"
  
          if self.api_keys:
              val = self.api_keys.get_secret_value()
              if not val.startswith("@"):
                  for provider_key in _RERANK_PROVIDERS:
                      if provider_key in val:
                          return "litellm"
          return "local"
  
      def resolve_rerank_model(self) -> str | None:
          """Resolve rerank model from config or auto-detect from API_KEYS."""
          if self.rerank_model:
              return self.rerank_model
  
          # Check env vars first
          for provider_key, model in _RERANK_PROVIDERS.items():
              if os.environ.get(provider_key):
                  return model
  
          if self.api_keys:
              val = self.api_keys.get_secret_value()
              if not val.startswith("@"):
                  for provider_key, model in _RERANK_PROVIDERS.items():
                      if provider_key in val:
                          return model
          return None
  
      # --- LiteLLM mode resolution ---
  
      def resolve_litellm_mode(self) -> str:
          """Detect LiteLLM mode: 'proxy', 'sdk', or 'local'."""
          if self.litellm_proxy_url:
              return "proxy"
          if (
              self.api_keys
              or self.embedding_api_base
              or self.rerank_api_base
              or self.llm_api_base
          ):
              return "sdk"
          return "local"
  
      def setup_litellm(self) -> str:
          """One-time LiteLLM configuration. Call once during lifespan startup.
  
          Returns mode string: 'proxy', 'sdk', or 'local'.
          """
          mode = self.resolve_litellm_mode()
  
          if mode == "proxy":
              import litellm
  
              os.environ["LITELLM_PROXY_API_BASE"] = self.litellm_proxy_url
              os.environ["LITELLM_PROXY_API_KEY"] = (
                  self.litellm_proxy_key.get_secret_value()
                  if self.litellm_proxy_key
                  else ""
              )
              litellm.use_litellm_proxy = True
              logger.info(f"LiteLLM Proxy mode: {self.litellm_proxy_url}")
          elif mode == "sdk":
              self.setup_api_keys()
              logger.info("LiteLLM SDK direct mode")
          else:
              logger.info("Local mode (no LiteLLM)")
  
          return mode
  
      def get_embedding_litellm_kwargs(self) -> dict:
          """Get extra kwargs for litellm embedding calls (api_base, api_key for Mode 2b)."""
          kwargs = {}
          if self.embedding_api_base:
              kwargs["api_base"] = self.embedding_api_base
          if self.embedding_api_key:
              kwargs["api_key"] = self.embedding_api_key.get_secret_value()
          return kwargs
  
      def get_rerank_litellm_kwargs(self) -> dict:
          """Get extra kwargs for litellm rerank calls."""
          kwargs = {}
          if self.rerank_api_base:
              kwargs["api_base"] = self.rerank_api_base
          if self.rerank_api_key:
              kwargs["api_key"] = self.rerank_api_key.get_secret_value()
          return kwargs
  
      def get_llm_litellm_kwargs(self) -> dict:
          """Get extra kwargs for litellm chat completion calls."""
          kwargs = {}
          if self.llm_api_base:
              kwargs["api_base"] = self.llm_api_base
          if self.llm_api_key:
              kwargs["api_key"] = self.llm_api_key.get_secret_value()
          return kwargs
  
  
  settings = Settings()

• src/wet_mcp/server.py:825-828 (schema)

  Input parameters for the config tool: action (required string), key (optional string for set/docs_reindex actions), and value (optional string for set action). The function signature defines the tool's input schema.

      action: str,
      key: str | None = None,
      value: str | None = None,
  ) -> str: