""" title: Linkup powered web search author: simplenerd authors_url: https://github.com/open-webui funding_url: https://github.com/open-webui version: 0.1.0 license: MIT #Section 2: edit depth ("standard" or "deep") and output_type ("sourcedAnswer", "searchResults", "structured") #https://docs.linkup.so/pages/documentation/get-started/introduction """ import os import requests import json from datetime import datetime from typing import Optional, Any, Callable from pydantic import BaseModel, Field from linkup import LinkupClient class Tools: class Valves(BaseModel): linkup_api_key: str = Field( default="", description="API key for Linkup service", ) # Additional configuration fields can be added here. class UserValves(BaseModel): # Define any user-specific configuration here if needed. pass def __init__(self): self.citation = False # Disable built-in citations to allow custom citations. self.valves = self.Valves() self.user_valves = self.UserValves() async def search_web( self, query: str, __event_emitter__: Callable[[dict], Any] = None ) -> str: """ Performs a Linkup search and formats the output for clarity, including citation events. :param query: The search query string. :param __event_emitter__: Optional event emitter for status and citation updates. :return: A JSON string containing the formatted answer and sources. """ # --- 1. Emitting "status" start event --- if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": "Starting Linkup search...", "done": False, "hidden": False, }, } ) # --- 2. Perform the search with Linkup --- client = LinkupClient(api_key=self.valves.linkup_api_key) response = client.search( query=query, depth="deep", output_type="sourcedAnswer", ) # --- 3. Emitting "status" completion event --- if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": "Linkup search completed successfully.", "done": True, "hidden": False, }, } ) # --- 4. Extract the answer and the sources --- answer_text = getattr(response, "answer", "No answer provided.") sources_obj = getattr(response, "sources", []) # --- 5. Build enumerated citations and optionally emit "citation" events --- citations_list = [] for i, source in enumerate(sources_obj, start=0): # Create a citation marker like [1], [2], ... citation_marker = f"[{i}]" citations_list.append( { "marker": citation_marker, "name": source.name, "url": source.url, "snippet": source.snippet, } ) # Emit the citation event for each source if __event_emitter__: await __event_emitter__( { "type": "citation", "data": { "document": [source.snippet], "metadata": [ { "date_accessed": datetime.now().isoformat(), "source": source.name, } ], "source": {"name": source.name, "url": source.url}, }, } ) # --- 6. Create a user-readable citations section and combine it with the answer --- if citations_list: # Example enumerated citations: # [1] SourceName (http://sourceURL): snippet ... citations_text = "\n".join( f"{c['marker']} {c['name']} ({c['url']}): {c['snippet']}" for c in citations_list ) answer_with_citations = ( f"{answer_text}\n\n" "References:\n" + citations_text ) else: answer_with_citations = answer_text # --- 7. Build the final structured output --- # Return both the plain answer and the annotated one, # as well as a structured list of citations. formatted_response = { "answer": answer_with_citations, } # --- 8. Return the response as JSON --- return json.dumps(formatted_response, ensure_ascii=False)