Add web search compatibility support

README.md

@@ -255,7 +255,7 @@ curl http://localhost:8000/v1/images/edits \
 <summary><code>POST /v1/chat/completions</code></summary>
 <br>
 
-面向图片场景的 Chat Completions 兼容接口，不是完整通用聊天代理。
+面向文本、网页搜索与图片场景的 Chat Completions 兼容接口，不是完整通用聊天代理。
 
 ```bash
 curl http://localhost:8000/v1/chat/completions \
@@ -279,10 +279,12 @@ curl http://localhost:8000/v1/chat/completions \
 
 | 字段         | 说明                |
 |:-----------|:------------------|
-| `model`    | 图片模型，默认按图片生成场景处理  |
-| `messages` | 消息数组，需要是图片相关请求内容  |
-| `n`        | 生成数量，按当前实现解析为图片数量 |
-| `stream`   | 已实现，但仍在测试         |
+| `model`    | 文本、搜索或图片模型；搜索模型会触发网页搜索兼容逻辑 |
+| `messages` | 消息数组，支持文本、搜索和图片请求内容 |
+| `n`        | 图片生成数量，按当前实现解析为图片数量 |
+| `stream`   | 文本、搜索和图片场景均支持，仍在测试 |
+| `tools`    | 文本场景支持 `web_search` / `web_search_preview` / `web_search_preview_2025_03_11` |
+| `web_search_options` | 传入时会触发网页搜索兼容逻辑 |
 
 <br>
 </details>
@@ -292,7 +294,7 @@ curl http://localhost:8000/v1/chat/completions \
 <summary><code>POST /v1/responses</code></summary>
 <br>
 
-面向图片生成工具调用的 Responses API 兼容接口，不是完整通用 Responses API 代理。
+面向文本、网页搜索和图片生成工具调用的 Responses API 兼容接口，不是完整通用 Responses API 代理。
 
 ```bash
 curl http://localhost:8000/v1/responses \
@@ -315,9 +317,9 @@ curl http://localhost:8000/v1/responses \
 
 | 字段       | 说明                            |
 |:---------|:------------------------------|
-| `model`  | 响应中会回显该模型字段，但图片生成当前仍走图片生成兼容逻辑 |
-| `input`  | 输入内容，需要能解析出图片生成提示词            |
-| `tools`  | 必须包含 `image_generation` 工具请求  |
+| `model`  | 响应中会回显该模型字段，搜索和图片生成会走对应兼容逻辑 |
+| `input`  | 输入内容；搜索使用最后一条用户文本，图片生成需能解析出提示词 |
+| `tools`  | 支持 `image_generation`、`web_search`、`web_search_preview`、`web_search_preview_2025_03_11` |
 | `stream` | 已实现，但仍在测试                     |
 
 <br>

services/protocol/openai_v1_chat_complete.py

@@ -21,6 +21,14 @@ from services.protocol.conversation import (
     stream_text_deltas,
     text_backend,
 )
