core/tools+config: search through the DeepSeek Search API

src/plyngent/cli/state.py

@@ -25,7 +25,7 @@ from plyngent.cli.models_source import (
 )
 from plyngent.config import EffectiveProvider, ReasoningConfig, resolve_effective_provider, resolve_reasoning
 from plyngent.memory.database.store import normalize_workspace
-from plyngent.runtime import create_client
+from plyngent.runtime import create_client, normalize_http_timeout
 from plyngent.tools import InstanceState, SessionState
 from plyngent.tools.view import MemoryViewStore, session_data_view
 
@@ -391,11 +391,11 @@ class ReplState:
         (see :func:`~plyngent.config.routing.local_web_search`).
         """
         from plyngent.config.routing import (
-            deepseek_search_base_url,
+            deepseek_api_base_url,
             local_web_search,
         )
-        from plyngent.lmproto.anthropic.config import AnthropicConfig
+        from plyngent.lmproto.deepseek.search import DeepseekSearchClient
-        from plyngent.lmproto.deepseek.anthropic.search import DeepseekSearchClient
+        from plyngent.lmproto.openai_compatible.config import OpenAIConfig
         from plyngent.tools.net.bing import BingSearchBackend
         from plyngent.tools.net.search import (
             DeepseekSearchBackend,
@@ -403,7 +403,7 @@ class ReplState:
         )
 
         provider = self.effective_provider
-        if not local_web_search(provider) or not self.model:
+        if not local_web_search(provider):
             return None
         backends: list[SearchBackend] = []
         for source in provider.search_sources:
@@ -411,12 +411,12 @@ class ReplState:
                 backends.append(
                     DeepseekSearchBackend(
                         DeepseekSearchClient(
-                            AnthropicConfig(
+                            OpenAIConfig(
-                                api_key=provider.access_key_or_token,
+                                access_key_or_token=provider.access_key_or_token,
-                                base_url=deepseek_search_base_url(provider),
+                                base_url=deepseek_api_base_url(provider),
+                                timeout=normalize_http_timeout(provider.timeout),
+                            )
                         )
-                        ),
-                        model=self.model,
                     )
                 )
             elif source == "bing":

src/plyngent/config/routing.py

@@ -98,27 +98,34 @@ def wants_web_search(effective: EffectiveProvider) -> bool:
 def local_web_search(effective: EffectiveProvider) -> bool:
     """True when this provider searches through plyngent's own ``web_search`` tool.
 
