Fix xAI sentiment: use Agent Tools web_search (Live Search deprecated)

xAI returned 410 — search_parameters/Live Search is retired. Route xAI
through the Responses API web_search tool instead (same path as OpenAI):
- OpenAISentimentProvider parametrized with base_url / tool_type / source
- xAI builds it against https://api.x.ai/v1 with the web_search tool
- Drop the dead Live Search code from the generic compatible provider
- Frontend label: "xAI Grok — web search"

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

app/providers/openai_compatible_sentiment.py

@@ -39,21 +39,6 @@ Rules:
 - reasoning should be a brief one-sentence explanation
 """
 
-_SENTIMENT_PROMPT_SEARCH = """\
-Search the web and X for the LATEST news, analyst opinions, and market developments \
-about the stock ticker {ticker} from the past 24-48 hours.
-
-Based on your search findings, analyze the CURRENT market sentiment.
-
-Respond ONLY with a JSON object in this exact format (no markdown, no extra text):
-{{"classification": "<bullish|bearish|neutral>", "confidence": <0-100>, "reasoning": "<brief explanation citing recent news>"}}
-
-Rules:
-- classification must be exactly one of: bullish, bearish, neutral
-- confidence must be an integer from 0 to 100
-- reasoning should cite specific recent news or events you found
-"""
-
 VALID_CLASSIFICATIONS = {"bullish", "bearish", "neutral"}
 
 
@@ -75,8 +60,6 @@ class OpenAICompatibleSentimentProvider:
         model: str,
         base_url: str,
         source: str = "openai_compatible",
-        live_search: bool = False,
-        extra_body: dict | None = None,
     ) -> None:
         if not api_key:
             raise ProviderError("API key is required")
@@ -92,29 +75,8 @@ class OpenAICompatibleSentimentProvider:
         self._client = AsyncOpenAI(api_key=api_key, base_url=base_url, http_client=http_client)
         self._model = model
         self._source = source
-        self._live_search = live_search
-        self._extra_body = extra_body
-
-    @staticmethod
-    def _extract_citations(response: object) -> list[dict[str, str]]:
-        """Best-effort extraction of xAI Live Search citations (list of URLs)."""
-        raw = getattr(response, "citations", None)
-        if not raw:
-            extra = getattr(response, "model_extra", None) or {}
-            raw = extra.get("citations") if isinstance(extra, dict) else None
-        citations: list[dict[str, str]] = []
-        for item in raw or []:
-            if isinstance(item, str):
-                citations.append({"url": item, "title": ""})
-            elif isinstance(item, dict) and item.get("url"):
-                citations.append({"url": str(item["url"]), "title": str(item.get("title", ""))})
-        return citations
 
     async def fetch_sentiment(self, ticker: str) -> SentimentData:
-        prompt = _SENTIMENT_PROMPT_SEARCH if self._live_search else _SENTIMENT_PROMPT
-        kwargs: dict = {}
-        if self._extra_body:
-            kwargs["extra_body"] = self._extra_body
         try:
             response = await self._client.chat.completions.create(
                 model=self._model,
@@ -123,10 +85,9 @@ class OpenAICompatibleSentimentProvider:
                         "role": "system",
                         "content": "You are a financial sentiment analyst. Always respond with valid JSON only, no markdown fences.",
                     },
-                    {"role": "user", "content": prompt.format(ticker=ticker)},
+                    {"role": "user", "content": _SENTIMENT_PROMPT.format(ticker=ticker)},
                 ],
                 temperature=0.3,
-                **kwargs,
             )
             raw_text = (response.choices[0].message.content or "").strip()
             if not raw_text:
@@ -155,7 +116,6 @@ class OpenAICompatibleSentimentProvider:
                 source=self._source,
                 timestamp=datetime.now(timezone.utc),
                 reasoning=reasoning,
-                citations=self._extract_citations(response) if self._live_search else [],
             )
 
         except json.JSONDecodeError as exc:

app/providers/openai_sentiment.py

@@ -55,17 +55,34 @@ VALID_CLASSIFICATIONS = {"bullish", "bearish", "neutral"}
 
 
 class OpenAISentimentProvider:
-    """Fetches sentiment analysis from OpenAI Responses API with live web search."""
+    """Sentiment via the Responses API + web-search tool, with live grounding.
 