+from services.protocol.web_search_tool import (
+    WEB_SEARCH_TOOL_TYPES,
+    has_unsupported_tools,
+    is_web_search_chat_request,
+    run_web_search,
+    search_query_from_messages,
+    text_with_url_citations,
+)
 from utils.helper import build_chat_image_markdown_content, extract_chat_image, extract_chat_prompt, is_image_chat_request, parse_image_count
 from utils.image_tokens import (
     chat_usage_from_image_usage,
@@ -30,7 +38,7 @@ from utils.image_tokens import (
 )
 
 TOOL_UNAVAILABLE_SYSTEM_MESSAGE = (
-    "This compatibility backend cannot execute local tools, shell commands, web searches, "
+    "This compatibility backend cannot execute local tools, shell commands, non-search tools, "
     "or file operations. Do not claim to have run tools or inspected external resources. "
     "If a user asks you to use a tool, say that tool execution is unavailable through this backend."
 )
@@ -51,11 +59,15 @@ def completion_response(
     content: str,
     created: int | None = None,
     messages: list[dict[str, Any]] | None = None,
+    annotations: list[dict[str, Any]] | None = None,
 ) -> dict[str, Any]:
     prompt_text_tokens = count_message_text_tokens(messages, model) if messages else 0
     prompt_image_tokens = count_message_image_tokens(messages, model) if messages else 0
     prompt_tokens = prompt_text_tokens + prompt_image_tokens
     completion_tokens = count_text_tokens(content, model) if messages else 0
+    message = {"role": "assistant", "content": content}
+    if annotations:
+        message["annotations"] = annotations
     return {
         "id": f"chatcmpl-{uuid.uuid4().hex}",
         "object": "chat.completion",
@@ -63,7 +75,7 @@ def completion_response(
         "model": model,
         "choices": [{
             "index": 0,
-            "message": {"role": "assistant", "content": content},
+            "message": message,
             "finish_reason": "stop",
         }],
         "usage": {
@@ -137,12 +149,52 @@ def chat_image_args(body: dict[str, Any]) -> tuple[str, str, int, list[tuple[byt
 def text_chat_parts(body: dict[str, Any]) -> tuple[str, list[dict[str, Any]]]:
     model = str(body.get("model") or "auto").strip() or "auto"
     messages = normalize_text_messages(normalize_messages(chat_messages_from_body(body)))
-    tools = body.get("tools")
-    if isinstance(tools, list) and tools:
+    if has_unsupported_tools(body, WEB_SEARCH_TOOL_TYPES):
         messages.insert(0, {"role": "system", "content": TOOL_UNAVAILABLE_SYSTEM_MESSAGE})
     return model, messages
 
 
+def chat_completion_annotations(annotations: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    output = []
+    for item in annotations:
+        if item.get("type") != "url_citation":
+            continue
+        output.append({
+            "type": "url_citation",
+            "url_citation": {
+                "start_index": item.get("start_index", 0),
+                "end_index": item.get("end_index", 0),
+                "url": item.get("url", ""),
+                "title": item.get("title", ""),
+            },
+        })
+    return output
+
+
+def web_search_chat_response(messages: list[dict[str, Any]], model: str) -> dict[str, Any]:
+    query = search_query_from_messages(messages)
+    if not query:
+        raise HTTPException(status_code=400, detail={"error": "messages or prompt is required for web search"})
+    text, annotations = text_with_url_citations(run_web_search(query))
+    return completion_response(
+        model,
+        text,
+        messages=messages,
+        annotations=chat_completion_annotations(annotations),
+    )
+
+
+def stream_web_search_chat_completion(messages: list[dict[str, Any]], model: str) -> Iterator[dict[str, Any]]:
+    query = search_query_from_messages(messages)
+    if not query:
+        raise HTTPException(status_code=400, detail={"error": "messages or prompt is required for web search"})
+    text, _annotations = text_with_url_citations(run_web_search(query))
+    completion_id = f"chatcmpl-{uuid.uuid4().hex}"
+    created = int(time.time())
+    yield completion_chunk(model, {"role": "assistant", "content": text}, None, completion_id, created)
+    yield completion_chunk(model, {}, "stop", completion_id, created)
+
+
 def image_result_content(result: dict[str, Any]) -> str:
     data = result.get("data")
     if isinstance(data, list) and data:
@@ -212,6 +264,8 @@ def handle(body: dict[str, Any]) -> dict[str, Any] | Iterator[dict[str, Any]]:
         if is_image_chat_request(body):
             return image_chat_events(body)
         model, messages = text_chat_parts(body)
+        if is_web_search_chat_request(body) and not has_unsupported_tools(body, WEB_SEARCH_TOOL_TYPES):
+            return stream_web_search_chat_completion(messages, model)
         key = cache_key(body, messages, stream=True)
         return chat_completion_cache.get_or_compute_stream(
             key,
@@ -220,6 +274,8 @@ def handle(body: dict[str, Any]) -> dict[str, Any] | Iterator[dict[str, Any]]:
     if is_image_chat_request(body):
         return image_chat_response(body)
     model, messages = text_chat_parts(body)
+    if is_web_search_chat_request(body) and not has_unsupported_tools(body, WEB_SEARCH_TOOL_TYPES):
+        return web_search_chat_response(messages, model)
     key = cache_key(body, messages, stream=False)
     return chat_completion_cache.get_or_compute_response(
         key,

services/protocol/openai_v1_response.py

@@ -19,6 +19,15 @@ from services.protocol.conversation import (
     stream_text_deltas,
     text_backend,
 )
+from services.protocol.web_search_tool import (
+    WEB_SEARCH_TOOL_TYPES,
+    has_unsupported_tools,
+    has_web_search_tool,
+    normalized_sources,
+    run_web_search,
+    search_query_from_messages,
+    text_with_url_citations,
+)
 from utils.helper import extract_image_from_message_content, extract_response_prompt, has_response_image_generation_tool
 from utils.image_tokens import (
     count_image_content_tokens,
@@ -28,7 +37,7 @@ from utils.image_tokens import (
 )
 
 TOOL_UNAVAILABLE_SYSTEM_MESSAGE = (
-    "This compatibility backend cannot execute local tools, shell commands, web searches, "
+    "This compatibility backend cannot execute local tools, shell commands, non-search tools, "
     "or file operations. Do not claim to have run tools or inspected external resources. "
     "If a user asks you to use a tool, say that tool execution is unavailable through this backend."
 )
@@ -40,14 +49,8 @@ def is_text_response_request(body: dict[str, Any]) -> bool:
     return not has_response_image_generation_tool(body)
 
 
-def has_non_image_tools(body: dict[str, Any]) -> bool:
-    tools = body.get("tools")
-    if not isinstance(tools, list):
-        return False
-    return any(
-        isinstance(tool, dict) and str(tool.get("type") or "").strip() != "image_generation"
-        for tool in tools
-    )
+def has_unsupported_response_tools(body: dict[str, Any]) -> bool:
+    return has_unsupported_tools(body, {"image_generation", *WEB_SEARCH_TOOL_TYPES})
 
 
 def response_image_tool(body: dict[str, Any]) -> dict[str, object]:
@@ -165,13 +168,43 @@ def messages_from_input(input_value: object, instructions: object = None) -> lis
     return messages
 
 
-def text_output_item(text: str, item_id: str | None = None, status: str = "completed") -> dict[str, Any]:
+def text_output_item(
+    text: str,
+    item_id: str | None = None,
+    status: str = "completed",
+    annotations: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
     return {
         "id": item_id or f"msg_{uuid.uuid4().hex}",
         "type": "message",
         "status": status,
         "role": "assistant",
-        "content": [{"type": "output_text", "text": text, "annotations": []}],
+        "content": [{"type": "output_text", "text": text, "annotations": annotations or []}],
+    }
+
+
+def web_search_call_item(
+    query: str,
+    item_id: str | None = None,
+    status: str = "completed",
+    sources: list[dict[str, str]] | None = None,
+) -> dict[str, Any]:
+    action: dict[str, Any] = {
+        "type": "search",
+        "query": query,
+        "queries": [query],
+    }
+    if sources:
+        action["sources"] = [
+            {"type": "url", "url": source["url"]}
+            for source in sources
+            if source.get("url")
+        ]
+    return {
+        "id": item_id or f"ws_{uuid.uuid4().hex}",
+        "type": "web_search_call",
+        "status": status,
+        "action": action,
     }
 
 
@@ -236,7 +269,7 @@ def response_completed(
 def text_response_parts(body: dict[str, Any]) -> tuple[str, list[dict[str, Any]]]:
     model = str(body.get("model") or "auto").strip() or "auto"
     messages = normalize_text_messages(normalize_messages(messages_from_input(body.get("input"), body.get("instructions"))))
-    if has_non_image_tools(body):
+    if has_unsupported_response_tools(body):
         messages.insert(0, {"role": "system", "content": TOOL_UNAVAILABLE_SYSTEM_MESSAGE})
     return model, messages
 
@@ -265,6 +298,44 @@ def stream_text_response(backend, body: dict[str, Any], messages: list[dict[str,
     yield response_completed(response_id, model, created, [item], usage)
 
 
+def stream_web_search_response(body: dict[str, Any], messages: list[dict[str, Any]] | None = None) -> Iterator[dict[str, Any]]:
+    model = str(body.get("model") or "auto").strip() or "auto"
+    messages = messages if messages is not None else messages_from_input(body.get("input"), body.get("instructions"))
+    query = search_query_from_messages(messages) or extract_response_prompt(body.get("input"))
+    if not query:
+        raise HTTPException(status_code=400, detail={"error": "input text is required for web_search"})
+
+    response_id = f"resp_{uuid.uuid4().hex}"
+    search_id = f"ws_{uuid.uuid4().hex}"
+    item_id = f"msg_{uuid.uuid4().hex}"
+    created = int(time.time())
+    yield response_created(response_id, model, created)
+
+    searching_item = web_search_call_item(query, search_id, "in_progress")
+    yield {"type": "response.output_item.added", "output_index": 0, "item": searching_item}
+    yield {"type": "response.web_search_call.in_progress", "output_index": 0, "item_id": search_id}
+    yield {"type": "response.web_search_call.searching", "output_index": 0, "item_id": search_id}
+    result = run_web_search(query)
+    search_item = web_search_call_item(query, search_id, "completed", normalized_sources(result))
+    yield {"type": "response.web_search_call.completed", "output_index": 0, "item_id": search_id}
+    yield {"type": "response.output_item.done", "output_index": 0, "item": search_item}
+
+    text, annotations = text_with_url_citations(result)
+    message_item = text_output_item("", item_id, "in_progress", annotations)
+    yield {"type": "response.output_item.added", "output_index": 1, "item": message_item}
+    if text:
+        yield {"type": "response.output_text.delta", "item_id": item_id, "output_index": 1, "content_index": 0, "delta": text}
+    yield {"type": "response.output_text.done", "item_id": item_id, "output_index": 1, "content_index": 0, "text": text}
+    message_item = text_output_item(text, item_id, "completed", annotations)
+    yield {"type": "response.output_item.done", "output_index": 1, "item": message_item}
+    usage = token_usage(
+        input_text_tokens=count_message_text_tokens(messages, model),
+        input_image_tokens=count_message_image_tokens(messages, model),
+        output_text_tokens=count_text_tokens(text, model),
+    )
+    yield response_completed(response_id, model, created, [search_item, message_item], usage)
+
+
 def stream_image_response(
     image_outputs: Iterable[ImageOutput],
     prompt: str,
@@ -319,6 +390,9 @@ def collect_response(events: Iterable[dict[str, Any]]) -> dict[str, Any]:
 def response_events(body: dict[str, Any]) -> Iterator[dict[str, Any]]:
     if is_text_response_request(body):
         model, messages = text_response_parts(body)
+        if has_web_search_tool(body) and not has_unsupported_response_tools(body):
+            yield from stream_web_search_response(body, messages)
+            return
         key = cache_key(body, messages, stream=bool(body.get("stream")))
         yield from chat_completion_cache.get_or_compute_stream(
             key,

services/protocol/web_search_tool.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from services.account_service import account_service
+from services.openai_backend_api import OpenAIBackendAPI
+
+WEB_SEARCH_TOOL_TYPES = {"web_search", "web_search_preview", "web_search_preview_2025_03_11"}
+SEARCH_CHAT_MODEL_PREFIXES = (
+    "gpt-4o-search-preview",
+    "gpt-4o-mini-search-preview",
+    "gpt-5-search-api",
+)
+
+
+def _tool_type(tool: object) -> str:
+    return str(tool.get("type") or "").strip() if isinstance(tool, dict) else ""
+
+
+def has_web_search_tool(body: dict[str, Any]) -> bool:
+    tools = body.get("tools")
+    if isinstance(tools, list):
+        return any(_tool_type(tool) in WEB_SEARCH_TOOL_TYPES for tool in tools)
+    tool_choice = body.get("tool_choice")
+    return _tool_type(tool_choice) in WEB_SEARCH_TOOL_TYPES
+
+
+def is_web_search_chat_request(body: dict[str, Any]) -> bool:
+    model = str(body.get("model") or "").strip()
+    return (
+        has_web_search_tool(body)
+        or isinstance(body.get("web_search_options"), dict)
+        or any(
+            model == prefix or model.startswith(f"{prefix}-")
+            for prefix in SEARCH_CHAT_MODEL_PREFIXES
+        )
+    )
+
+
+def has_unsupported_tools(body: dict[str, Any], allowed_types: set[str]) -> bool:
+    tools = body.get("tools")
+    if not isinstance(tools, list):
+        return False
+    return any(_tool_type(tool) not in allowed_types for tool in tools if isinstance(tool, dict))
+
+
+def message_text(value: object) -> str:
+    if isinstance(value, str):
+        return value.strip()
+    if isinstance(value, list):
+        parts: list[str] = []
+        for item in value:
+            if isinstance(item, str):
+                text = item.strip()
+            elif isinstance(item, dict):
+                text = str(item.get("text") or item.get("input_text") or "").strip()
+            else:
+                text = ""
+            if text:
+                parts.append(text)
+        return "\n".join(parts).strip()
+    return ""
+
+
+def search_query_from_messages(messages: list[dict[str, Any]]) -> str:
+    for message in reversed(messages):
+        if str(message.get("role") or "").strip().lower() != "user":
+            continue
+        text = message_text(message.get("content"))
+        if text:
+            return text
+    return ""
+
+
+def _readable_annotation_part(parts: list[str]) -> str:
+    for part in parts:
+        value = part.strip()
+        lower = value.lower()
+        if value and not (
+            lower.startswith(("turn", "source", "sources"))
+            or re.fullmatch(r"\d+", value)
+        ):
+            return value
+    return ""
+
+
+def clean_search_text(text: str) -> str:
+    def replace_annotation(match: re.Match[str]) -> str:
+        parts = [part.strip() for part in match.group(1).split("\ue202")]
+        kind = (parts[0] if parts else "").lower()
+        data = parts[1:]
+        if kind == "url":
+            label = data[0] if data else ""
+            url = data[1] if len(data) > 1 else ""
+            if label and url.startswith(("http://", "https://")):
+                return f"{label} ({url})"
+            return label or url
+        if kind == "cite":
+            return _readable_annotation_part(data)
+        return _readable_annotation_part(data)
+
+    text = re.sub(r"\ue200([^\ue201]*)\ue201", replace_annotation, text)
+    text = re.sub(r"\ue200[^\ue201]*$", "", text)
+    return re.sub(r"\s+([.,;:!?])", r"\1", text).strip()
+
+
+def normalized_sources(result: dict[str, Any]) -> list[dict[str, str]]:
+    sources = result.get("sources")
+    if not isinstance(sources, list):
+        return []
+    output: list[dict[str, str]] = []
+    seen: set[str] = set()
+    for item in sources:
+        if not isinstance(item, dict):
+            continue
+        url = str(item.get("url") or "").strip()
+        title = str(item.get("title") or "").strip()
+        snippet = str(item.get("snippet") or "").strip()
+        if not url or url in seen:
+            continue
+        seen.add(url)
+        output.append({"title": title, "url": url, "snippet": snippet})
+    return output
+
+
+def text_with_url_citations(result: dict[str, Any]) -> tuple[str, list[dict[str, Any]]]:
+    text = clean_search_text(str(result.get("answer") or ""))
+    annotations: list[dict[str, Any]] = []
+    sources = normalized_sources(result)
+    if sources:
+        text = text.rstrip()
+        if text:
+            text += "\n\n"
+        text += "Sources:\n"
+        for index, source in enumerate(sources, start=1):
+            title = source["title"] or source["url"]
+            line_prefix = f"{index}. {title}"
+            text += line_prefix
+            if source["url"]:
+                if source["title"]:
+                    text += " - "
+                start = len(text)
+                text += source["url"]
+                annotations.append({
+                    "type": "url_citation",
+                    "start_index": start,
+                    "end_index": len(text),
+                    "url": source["url"],
+                    "title": source["title"] or source["url"],
+                })
+            text += "\n"
+    return text.strip(), annotations
+
+
+def run_web_search(query: str) -> dict[str, Any]:
+    token = account_service.get_text_access_token()
+    result = OpenAIBackendAPI(token).search(query)
+    account_service.mark_text_used(token)
+    return result

test/test_chat_completion_cache.py

@@ -237,6 +237,142 @@ class ChatCompletionCacheTests(unittest.TestCase):
         self.assertEqual(messages[0]["role"], "system")
         self.assertIn("cannot execute local tools", str(messages[0]["content"]))
 
+    def test_responses_web_search_tool_returns_search_output(self) -> None:
+        search_result = {
+            "answer": "Latest answer.",
+            "sources": [{"title": "Example", "url": "https://example.com/news", "snippet": "Snippet"}],
+        }
+        body = {
+            "model": "auto",
+            "input": "latest example news",
+            "tools": [{"type": "web_search"}],
+        }
+
+        with mock.patch("services.protocol.openai_v1_response.run_web_search", return_value=search_result) as search:
+            response = openai_v1_response.handle(body)
+
+        search.assert_called_once_with("latest example news")
+        self.assertEqual(response["output"][0]["type"], "web_search_call")
+        self.assertEqual(response["output"][0]["status"], "completed")
+        self.assertEqual(response["output"][0]["action"]["query"], "latest example news")
+        message = response["output"][1]
+        self.assertEqual(message["type"], "message")
+        content = message["content"][0]
+        self.assertIn("Latest answer.", content["text"])
+        self.assertEqual(content["annotations"][0]["type"], "url_citation")
+        self.assertEqual(content["annotations"][0]["url"], "https://example.com/news")
+
+    def test_responses_web_search_tool_streams_search_events(self) -> None:
+        search_result = {
+            "answer": "Streamed search answer.",
+            "sources": [{"title": "Example", "url": "https://example.com/stream", "snippet": ""}],
+        }
+        body = {
+            "model": "auto",
+            "stream": True,
+            "input": "stream search",
+            "tools": [{"type": "web_search_preview"}],
+        }
+
+        with mock.patch("services.protocol.openai_v1_response.run_web_search", return_value=search_result):
+            events = list(openai_v1_response.handle(body))
+
+        event_types = [event["type"] for event in events]
+        self.assertIn("response.web_search_call.in_progress", event_types)
+        self.assertIn("response.web_search_call.searching", event_types)
+        self.assertIn("response.web_search_call.completed", event_types)
+        completed = events[-1]["response"]
+        self.assertEqual(completed["output"][0]["type"], "web_search_call")
+        self.assertEqual(completed["output"][1]["type"], "message")
+
+    def test_responses_versioned_web_search_tool_returns_search_output(self) -> None:
+        search_result = {
+            "answer": "Versioned search answer.",
+            "sources": [{"title": "Example", "url": "https://example.com/versioned", "snippet": ""}],
+        }
+        body = {
+            "model": "auto",
+            "input": "versioned search",
+            "tools": [{"type": "web_search_preview_2025_03_11"}],
+        }
+
+        with mock.patch("services.protocol.openai_v1_response.run_web_search", return_value=search_result) as search:
+            response = openai_v1_response.handle(body)
+
+        search.assert_called_once_with("versioned search")
+        self.assertEqual(response["output"][0]["type"], "web_search_call")
+        self.assertIn("Versioned search answer.", response["output"][1]["content"][0]["text"])
+
+    def test_chat_completions_web_search_tool_returns_search_answer(self) -> None:
+        search_result = {
+            "answer": "Chat search answer.",
+            "sources": [{"title": "Example", "url": "https://example.com/chat", "snippet": ""}],
+        }
+        body = {
+            "model": "auto",
+            "messages": [{"role": "user", "content": "search chat"}],
+            "tools": [{"type": "web_search"}],
+        }
+
+        with mock.patch("services.protocol.openai_v1_chat_complete.run_web_search", return_value=search_result) as search:
+            response = openai_v1_chat_complete.handle(body)
+
+        search.assert_called_once_with("search chat")
+        message = response["choices"][0]["message"]
+        self.assertIn("Chat search answer.", message["content"])
+        self.assertEqual(message["annotations"][0]["type"], "url_citation")
+        self.assertEqual(message["annotations"][0]["url_citation"]["url"], "https://example.com/chat")
+
+    def test_chat_completions_web_search_options_trigger_search(self) -> None:
+        search_result = {
+            "answer": "Options search answer.",
+            "sources": [{"title": "Example", "url": "https://example.com/options", "snippet": ""}],
+        }
+        body = {
+            "model": "auto",
+            "messages": [{"role": "user", "content": "search options"}],
+            "web_search_options": {"search_context_size": "low"},
+        }
+
+        with mock.patch("services.protocol.openai_v1_chat_complete.run_web_search", return_value=search_result) as search:
+            response = openai_v1_chat_complete.handle(body)
+
+        search.assert_called_once_with("search options")
+        self.assertIn("Options search answer.", response["choices"][0]["message"]["content"])
+
+    def test_chat_completions_search_model_triggers_search(self) -> None:
+        search_result = {
+            "answer": "Search model answer.",
+            "sources": [{"title": "Example", "url": "https://example.com/model", "snippet": ""}],
+        }
+        body = {
+            "model": "gpt-5-search-api-2026-06-01",
+            "messages": [{"role": "user", "content": "search model"}],
+        }
+
+        with mock.patch("services.protocol.openai_v1_chat_complete.run_web_search", return_value=search_result) as search:
+            response = openai_v1_chat_complete.handle(body)
+
+        search.assert_called_once_with("search model")
+        self.assertEqual(response["model"], "gpt-5-search-api-2026-06-01")
+        self.assertIn("Search model answer.", response["choices"][0]["message"]["content"])
+
+    def test_chat_completions_search_like_model_does_not_trigger_search(self) -> None:
+        body = {
+            "model": "gpt-5-search-apiary",
+            "messages": [{"role": "user", "content": "not actually a search model"}],
+        }
+
+        with (
+            mock.patch("services.protocol.openai_v1_chat_complete.run_web_search") as search,
+            mock.patch("services.protocol.openai_v1_chat_complete.text_backend", return_value=object()),
+            mock.patch("services.protocol.openai_v1_chat_complete.collect_text", return_value="plain text answer"),
+        ):
+            response = openai_v1_chat_complete.handle(body)
+
+        search.assert_not_called()
+        self.assertIn("plain text answer", response["choices"][0]["message"]["content"])
+
     def test_chat_completions_accepts_remote_image_url(self) -> None:
         class FakeImageResponse:
             status_code = 200