"""Sub-app FastAPI con la web-search tool para watsonx Orchestrate. Se monta en la app principal bajo `/web-search`, por lo que: - `GET /web-search/search` → endpoint de búsqueda (abierto, sin auth) - `GET /web-search/health` → health check (fuera del schema) - `GET /web-search/openapi.json`→ spec OpenAPI 3.0.3 listo para importar en wxO - `GET /web-search/docs` → Swagger UI para probar """ from __future__ import annotations from typing import Annotated from fastapi import Depends, FastAPI, HTTPException, Query, status from fastapi.openapi.utils import get_openapi from pydantic import BaseModel, Field from app.web_search.config import WebSearchSettings, get_web_search_settings from app.web_search.search import SearchProviderError, search_web OPERATION_DESCRIPTION = ( "Busca en la web y devuelve los resultados más relevantes (título, URL y " "resumen). Úsala cuando necesites información actual o reciente, noticias, " "datos que puedan haber cambiado, o cualquier información que no esté en tu " "conocimiento base." ) class SearchResultModel(BaseModel): title: str = Field(..., description="Título del resultado de búsqueda.") url: str = Field(..., description="URL del resultado.") snippet: str = Field( ..., description="Fragmento de texto / resumen del resultado (texto plano, sin HTML, máx. 300 caracteres).", ) class SearchResponse(BaseModel): query: str = Field(..., description="Consulta original recibida.") count: int = Field(..., description="Cantidad de resultados devueltos.") results: list[SearchResultModel] = Field( ..., description="Lista de resultados, ordenados de mayor a menor relevancia." ) web_search_app = FastAPI( title="FIT Web Search Tool", version="1.0.0", description=( "Herramienta de búsqueda web para agentes de watsonx Orchestrate. " "Devuelve resultados normalizados desde DuckDuckGo." ), docs_url="/docs", redoc_url=None, openapi_url="/openapi.json", ) @web_search_app.get( "/health", include_in_schema=False, ) def health() -> dict[str, str]: return {"status": "ok"} @web_search_app.get( "/search", operation_id="webSearch", summary="Buscar en la web", description=OPERATION_DESCRIPTION, response_model=SearchResponse, responses={ 422: {"description": "Parámetros de entrada inválidos."}, 502: {"description": "Error en el proveedor de búsqueda."}, }, ) def search( settings: Annotated[WebSearchSettings, Depends(get_web_search_settings)], query: Annotated[ str, Query( min_length=1, max_length=400, description="Consulta de búsqueda en lenguaje natural.", examples=["últimas noticias IBM watsonx orchestrate"], ), ], max_results: Annotated[ int, Query( ge=1, le=10, description="Cantidad de resultados a devolver (1–10). Por defecto 5.", ), ] = 5, region: Annotated[ str, Query( min_length=2, max_length=10, description=( "Región/idioma del motor de búsqueda. Ejemplos: 'wt-wt' (global), " "'cl-es' (Chile/español), 'mx-es' (México/español), 'us-en' (EEUU/inglés)." ), examples=["wt-wt", "cl-es", "mx-es", "us-en"], ), ] = "wt-wt", ) -> SearchResponse: try: results = search_web( query=query, max_results=max_results, region=region, timeout=settings.request_timeout, ) except SearchProviderError: raise HTTPException( status_code=status.HTTP_502_BAD_GATEWAY, detail="Error en el proveedor de búsqueda", ) return SearchResponse( query=query, count=len(results), results=[ SearchResultModel(title=r.title, url=r.url, snippet=r.snippet) for r in results ], ) def _custom_openapi() -> dict: if web_search_app.openapi_schema: return web_search_app.openapi_schema settings = get_web_search_settings() schema = get_openapi( title=web_search_app.title, version=web_search_app.version, description=web_search_app.description, routes=web_search_app.routes, ) # wxO prefiere 3.0.x (FastAPI emite 3.1.0 por defecto) schema["openapi"] = "3.0.3" # Bloque servers absoluto — obligatorio para wxO schema["servers"] = [{"url": settings.public_base_url}] web_search_app.openapi_schema = schema return schema web_search_app.openapi = _custom_openapi # type: ignore[assignment]