-    def __init__(self, api_key: str, model: str = "gpt-4o-mini") -> None:
+    Works against any provider implementing the OpenAI Responses API. OpenAI
+    uses the ``web_search_preview`` tool; xAI Grok uses ``web_search`` at the
+    ``https://api.x.ai/v1`` base URL.
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        model: str = "gpt-4o-mini",
+        base_url: str | None = None,
+        tool_type: str = "web_search_preview",
+        source: str = "openai",
+    ) -> None:
         if not api_key:
-            raise ProviderError("OpenAI API key is required")
+            raise ProviderError(f"{source} API key is required")
         http_kwargs: dict = {}
         if _CA_BUNDLE and Path(_CA_BUNDLE).exists():
             http_kwargs["verify"] = _CA_BUNDLE
         http_client = httpx.AsyncClient(**http_kwargs)
-        self._client = AsyncOpenAI(api_key=api_key, http_client=http_client)
+        client_kwargs: dict = {"api_key": api_key, "http_client": http_client}
+        if base_url:
+            client_kwargs["base_url"] = base_url
+        self._client = AsyncOpenAI(**client_kwargs)
         self._model = model
+        self._tool_type = tool_type
+        self._source = source
 
     @staticmethod
     def _extract_raw_text(response: object, ticker_context: str) -> str:
@@ -89,12 +106,11 @@ class OpenAISentimentProvider:
             clean = clean[:-3]
         return clean.strip()
 
-    @staticmethod
+    def _normalize_single_result(self, parsed: dict, ticker: str, citations: list[dict[str, str]]) -> SentimentData:
-    def _normalize_single_result(parsed: dict, ticker: str, citations: list[dict[str, str]]) -> SentimentData:
         classification = str(parsed.get("classification", "")).lower()
         if classification not in VALID_CLASSIFICATIONS:
             raise ProviderError(
-                f"Invalid classification '{classification}' from OpenAI for {ticker}"
+                f"Invalid classification '{classification}' from {self._source} for {ticker}"
             )
 
         confidence = int(parsed.get("confidence", 50))