-    DeepSeek runs a hosted search only on its Anthropic-compatible surface
+    ``search_sources`` owns the local tool; ``provider_tools`` owns the hosted
-    (``convention = "anthropic"``): the Responses and chat-completions surfaces
+    one. DeepSeek's Anthropic-compatible surface can run the search inside a
-    ignore a hosted tool — a live probe answers "I can't browse the web" with
+    response, so that combination keeps the hosted tool and the local one stays
-    either ``{"type": "web_search"}`` or the server-tool shape — so a DeepSeek
+    out of the way; turning hosted tools off (``provider_tools = []``) brings
-    provider that talks them searches through
+    the local tool back there too. Every other surface ignores a hosted search,
-    :func:`~plyngent.tools.net.search.build_web_search_tool`, which asks the
+    so the local tool is the only one it can use.
-    sources named by ``search_sources`` behind the scenes.
     """
-    if effective.preset != "deepseek" or effective.convention == "anthropic":
+    if effective.preset != "deepseek" or not effective.search_sources:
         return False
-    if not effective.search_sources:
+    return not (effective.convention == "anthropic" and wants_web_search(effective))
-        return False
-    return wants_web_search(effective)
 
 
-def deepseek_search_base_url(effective: EffectiveProvider) -> str:
+def deepseek_api_base_url(effective: EffectiveProvider) -> str:
-    """The endpoint a DeepSeek search turn runs on (its Anthropic-compatible base)."""
+    """The platform base URL shared by model and tool APIs.
-    url = effective.url.rstrip("/")
+
-    if url.endswith("/anthropic"):
+    The Anthropic convention lives under ``/anthropic`` and OpenAI-compatible
-        return url
+    configs often carry a ``/v1`` suffix; neither belongs in the base that the
-    return DEFAULT_DEEPSEEK_ANTHROPIC_BASE_URL
+    Search API client appends ``/v1/tools/search`` to.
+    """
+    url = (effective.url or DEFAULT_DEEPSEEK_BASE_URL).strip().rstrip("/")
+    stripped = True
+    while stripped:
+        stripped = False
+        for suffix in ("/anthropic", "/v1"):
+            if url.endswith(suffix):
+                url = url[: -len(suffix)].rstrip("/")
+                stripped = True
+    return url or DEFAULT_DEEPSEEK_BASE_URL
 
 
 def resolve_effective_provider(provider: Provider, *, model: str | None = None) -> EffectiveProvider:

src/plyngent/lmproto/deepseek/anthropic/search.py

@@ -1,148 +0,0 @@
-"""Server-side web search on DeepSeek's Anthropic-compatible endpoint.
-
-DeepSeek runs ``web_search`` *inside* a turn: the request carries the server
-tool, and the response comes back with a ``server_tool_use`` block per query
-plus the ``web_search_tool_result`` that query produced — nothing is executed
-here. Results only carry a title, a URL and an ``encrypted_content`` the index
-keeps the page body in, which is why the hit list is all a caller can pass on.
-
-Only this surface searches: the Responses and chat-completions surfaces ignore a
-hosted search tool (a live probe answers "I can't browse the web" whether the
-tool arrives as ``{"type": "web_search"}`` or in the server-tool shape), so a
-DeepSeek provider that talks them searches through
-:func:`~plyngent.tools.net.search.build_web_search_tool` instead.
-"""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, Any, cast
-
-import msgspec
-import niquests
-
-if TYPE_CHECKING:
-    from plyngent.lmproto.anthropic.config import AnthropicConfig
-
-# A search turn stops as soon as the search came back: the answer that follows is
-# thrown away, so the budget only has to cover the thinking that picks the query
-# and the results. A live probe with 64 tokens spent all of it before searching
-# and returned no results at all.
-DEFAULT_SEARCH_MAX_TOKENS = 512
-DEFAULT_SEARCH_MAX_USES = 3
-_HTTP_ERROR = 400
-_SEARCH_TOOL: dict[str, Any] = {"type": "web_search_20250305", "name": "web_search"}
-
-
-class SearchHit(msgspec.Struct, frozen=True, omit_defaults=True):
-    """One page a search source matched.
-
-    DeepSeek's index keeps the page bodies server-side, so its hits carry a
-    title and a URL only; a source that ships snippets (Bing's RSS) fills
-    ``snippet`` too.
-    """
-
-    title: str = ""
-    url: str = ""
-    snippet: str = ""
-
-
-def hits_from_content(content: object) -> list[SearchHit]:
-    """Collect the search hits from a response's content blocks (deduplicated)."""
-    if not isinstance(content, list):
-        return []
-    hits: list[SearchHit] = []
-    seen: set[str] = set()
-    for block_obj in cast("list[object]", content):
-        if not isinstance(block_obj, dict):
-            continue
-        block = cast("dict[str, object]", block_obj)
-        if block.get("type") != "web_search_tool_result":
-            continue
-        entries = block.get("content")
-        if not isinstance(entries, list):
-            continue
-        for entry_obj in cast("list[object]", entries):
-            if not isinstance(entry_obj, dict):
-                continue
-            entry = cast("dict[str, object]", entry_obj)
-            if entry.get("type") != "web_search_result":
-                continue
-            url = entry.get("url")
-            if not isinstance(url, str) or not url or url in seen:
-                continue
-            seen.add(url)
-            title = entry.get("title")
-            hits.append(SearchHit(title=title if isinstance(title, str) else "", url=url))
-    return hits
-
-
-class DeepseekSearchClient:
-    """``POST /messages`` with a hosted ``web_search`` tool, read for its hits."""
-
-    session: niquests.AsyncSession
-    encoder: msgspec.json.Encoder
-    _api_key: str
-    _api_version: str
-
-    def __init__(self, config: AnthropicConfig) -> None:
-        self.session = niquests.AsyncSession(base_url=config.base_url, timeout=config.timeout)
-        self.encoder = msgspec.json.Encoder()
-        self._api_key = config.api_key
-        self._api_version = config.anthropic_version
-
-    def _headers(self) -> dict[str, str]:
-        return {
-            "Content-Type": "application/json",
-            "x-api-key": self._api_key,
-            "anthropic-version": self._api_version,
-        }
-
-    async def search(
-        self,
-        query: str,
-        *,
-        model: str,
-        max_uses: int = DEFAULT_SEARCH_MAX_USES,
-        max_tokens: int = DEFAULT_SEARCH_MAX_TOKENS,
-    ) -> list[SearchHit]:
-        """Run *query* through the server's search; return the pages it matched."""
-        payload: dict[str, Any] = {
-            "model": model,
-            "max_tokens": max_tokens,
-            "messages": [{"role": "user", "content": query}],
-            "tools": [{**_SEARCH_TOOL, "max_uses": max_uses}],
-        }
-        resp = await self.session.post(
-            "/messages",
-            data=self.encoder.encode(payload),
-            headers=self._headers(),
-            stream=False,
-        )
-        body = await self._read_body(resp)
-        parsed = msgspec.json.decode(body, type=dict[str, object])
-        return hits_from_content(parsed.get("content"))
-
-    async def _read_body(self, resp: object) -> bytes:
-        from plyngent.lmproto.openai_compatible.client import (
-            http_error_message,
-            read_response_body,
-        )
-
-        body = await read_response_body(resp)
-        status = int(getattr(resp, "status_code", 0) or 0)
-        if status >= _HTTP_ERROR:
-            detail = http_error_message(status, body, what="search") or f"search HTTP {status}"
-            raise RuntimeError(detail)
-        if body is None:
-            msg = "search response body is empty"
-            raise RuntimeError(msg)
-        return body.encode() if isinstance(body, str) else bytes(body)
-
-
-__all__ = [
-    "DEFAULT_SEARCH_MAX_TOKENS",
-    "DEFAULT_SEARCH_MAX_USES",
-    "DeepseekSearchClient",
-    "SearchHit",
-    "hits_from_content",
-]

src/plyngent/tools/net/bing.py

@@ -12,7 +12,7 @@ from __future__ import annotations
 import urllib.parse
 import xml.etree.ElementTree as ET
 
-from plyngent.lmproto.deepseek.anthropic.search import SearchHit
+from plyngent.lmproto.search import SearchHit
 from plyngent.tools.net.client import http_fetch
 
 SEARCH_URL = "https://www.bing.com/search"

src/plyngent/tools/net/search.py

@@ -1,13 +1,15 @@
 """``web_search``: the model's way to search when its surface cannot host one.
 
-DeepSeek only searches on its Anthropic-compatible surface, so a DeepSeek
+DeepSeek hosts a search only on its Anthropic-compatible surface, so a DeepSeek
-provider that talks the Responses or chat-completions surface gets this local
+provider that talks the Responses or chat-completions surface — or one that
-tool instead of a hosted one (see ``config.routing.local_web_search``): the
+turns hosted tools off — gets this local tool instead (see
-model calls it like any other tool and the handler asks the configured search
+``config.routing.local_web_search``): the model calls it like any other tool and
-sources behind the scenes — ``provider.search_sources``, in fallback order, so
+the handler asks the configured search sources behind the scenes —
-the next source is tried when one fails or finds nothing. A hit is a title, a
+``provider.search_sources``, in fallback order, so the next source is tried when
-URL and (when the source has one) a snippet: read a page with ``fetch`` before
+one fails or finds nothing. DeepSeek's own source is the Search API
-relying on its content.
+(``POST /v1/tools/search``), which answers in one request instead of a model
+round. A hit is a title, a URL and (when the source has one) a snippet: read a
+page with ``fetch`` before relying on its content.
 """
 
 from __future__ import annotations
@@ -20,10 +22,8 @@ from plyngent.agent.tools import ToolDefinition, ToolTag, schema_from_callable
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
-    from plyngent.lmproto.deepseek.anthropic.search import (
+    from plyngent.lmproto.deepseek.search import DeepseekSearchClient
-        DeepseekSearchClient,
+    from plyngent.lmproto.search import SearchHit
-        SearchHit,
-    )
 
 WEB_SEARCH_TOOL_NAME = "web_search"
 DEFAULT_MAX_HITS = 8
@@ -36,17 +36,15 @@ class SearchBackend(Protocol):
 
 
 class DeepseekSearchBackend:
-    """DeepSeek's own index: one server-side search turn per query."""
+    """DeepSeek's own index: one Search API request per query."""
 
     _client: DeepseekSearchClient
-    _model: str
 
-    def __init__(self, client: DeepseekSearchClient, *, model: str) -> None:
+    def __init__(self, client: DeepseekSearchClient) -> None:
         self._client = client
-        self._model = model
 
     async def search(self, query: str) -> Sequence[SearchHit]:
-        return await self._client.search(query, model=self._model)
+        return await self._client.search(query)
 
 
 async def search_in_order(

tests/test_config/test_routing.py

@@ -7,7 +7,7 @@ from dataclasses import replace
 from plyngent.config.models import DeepseekProvider
 from plyngent.config.routing import (
     EffectiveProvider,
-    deepseek_search_base_url,
+    deepseek_api_base_url,
     local_web_search,
     resolve_effective_provider,
     wants_web_search,
@@ -46,8 +46,14 @@ def test_chat_and_responses_conventions_use_the_local_tool() -> None:
     assert local_web_search(_effective(convention="anthropic")) is False
 
 
-def test_without_provider_tools_there_is_no_search() -> None:
+def test_provider_tools_only_gate_the_hosted_search() -> None:
-    assert local_web_search(_effective(convention="responses", tools=[])) is False
+    # ``search_sources`` owns the local tool; ``provider_tools`` owns the hosted
+    # one. Turning hosted tools off must not take local search with it.
+    assert local_web_search(_effective(convention="responses", tools=[])) is True
+    assert local_web_search(_effective(convention="anthropic", tools=[])) is True
+    # With hosted tools on, the Anthropic surface keeps the hosted search and
+    # the local tool stays out of the way.
+    assert local_web_search(_effective(convention="anthropic")) is False
 
 
 def test_without_search_sources_there_is_no_local_tool() -> None:
@@ -58,9 +64,17 @@ def test_without_search_sources_there_is_no_local_tool() -> None:
     assert _effective(convention="responses").search_sources == ["deepseek", "bing"]
 
 
-def test_search_base_url_prefers_an_anthropic_endpoint() -> None:
+def test_api_base_url_drops_the_convention_and_version_paths() -> None:
-    assert deepseek_search_base_url(_effective(convention="responses")) == "https://api.deepseek.com/anthropic"
+    # The Search API appends ``/v1/tools/search`` itself, so neither the
+    # Anthropic convention's ``/anthropic`` nor an OpenAI-compatible ``/v1``
+    # may survive in the base.
+    assert deepseek_api_base_url(_effective(convention="responses")) == "https://api.deepseek.com"
+    assert deepseek_api_base_url(_effective(convention="anthropic")) == "https://api.deepseek.com"
     assert (
-        deepseek_search_base_url(_effective(convention="responses", url="https://gateway.example/anthropic"))
+        deepseek_api_base_url(_effective(convention="anthropic", url="https://gateway.example/api/anthropic"))
-        == "https://gateway.example/anthropic"
+        == "https://gateway.example/api"
+    )
+    assert (
+        deepseek_api_base_url(_effective(convention="responses", url="https://gateway.example/v1"))
+        == "https://gateway.example"
     )

tests/test_lmproto/test_deepseek_anthropic_search.py

@@ -1,115 +0,0 @@
-"""DeepSeek server-side search backend (Anthropic-compatible endpoint)."""
-
-from __future__ import annotations
-
-import msgspec
-import pytest
-
-from plyngent.lmproto.anthropic.config import AnthropicConfig
-from plyngent.lmproto.deepseek.anthropic.search import (
-    DeepseekSearchClient,
-    SearchHit,
-    hits_from_content,
-)
-
-
-def test_hits_from_content_dedupes_and_skips_noise() -> None:
-    content = [
-        {"type": "thinking", "thinking": "search"},
-        {
-            "type": "web_search_tool_result",
-            "tool_use_id": "call_1",
-            "content": [
-                {"type": "web_search_result", "title": "First", "url": "https://a.example/x"},
-                {"type": "web_search_result", "title": "First again", "url": "https://a.example/x"},
-                {"type": "web_search_result", "url": "https://b.example/y"},
-                {"type": "web_search_result", "title": "no url"},
-            ],
-        },
-        {
-            "type": "web_search_tool_result",
-            "tool_use_id": "call_2",
-            "content": [{"type": "web_search_tool_result_error", "error_code": "max_uses_exceeded"}],
-        },
-    ]
-    hits = hits_from_content(content)
-    assert hits == [
-        SearchHit(title="First", url="https://a.example/x"),
-        SearchHit(title="", url="https://b.example/y"),
-    ]
-    assert hits_from_content("not a list") == []
-
-
-class _Resp:
-    def __init__(self, status_code: int, body: bytes) -> None:
-        self.status_code = status_code
-        self.content = body
-
-
-def _search_body() -> bytes:
-    return msgspec.json.encode(
-        {
-            "id": "msg_1",
-            "content": [
-                {
-                    "type": "server_tool_use",
-                    "id": "call_00_1",
-                    "name": "web_search",
-                    "input": {"query": "DeepSeek-V4"},
-                },
-                {
-                    "type": "web_search_tool_result",
-                    "tool_use_id": "call_00_1",
-                    "content": [
-                        {
-                            "type": "web_search_result",
-                            "title": "DeepSeek-V4 preview",
-                            "url": "https://api-docs.deepseek.com/news/news260424/",
-                            "encrypted_content": "opaque",
-                        }
-                    ],
-                },
-            ],
-            "usage": {"input_tokens": 9932, "output_tokens": 377},
-        }
-    )
-
-
-@pytest.mark.asyncio
-async def test_search_sends_the_hosted_tool_and_reads_its_hits(monkeypatch: pytest.MonkeyPatch) -> None:
-    client = DeepseekSearchClient(AnthropicConfig(api_key="sk-test", base_url="https://api.deepseek.com/anthropic"))
-    sent: dict[str, object] = {}
-
-    async def fake_post(path: str, **kwargs: object) -> _Resp:
-        sent["path"] = path
-        sent.update(kwargs)
-        return _Resp(200, _search_body())
-
-    monkeypatch.setattr(client.session, "post", fake_post)
-    hits = await client.search("DeepSeek-V4", model="deepseek-flash")
-
-    assert hits == [SearchHit(title="DeepSeek-V4 preview", url="https://api-docs.deepseek.com/news/news260424/")]
-    assert sent["path"] == "/messages"
-    assert sent["stream"] is False
-    data = sent["data"]
-    assert isinstance(data, bytes)
-    payload = msgspec.json.decode(data, type=dict[str, object])
-    assert payload["tools"] == [{"type": "web_search_20250305", "name": "web_search", "max_uses": 3}]
-    # A search turn stops as soon as it has searched; the answer is not needed.
-    assert payload["max_tokens"] == 512
-    assert payload["messages"] == [{"role": "user", "content": "DeepSeek-V4"}]
-    headers = sent["headers"]
-    assert isinstance(headers, dict)
-    assert headers["x-api-key"] == "sk-test"
-
-
-@pytest.mark.asyncio
-async def test_search_raises_with_the_endpoint_detail(monkeypatch: pytest.MonkeyPatch) -> None:
-    client = DeepseekSearchClient(AnthropicConfig(api_key="sk-test", base_url="https://api.deepseek.com/anthropic"))
-
-    async def fake_post(path: str, **kwargs: object) -> _Resp:
-        return _Resp(422, b'{"error": {"message": "tools[0]: unknown variant"}}')
-
-    monkeypatch.setattr(client.session, "post", fake_post)
-    with pytest.raises(RuntimeError, match="unknown variant"):
-        _ = await client.search("DeepSeek-V4", model="deepseek-flash")

tests/test_tools/test_bing.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import pytest
 
-from plyngent.lmproto.deepseek.anthropic.search import SearchHit
+from plyngent.lmproto.search import SearchHit
 from plyngent.tools.net.bing import parse_bing_rss
 
 DOCUMENT = """<?xml version="1.0" encoding="utf-8" ?>

tests/test_tools/test_search.py

@@ -2,12 +2,10 @@
 
 from __future__ import annotations
 
-from typing import Any
-
 import pytest
 
 from plyngent.agent.tools import ToolTag
-from plyngent.lmproto.deepseek.anthropic.search import SearchHit
+from plyngent.lmproto.search import SearchHit
 from plyngent.tools.net.search import (
     WEB_SEARCH_TOOL_NAME,
     build_web_search_tool,
@@ -93,25 +91,21 @@ async def test_handler_without_sources_says_so() -> None:
     assert "no source configured" in await definition.handler(query="q")
 
 
-def test_deepseek_backend_passes_the_model_through() -> None:
+@pytest.mark.asyncio
+async def test_deepseek_backend_asks_the_search_api() -> None:
     from plyngent.tools.net.search import DeepseekSearchBackend
 
     class _Client:
         def __init__(self) -> None:
-            self.seen: list[tuple[str, str]] = []
+            self.queries: list[str] = []
 
-        async def search(self, query: str, *, model: str, **kwargs: object) -> list[SearchHit]:
+        async def search(self, query: str) -> list[SearchHit]:
-            self.seen.append((query, model))
+            self.queries.append(query)
             return _hits(("T", "https://example.com"))
 
-    client: Any = _Client()
+    client = _Client()
-    backend = DeepseekSearchBackend(client, model="deepseek-flash")
+    backend = DeepseekSearchBackend(client)  # type: ignore[arg-type] — structural stub
 
-    async def run() -> list[SearchHit]:
+    hits = await backend.search("q")
-        return list(await backend.search("q"))
-
-    import asyncio
-
-    hits = asyncio.run(run())
     assert [hit.url for hit in hits] == ["https://example.com"]
-    assert client.seen == [("q", "deepseek-flash")]
+    assert client.queries == ["q"]