How to Add a Search API to OpenWebUI

# OpenWebUI Function Tool: scavio_search.py
# Add this as a Function in OpenWebUI Admin > Functions

import os
import requests
from typing import Optional

class Tools:
    def __init__(self):
        self.api_key = os.environ.get('SCAVIO_API_KEY', '')
        self.base_url = 'https://api.scavio.dev/api/v2/google'

    def search_web(self, query: str, num_results: Optional[int] = 5) -> str:
        """Search the web for current information using Scavio.
        Returns titles, snippets, and source URLs.
        """
        try:
            resp = requests.post(self.base_url,
                headers={'Authorization': 'Bearer ' + self.api_key,
                         'Content-Type': 'application/json'},
                json={'query': query,
 'gl': 'us'},
                timeout=10)
            resp.raise_for_status()
            results = resp.json().get('organic_results', [])
            if not results:
                return 'No results found.'
            return '\n\n'.join(
                f'[{i+1}] {r["title"]}\n{r.get("snippet", "")}\nSource: {r["link"]}'
                for i, r in enumerate(results)
            )
        except Exception as e:
            return f'Search error: {str(e)}'

# Extended Tools class with platform-specific methods
class Tools:
    def __init__(self):
        self.api_key = os.environ.get('SCAVIO_API_KEY', '')
        self.base_url = 'https://api.scavio.dev/api/v2/google'
        self.headers = {'Authorization': 'Bearer ' + self.api_key, 'Content-Type': 'application/json'}

    def _search(self, query: str, num: int = 5) -> list:
        resp = requests.post(self.base_url, headers=self.headers,
            json={'query': query}, timeout=10)
        resp.raise_for_status()
        return resp.json().get('organic_results', [])

    def search_web(self, query: str) -> str:
        """Search the web for current information."""
        results = self._search(query)
        return '\n\n'.join(f'[{i+1}] {r["title"]}\n{r.get("snippet","")}\nURL: {r["link"]}'
                            for i, r in enumerate(results)) or 'No results.'

    def search_reddit(self, query: str) -> str:
        """Search Reddit for community discussions and opinions."""
        results = self._search(f'site:reddit.com {query}')
        return '\n\n'.join(f'[{i+1}] {r["title"]}\n{r.get("snippet","")}\nURL: {r["link"]}'
                            for i, r in enumerate(results)) or 'No Reddit results.'

    def search_youtube(self, query: str) -> str:
        """Search YouTube for videos on a topic."""
        results = self._search(f'site:youtube.com {query}')
        return '\n\n'.join(f'[{i+1}] {r["title"]}\nURL: {r["link"]}'
                            for i, r in enumerate(results)) or 'No YouTube results.'

    def search_amazon(self, query: str) -> str:
        """Search Amazon for product information."""
        results = self._search(f'site:amazon.com {query}')
        return '\n\n'.join(f'[{i+1}] {r["title"]}\n{r.get("snippet","")}\nURL: {r["link"]}'
                            for i, r in enumerate(results)) or 'No Amazon results.'

# Docker deployment: add to your docker-compose.yml or .env
# SCAVIO_API_KEY=your_key_here

# Or set in the OpenWebUI admin panel:
# 1. Go to Admin Panel > Functions
# 2. Click 'Create Function'
# 3. Paste the Tools class code
# 4. Set the function name to 'scavio_search'
# 5. Enable it globally or per-model

# Verify the setup with a test
import os, requests

def verify_openwebui_search():
    key = os.environ.get('SCAVIO_API_KEY')
    if not key:
        print('ERROR: SCAVIO_API_KEY not set')
        return False
    tools = Tools()
    result = tools.search_web('test query 2026')
    if 'error' in result.lower():
        print(f'ERROR: {result}')
        return False
    print('OpenWebUI search configured successfully')
    print(f'Test result preview: {result[:150]}')
    return True

verify_openwebui_search()

# Test all search functions
def test_all_tools():
    tools = Tools()
    tests = [
        ('search_web', 'latest AI news 2026'),
        ('search_reddit', 'best mechanical keyboard'),
        ('search_youtube', 'python tutorial beginners'),
        ('search_amazon', 'noise cancelling headphones'),
    ]
    for method_name, query in tests:
        method = getattr(tools, method_name)
        result = method(query)
        lines = result.split('\n')
        has_results = len(lines) > 1
        print(f'[{"PASS" if has_results else "FAIL"}] {method_name}')
        print(f'  Query: {query}')
        print(f'  First result: {lines[0][:60]}')
        print()

test_all_tools()
print('All tools ready for OpenWebUI')
print('Cost: $0.005 per search (50 free credits on signup)')