@@ -103,7 +119,8 @@ class OpenAISentimentProvider:
 
         if reasoning:
             logger.info(
-                "OpenAI sentiment for %s: %s (confidence=%d) — %s",
+                "%s sentiment for %s: %s (confidence=%d) — %s",
+                self._source,
                 ticker,
                 classification,
                 confidence,
@@ -114,7 +131,7 @@ class OpenAISentimentProvider:
             ticker=ticker,
             classification=classification,
             confidence=confidence,
-            source="openai",
+            source=self._source,
             timestamp=datetime.now(timezone.utc),
             reasoning=reasoning,
             citations=citations,
@@ -125,7 +142,7 @@ class OpenAISentimentProvider:
         try:
             response = await self._client.responses.create(
                 model=self._model,
-                tools=[{"type": "web_search_preview"}],
+                tools=[{"type": self._tool_type}],
                 instructions="You are a financial sentiment analyst. Always respond with valid JSON only, no markdown fences.",
                 input=_SENTIMENT_PROMPT.format(ticker=ticker),
             )
@@ -172,7 +189,7 @@ class OpenAISentimentProvider:
         try:
             response = await self._client.responses.create(
                 model=self._model,
-                tools=[{"type": "web_search_preview"}],
+                tools=[{"type": self._tool_type}],
                 instructions="You are a financial sentiment analyst. Always respond with valid JSON only, no markdown fences.",
                 input=_SENTIMENT_BATCH_PROMPT.format(tickers_csv=", ".join(normalized)),
             )

app/services/sentiment_provider_service.py

@@ -39,12 +39,10 @@ PROVIDER_BASE_URLS: dict[str, str] = {
 }
 
 # Providers grounded in live web search. The rest score from model knowledge.
-# xAI grounds via Live Search (search_parameters); OpenAI/Gemini via their tools.
+# xAI and OpenAI ground via the Responses API web-search tool; Gemini via its
+# own search grounding.
 WEB_SEARCH_PROVIDERS = {"openai", "gemini", "xai"}
 
-# xAI Live Search: auto mode lets Grok search web + X when the query needs it.
-_XAI_SEARCH_PARAMETERS = {"mode": "auto", "return_citations": True}
-
 # Providers needing a user-supplied base URL (generic compatible endpoints).
 CUSTOM_BASE_URL_PROVIDERS = {"openai_compatible"}
 
@@ -181,16 +179,17 @@ async def build_sentiment_provider(db: AsyncSession):
     if provider == "gemini":
         from app.providers.gemini_sentiment import GeminiSentimentProvider
         return GeminiSentimentProvider(api_key, model)
-    if provider in {"deepseek", "xai", "openai_compatible"}:
+    if provider == "xai":
+        # xAI grounds via the Responses API web_search tool (the former Live
+        # Search / search_parameters API is deprecated).
+        from app.providers.openai_sentiment import OpenAISentimentProvider
+        return OpenAISentimentProvider(
+            api_key, model, base_url=base_url, tool_type="web_search", source="xai",
+        )
+    if provider in {"deepseek", "openai_compatible"}:
         if not base_url:
             raise ProviderError(f"No base_url configured for sentiment provider '{provider}'")
         from app.providers.openai_compatible_sentiment import OpenAICompatibleSentimentProvider
-        if provider == "xai":
-            return OpenAICompatibleSentimentProvider(
-                api_key, model, base_url, source="xai",
-                live_search=True,
-                extra_body={"search_parameters": _XAI_SEARCH_PARAMETERS},
-            )
         return OpenAICompatibleSentimentProvider(api_key, model, base_url, source=provider)
 
     raise ProviderError(f"Unsupported sentiment provider '{provider}'")

frontend/src/components/admin/SentimentProviderSettings.tsx

@@ -18,7 +18,7 @@ const PROVIDER_LABELS: Record<string, string> = {
   openai: 'OpenAI — web search',
   gemini: 'Google Gemini — web search',
   deepseek: 'DeepSeek — cheap, no web search',
-  xai: 'xAI Grok — Live Search',
+  xai: 'xAI Grok — web search',
   openai_compatible: 'OpenAI-compatible — custom URL',
 };
 

tests/unit/test_sentiment_provider_service.py

@@ -102,13 +102,13 @@ class TestBuildProvider:
         assert config["base_url"] == "https://api.deepseek.com"
         assert config["web_search"] is False
 
-    async def test_builds_xai_with_live_search(self, session: AsyncSession):
+    async def test_builds_xai_via_responses_web_search(self, session: AsyncSession):
         await sps.update_sentiment_config(session, provider="xai", api_key="xai-x")
         provider = await sps.build_sentiment_provider(session)
-        assert type(provider).__name__ == "OpenAICompatibleSentimentProvider"
+        # xAI grounds via the Responses API web_search tool
-        # xAI is wired with Live Search enabled
+        assert type(provider).__name__ == "OpenAISentimentProvider"
-        assert provider._live_search is True
+        assert provider._tool_type == "web_search"
-        assert provider._extra_body == {"search_parameters": {"mode": "auto", "return_citations": True}}
+        assert provider._source == "xai"
         config = await sps.get_sentiment_config(session)
         assert config["base_url"] == "https://api.x.ai/v1"
         assert config["web_search"] is True