feat(web-search): tool DuckDuckGo abierta en /web-search para watsonx Orchestrate

Sub-app FastAPI montada en /web-search con búsqueda vía ddgs.
OpenAPI 3.0.3 + servers absoluto + operationId webSearch listo para
importar como external tool en wxO. Sin auth (uso de taller/demo).

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-28 03:37:55 +00:00
parent f8c7139e6e
commit 385c448c1f
8 changed files with 274 additions and 0 deletions

154
app/web_search/api.py Normal file
View File

@@ -0,0 +1,154 @@
"""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 (110). 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]