diff --git a/docs/mkdocs/en/tool.md b/docs/mkdocs/en/tool.md index 48361fba..9bde6ce7 100644 --- a/docs/mkdocs/en/tool.md +++ b/docs/mkdocs/en/tool.md @@ -2532,21 +2532,23 @@ The example covers the following scenarios: `WebSearchTool` is the built-in **public web search tool** in the trpc-agent-python framework. When an Agent needs to answer questions about "latest developments / version numbers / events / definitions / factual topics" that fall beyond the model's knowledge cutoff, it can use this tool to call mainstream search-engine retrieval APIs, obtain structured results containing titles, URLs, and snippets, and then list all citations in a `Sources:` section as Markdown hyperlinks according to the prescribed format. -This tool uses a **pluggable provider** design and currently includes two built-in backends: +This tool uses a **pluggable provider** design and currently includes four built-in backends: - **`duckduckgo` (default)**: DuckDuckGo Instant Answer API, which **does not require an API key**. It returns DDG-curated instant answers / abstracts / definitions and related topics, making it suitable for encyclopedia, definition, and fact-based queries. Note that it does not return full real-time web search results, but rather DDG's curated result set - **`google`**: Google Custom Search (CSE) JSON API, which requires `api_key` and `engine_id` (that is, the CSE `cx`). It returns real public web search results and supports native CSE parameters such as `siteSearch`, `hl` (language), `safe` (SafeSearch), and `dateRestrict` (freshness) +- **`tavily`**: Tavily Search API, which requires `api_key` (or the `TAVILY_API_KEY` environment variable). It returns LLM-oriented web results and can optionally return image URLs (call-time parameter `include_images=true`) +- **`youcom`**: You.com `you-search` MCP tool over stateless JSON-RPC (streamable HTTP). It **runs keyless by default** through the free-profile endpoint (`https://api.you.com/mcp?profile=free`), just like the DDG backend — zero configuration. Setting `YDC_API_KEY` (or the `api_key` constructor argument) automatically switches requests to the authenticated endpoint. It returns web and news hits whose snippets fall back from the hit `description` to You.com's canonical `snippets` field and then to query-relevant `contents.highlights`. Only the tool's own headers reach the endpoint, so credentials configured on a shared `http_client` (a gateway token, a cookie jar) are never forwarded to You.com On top of that, `WebSearchTool` also provides **domain allowlist/blocklist filtering, URL normalization and deduplication, result truncation, mandatory citation-policy injection, and HTTP connection-pool reuse**, helping you integrate network retrieval into LLM Agents in a stable and controllable way for production environments. ### Features -- **Dual provider support**: switch between `duckduckgo` (keyless, suitable for definitions/encyclopedia/facts) and `google` (requires CSE API credentials and supports real public web search) through the `provider` parameter, while exposing a consistent `FunctionDeclaration` to the LLM +- **Multi-provider support**: switch between `duckduckgo` (keyless, suitable for definitions/encyclopedia/facts), `google` (requires CSE API credentials and supports real public web search), `tavily` (LLM-oriented search with optional images), and `youcom` (You.com MCP, keyless via the free profile or authenticated via `YDC_API_KEY`) through the `provider` parameter, while exposing a consistent `FunctionDeclaration` to the LLM - **Domain allowlist/blocklist**: the LLM may fill in `allowed_domains` or `blocked_domains` at call time (the two are mutually exclusive). The tool performs **subdomain-aware** matching (`www.` is stripped, and `python.org` also matches `docs.python.org`). For Google, a single allowed domain takes the server-side `siteSearch` fast path, while multiple domains automatically fall back to client-side filtering - **URL normalization and deduplication**: `dedup_urls=True` (default) merges duplicate hits by a normalized scheme/host/path key, preventing the same source from appearing multiple times in the `Sources:` section. Set it to `False` to preserve the raw recall list for downstream re-rankers, diversified sampling, or offline evaluation - **Result truncation**: `results_num`, `snippet_len`, and `title_len` control the number of returned items and the character limits for each snippet and title. All parameters are clamped to `[1, _MAX_*]` to avoid misconfiguration that would exceed the context window. The LLM can further control the returned count at call time via the `count` parameter - **Mandatory citation policy**: during `process_request`, the tool automatically appends instructions to the LLM and **strictly requires** that: (1) the final answer must end with `Sources:` section listing the URLs returned by the tool in `[Title](URL)` format; (2) the Agent must not fabricate URLs; and (3) for "latest/recent" queries, the **current year and month** must be included in the search input to reduce stale-year hallucinations -- **Provider-native parameter passthrough**: `ddg_extra_params` and `google_extra_params` let you pin provider-specific advanced parameters at the Agent level, such as Google CSE's `safe`, `dateRestrict`, `gl`, and `cr`, so they are automatically included in every tool invocation without exposing additional fields in the `FunctionDeclaration` +- **Provider-native parameter passthrough**: `ddg_extra_params`, `google_extra_params`, `tavily_extra_params`, and `youcom_extra_params` let you pin provider-specific advanced parameters at the Agent level, such as Google CSE's `safe`, `dateRestrict`, `gl`, and `cr`, Tavily's `search_depth`, or You.com's `freshness` and `safesearch`, so they are automatically included in every tool invocation without exposing additional fields in the `FunctionDeclaration` - **Shared `httpx` connection pool**: by passing a prebuilt `httpx.AsyncClient` through the `http_client` constructor argument, you can reuse the connection pool across multiple Agents or multiple invocations. The caller is responsible for managing its lifecycle (the tool will not call `aclose`) - **Structured output**: always returns a `WebSearchResult` containing `query`, `provider`, `results: List[{title, url, snippet}]`, and `summary`, making it easier both for the LLM to compose citations and for downstream re-ranking or RAG pipelines @@ -2554,13 +2556,13 @@ On top of that, `WebSearchTool` also provides **domain allowlist/blocklist filte | Parameter | Type | Default | Description | |------|------|--------|------| -| `provider` | `Literal["duckduckgo", "google"]` | `"duckduckgo"` | Search backend. `google` requires both `api_key` and `engine_id`; otherwise a not-configured message is returned at call time | -| `api_key` | `Optional[str]` | `None` | Google CSE API key. If omitted, falls back to the `GOOGLE_CSE_API_KEY` environment variable | +| `provider` | `Literal["duckduckgo", "google", "tavily", "youcom"]` | `"duckduckgo"` | Search backend. `google` requires both `api_key` and `engine_id`; `tavily` requires `api_key` (otherwise a not-configured message is returned at call time); `youcom` runs keyless by default (free profile) and switches to the authenticated endpoint when `YDC_API_KEY` is set | +| `api_key` | `Optional[str]` | `None` | Provider API key. Google falls back to the `GOOGLE_CSE_API_KEY` environment variable; Tavily to `TAVILY_API_KEY`; You.com to `YDC_API_KEY` (optional) | | `engine_id` | `Optional[str]` | `None` | Google CSE engine ID (that is, `cx`). If omitted, falls back to the `GOOGLE_CSE_ENGINE_ID` environment variable | | `base_url` | `Optional[str]` | provider default | Override the provider's API base URL (mainly for testing or proxies) | | `user_agent` | `str` | `"trpc-agent-python-websearch/1.0"` | HTTP `User-Agent` header, used to help downstream systems distinguish request traffic in logs | | `proxy` | `Optional[str]` | `None` | Optional HTTP proxy URL, passed directly to `httpx` | -| `lang` | `Optional[str]` | `None` | Default Google CSE language (maps to `hl`). Ignored by DDG. Can be overridden by the call-time `lang` parameter | +| `lang` | `Optional[str]` | `None` | Default Google CSE language (maps to `hl`); You.com maps it to the inline `lang:` filter. Ignored by DDG. Can be overridden by the call-time `lang` parameter | | `http_client` | `Optional[httpx.AsyncClient]` | `None` | Optional prebuilt `httpx.AsyncClient` for connection-pool reuse (its lifecycle is managed by the caller) | | `results_num` | `int` | `5` | Default upper bound on returned results, clamped to `[1, 10]`; can be overridden by the call-time `count` parameter | | `snippet_len` | `int` | `300` | Character limit for each `snippet`, clamped to `[1, 1000]` | @@ -2569,6 +2571,8 @@ On top of that, `WebSearchTool` also provides **domain allowlist/blocklist filte | `dedup_urls` | `bool` | `True` | Whether to merge duplicate URLs by normalized key; if `False`, the original hit order is preserved | | `ddg_extra_params` | `Optional[dict]` | `None` | Additional query parameters passed through to DDG | | `google_extra_params` | `Optional[dict]` | `None` | Additional query parameters passed through to Google CSE, such as `{"safe": "active"}`, `{"dateRestrict": "m6"}`, or `{"gl": "us"}` | +| `tavily_extra_params` | `Optional[dict]` | `None` | Additional JSON parameters passed through to Tavily Search, such as `{"search_depth": "advanced"}` or `{"include_answer": True}` | +| `youcom_extra_params` | `Optional[dict]` | `None` | Additional parameters passed through to the You.com `you-search` tool, such as `{"freshness": "week"}` or `{"safesearch": "strict"}`. Per-call arguments `query` / `count` / `exclude_domains` / `lang` are reserved and cannot be pinned here (`lang` is an inline `lang:` operator inside the per-call query) | | `filters_name` | `Optional[List[str]]` | `None` | Names of associated filters, passed through to `BaseTool` | | `filters` | `Optional[List[BaseFilter]]` | `None` | Filter instances injected directly, passed through to `BaseTool` | @@ -2580,16 +2584,16 @@ On top of that, `WebSearchTool` also provides **domain allowlist/blocklist filte | `count` | `integer` | No | Upper bound on the number of results returned in this call, clamped to `1-10`; defaults to the tool-level `results_num` | | `allowed_domains` | `array[string]` | No | Domain allowlist (host only, subdomain-aware, with `www.` stripped automatically); mutually exclusive with `blocked_domains` | | `blocked_domains` | `array[string]` | No | Domain blocklist, using the same matching rules as `allowed_domains`; mutually exclusive with `allowed_domains` | -| `lang` | `string` | No | Only effective for Google CSE (maps to `hl`). Ignored by DDG. Overrides the tool-level `lang` | +| `lang` | `string` | No | Effective for Google CSE (maps to `hl`); You.com maps it to the inline `lang:` filter. Ignored by DDG. Overrides the tool-level `lang` | **`WebSearchResult` return fields**: | Field | Type | Description | |------|------|------| | `query` | `str` | Query term used for this search (echoed back verbatim) | -| `provider` | `"duckduckgo" \| "google"` | Provider actually used | +| `provider` | `"duckduckgo" \| "google" \| "tavily" \| "youcom"` | Provider actually used | | `results` | `List[SearchHit]` | Structured hit list, where each item contains `title`, `url`, and `snippet` | -| `summary` | `str` | Aggregated summary from DDG instant answers / abstracts / definitions. For Google, this field is also populated when spelling correction or API errors occur | +| `summary` | `str` | Aggregated summary from DDG instant answers / abstracts / definitions. For Google, this field is also populated when spelling correction or API errors occur; for Tavily, when the response carries an `answer`; You.com also writes JSON-RPC / tool errors to this field | When call-time arguments are invalid, such as passing both `allowed_domains` and `blocked_domains`, or when `query` is too short, or when an HTTP error occurs, the tool returns a structured error object such as `{"error": "INVALID_ARGS: ..."}` or `{"error": "HTTP_ERROR: ..."}`, making it easier for the LLM to perform graceful fallback handling. @@ -2676,11 +2680,41 @@ def create_google_agent() -> LlmAgent: ) ``` +**Switch to You.com (keyless)**: when you want real public web search results but do not want to sign up for any API keys, switch to `provider="youcom"`. By default it uses You.com's keyless free-profile MCP endpoint — the same zero-configuration story as DuckDuckGo, but returning real web hits. Setting `YDC_API_KEY` (optional) automatically switches requests to the authenticated endpoint: + +```python +import os + +from trpc_agent_sdk.tools import WebSearchTool + + +def create_youcom_agent() -> LlmAgent: + """Create a WebSearchTool Agent based on You.com (keyless by default).""" + web_search = WebSearchTool( + provider="youcom", + # api_key=os.getenv("YDC_API_KEY"), # Optional: switches to the authenticated endpoint + results_num=5, + snippet_len=300, + title_len=100, + timeout=15.0, + youcom_extra_params={ + "freshness": "week", # Optional: keep only results from the last week + }, + ) + return LlmAgent( + name="youcom_research_assistant", + description="Web research assistant powered by You.com Search.", + model=_create_model(), + instruction=INSTRUCTION, + tools=[web_search], + ) +``` + > **Note**: > - `allowed_domains` and `blocked_domains` are filled in by the **LLM in the call arguments** rather than the constructor. The LLM may decide whether to use them based on the user prompt. The two are mutually exclusive, and passing both returns `INVALID_ARGS` > - When an external `http_client` is provided, `WebSearchTool` will **not** call `aclose()` for you. The caller must explicitly close it in the **same event loop** to avoid `Unclosed client` warnings > - Even when an external client is reused, the tool still force-applies the constructor's `timeout` and `user_agent` on every `GET` request, ensuring that the Agent-level constraints always remain in effect -> - Other commonly used Google CSE passthrough parameters include `gl` (geographic bias), `cr` (country restriction), `filter`, and `sort`. For DuckDuckGo, parameters such as `region` and `kl` can be passed through via `ddg_extra_params` +> - Other commonly used Google CSE passthrough parameters include `gl` (geographic bias), `cr` (country restriction), `filter`, and `sort`. For DuckDuckGo, parameters such as `region` and `kl` can be passed through via `ddg_extra_params`; for Tavily, `search_depth` and `include_answer` via `tavily_extra_params`; for You.com, `freshness`, `safesearch`, and `extraction` via `youcom_extra_params` #### Drive the Agent and Print Tool Events diff --git a/docs/mkdocs/zh/tool.md b/docs/mkdocs/zh/tool.md index 5d36a818..fba40e3b 100644 --- a/docs/mkdocs/zh/tool.md +++ b/docs/mkdocs/zh/tool.md @@ -2584,22 +2584,23 @@ if __name__ == "__main__": `WebSearchTool` 是 trpc-agent-python 框架内置的**公网搜索工具**。当 Agent 需要回答"最新动态 / 版本号 / 事件 / 定义 / 事实类"等超出模型知识截止日期的问题时，可以通过该工具调用主流搜索引擎的检索 API，获取带标题、URL 与摘要的结构化结果，并按约定将所有引用以 Markdown 超链接的形式列在 `Sources:` 段落中。 -该工具采用**可插拔 provider** 设计，目前内置三种后端： +该工具采用**可插拔 provider** 设计，目前内置四种后端： - **`duckduckgo`（默认）**：DuckDuckGo Instant Answer API，**无需 API Key**。返回 DDG 精选的 instant answer / abstract / definition 摘要及相关主题，适合百科/定义/事实类查询；注意返回的并非完整的实时网页结果，而是 DDG 的 curated 结果集 - **`google`**：Google Custom Search（CSE）JSON API，需要配置 `api_key` 与 `engine_id`（即 CSE 的 `cx`）；返回真实的公网搜索结果，支持 `siteSearch`、`hl`（语言）、`safe`（SafeSearch）、`dateRestrict`（时效性）等 CSE 原生参数 - **`tavily`**：Tavily Search API，需要配置 `api_key`（或环境变量 `TAVILY_API_KEY`）；返回面向 LLM 的网页结果，并可选返回图片 URL（调用参数 `include_images=true`） +- **`youcom`**：You.com `you-search` MCP 工具（streamable HTTP，无状态 JSON-RPC）。**默认走免密钥的 free profile 端点**（`https://api.you.com/mcp?profile=free`），与 DDG 同为零配置；配置 `YDC_API_KEY`（或构造参数 `api_key`）后会自动切换到认证端点。返回网页/新闻命中，摘要依次取自命中 `description`、You.com 规范的 `snippets` 字段与查询相关的 `contents.highlights`。请求仅携带工具自身的头部，共享 `http_client` 上配置的凭据（网关令牌、Cookie 等）不会随请求转发给 You.com 在此基础上，`WebSearchTool` 还内置了**域名白/黑名单过滤、URL 归一化去重、结果裁剪、引用规范强制注入、HTTP 连接池复用**等能力，帮助你在生产环境中稳定、可控地把联网检索接入 LLM Agent。 ### 功能特性 -- **多 Provider 支持**：`duckduckgo`（keyless，适合定义/百科）、`google`（CSE，真实公网搜索）与 `tavily`（LLM-ready 搜索，可选图片）通过 `provider` 参数切换；基础 `FunctionDeclaration` 保持一致，`tavily` 额外暴露 `include_images` -- **域名白/黑名单**：LLM 可在调用时填入 `allowed_domains` / `blocked_domains`（二者互斥），工具会做**子域感知**匹配（`www.` 前缀剥离，`python.org` 同时匹配 `docs.python.org`）；Google 单域名时走服务端 `siteSearch` 快速路径，多域名自动回退到客户端过滤；Tavily 会映射为 `include_domains` / `exclude_domains`，并仍做客户端二次过滤 +- **多 Provider 支持**：`duckduckgo`（keyless，适合定义/百科）、`google`（CSE，真实公网搜索）、`tavily`（LLM-ready 搜索，可选图片）与 `youcom`（You.com MCP，keyless free profile 或 `YDC_API_KEY` 认证）通过 `provider` 参数切换；基础 `FunctionDeclaration` 保持一致，`tavily` 额外暴露 `include_images` +- **域名白/黑名单**：LLM 可在调用时填入 `allowed_domains` / `blocked_domains`（二者互斥），工具会做**子域感知**匹配（`www.` 前缀剥离，`python.org` 同时匹配 `docs.python.org`）；Google 单域名时走服务端 `siteSearch` 快速路径，多域名自动回退到客户端过滤；Tavily 会映射为 `include_domains` / `exclude_domains`，并仍做客户端二次过滤；You.com 会把 `blocked_domains` 映射为服务端 `exclude_domains`，`allowed_domains` 则在客户端过滤 - **URL 归一化去重**：`dedup_urls=True`（默认）会按 scheme/host/path 归一化键合并重复命中，避免 `Sources:` 段里出现同一来源多次；设置为 `False` 可保留原始召回列表，便于接入下游 re-ranker / 多样化采样 / 离线评估 - **结果裁剪**：`results_num` / `snippet_len` / `title_len` 分别控制返回条数、单条摘要与标题的字符上限，所有参数都会按 `[1, _MAX_*]` 做 clamp，避免误配超上下文窗口；LLM 还可通过 `count` 参数在调用时进一步控制返回条数 - **强制引用规范**：工具在 `process_request` 阶段自动向 LLM 追加指令，**强制**要求：（1）回答末尾必须追加 `Sources:` 段并以 `[Title](URL)` 列出工具返回的 URL；（2）不得编造 URL；（3）涉及"最新/recent"类查询时使用**当前年月**入参，避免幻觉旧年份 -- **Provider 原生参数透传**：`ddg_extra_params` / `google_extra_params` / `tavily_extra_params` 让你把 provider 专属的高级参数（如 Google CSE 的 `safe`、`dateRestrict`，或 Tavily 的 `search_depth`）固化在 agent 层，每次工具调用自动带上，无需在 `FunctionDeclaration` 里额外暴露 +- **Provider 原生参数透传**：`ddg_extra_params` / `google_extra_params` / `tavily_extra_params` / `youcom_extra_params` 让你把 provider 专属的高级参数（如 Google CSE 的 `safe`、`dateRestrict`，Tavily 的 `search_depth`，或 You.com 的 `freshness`、`safesearch`）固化在 agent 层，每次工具调用自动带上，无需在 `FunctionDeclaration` 里额外暴露 - **共享 httpx 连接池**：通过构造参数 `http_client` 传入预建好的 `httpx.AsyncClient`，可在多个 agent / 多次调用之间复用连接池；调用方负责其生命周期（工具不会帮你 `aclose`） - **结构化输出**：统一返回 `WebSearchResult`（Tavily 为扩展的 `TavilyWebSearchResult`），包含 `query` / `provider` / `results: List[{title, url, snippet}]` / `summary`；Tavily 还可额外返回 `images` @@ -2607,13 +2608,13 @@ if __name__ == "__main__": | 参数 | 类型 | 默认值 | 说明 | |------|------|--------|------| -| `provider` | `Literal["duckduckgo", "google", "tavily"]` | `"duckduckgo"` | 搜索后端；`google` 需要 `api_key` + `engine_id`，`tavily` 需要 `api_key`，否则调用时返回未配置提示 | -| `api_key` | `Optional[str]` | `None` | Provider API Key。Google 缺省回退到 `GOOGLE_CSE_API_KEY`；Tavily 缺省回退到 `TAVILY_API_KEY` | +| `provider` | `Literal["duckduckgo", "google", "tavily", "youcom"]` | `"duckduckgo"` | 搜索后端；`google` 需要 `api_key` + `engine_id`，`tavily` 需要 `api_key`，否则调用时返回未配置提示；`youcom` 默认免密钥（free profile），配置 `YDC_API_KEY` 后走认证端点 | +| `api_key` | `Optional[str]` | `None` | Provider API Key。Google 缺省回退到 `GOOGLE_CSE_API_KEY`；Tavily 缺省回退到 `TAVILY_API_KEY`；You.com 缺省回退到 `YDC_API_KEY`（可选） | | `engine_id` | `Optional[str]` | `None` | Google CSE 引擎 ID（即 `cx`），缺省时回退到环境变量 `GOOGLE_CSE_ENGINE_ID` | | `base_url` | `Optional[str]` | provider 默认 | 覆盖 provider 的 API Base URL（主要用于测试 / 代理） | | `user_agent` | `str` | `"trpc-agent-python-websearch/1.0"` | HTTP `User-Agent` 头，便于下游日志区分来源流量 | | `proxy` | `Optional[str]` | `None` | 可选的 HTTP 代理 URL，直接转发给 `httpx` | -| `lang` | `Optional[str]` | `None` | Google CSE 默认语言（对应 `hl` 参数），DDG / Tavily 会忽略；LLM 可通过调用参数 `lang` 覆盖 | +| `lang` | `Optional[str]` | `None` | Google CSE 默认语言（对应 `hl` 参数），You.com 映射为内联 `lang:` 过滤符，DDG / Tavily 会忽略；LLM 可通过调用参数 `lang` 覆盖 | | `http_client` | `Optional[httpx.AsyncClient]` | `None` | 可选的预构建 `httpx.AsyncClient`，用于复用连接池（调用方负责生命周期） | | `results_num` | `int` | `5` | 默认返回条数上限，clamp 到 `[1, 10]`；可被调用参数 `count` 覆盖 | | `snippet_len` | `int` | `300` | 单条 `snippet` 的字符上限，clamp 到 `[1, 1000]` | @@ -2623,6 +2624,7 @@ if __name__ == "__main__": | `ddg_extra_params` | `Optional[dict]` | `None` | 透传给 DDG 的额外查询参数 | | `google_extra_params` | `Optional[dict]` | `None` | 透传给 Google CSE 的额外查询参数（如 `{"safe": "active"}`、`{"dateRestrict": "m6"}`、`{"gl": "us"}` 等） | | `tavily_extra_params` | `Optional[dict]` | `None` | 透传给 Tavily Search 的额外 JSON 参数（如 `{"search_depth": "advanced"}`、`{"include_answer": True}`） | +| `youcom_extra_params` | `Optional[dict]` | `None` | 透传给 You.com `you-search` 的额外参数（如 `{"freshness": "week"}`、`{"safesearch": "strict"}`）。每次调用生成的 `query` / `count` / `exclude_domains` / `lang` 为保留参数，不能在此固化（`lang` 是拼在每次调用 query 里的内联 `lang:` 过滤符） | | `filters_name` | `Optional[List[str]]` | `None` | 关联的 filter 名称，透传给 `BaseTool` | | `filters` | `Optional[List[BaseFilter]]` | `None` | 直接注入的 filter 实例，透传给 `BaseTool` | @@ -2634,7 +2636,7 @@ if __name__ == "__main__": | `count` | `integer` | 否 | 本次调用的返回条数上限，`1-10`（clamp）；默认为工具级 `results_num` | | `allowed_domains` | `array[string]` | 否 | 域名白名单（host only，子域感知，`www.` 自动剥离）；与 `blocked_domains` 互斥 | | `blocked_domains` | `array[string]` | 否 | 域名黑名单，匹配规则同上；与 `allowed_domains` 互斥 | -| `lang` | `string` | 否 | 仅 Google CSE 生效（对应 `hl`），DDG / Tavily 会忽略；覆盖工具级 `lang` | +| `lang` | `string` | 否 | Google CSE 生效（对应 `hl`），You.com 映射为内联 `lang:` 过滤符，DDG / Tavily 会忽略；覆盖工具级 `lang` | | `include_images` | `boolean` | 否 | **仅 `provider="tavily"` 暴露**。为 `true` 时请求 Tavily 返回图片 URL；默认 `false` | **`WebSearchResult` 返回字段**： @@ -2642,9 +2644,9 @@ if __name__ == "__main__": | 字段 | 类型 | 说明 | |------|------|------| | `query` | `str` | 本次检索的查询词（原样回显） | -| `provider` | `"duckduckgo" \| "google" \| "tavily"` | 实际使用的 provider | +| `provider` | `"duckduckgo" \| "google" \| "tavily" \| "youcom"` | 实际使用的 provider | | `results` | `List[SearchHit]` | 结构化命中列表，每项包含 `title` / `url` / `snippet` | -| `summary` | `str` | DDG 的 instant answer / abstract / definition 聚合摘要；Google 在发生拼写纠错或 API 错误时也会写入此字段；Tavily 响应包含 `answer` 时会将其写入此字段。工具默认发送 `include_answer=false`，可通过 `tavily_extra_params` 开启 | +| `summary` | `str` | DDG 的 instant answer / abstract / definition 聚合摘要；Google 在发生拼写纠错或 API 错误时也会写入此字段；Tavily 响应包含 `answer` 时会将其写入此字段。工具默认发送 `include_answer=false`，可通过 `tavily_extra_params` 开启；You.com 在 JSON-RPC / 工具错误时也会把错误信息写入此字段 | | `images` | `List[ImageHit]` | **仅 Tavily**：可选图片列表，每项包含 `url` / `description`（未请求图片时为空列表） | 当调用参数非法（如同时传入 `allowed_domains` 与 `blocked_domains`、`query` 过短）或 HTTP 出错时，工具会返回结构化错误对象（如 `{"error": "INVALID_ARGS: ..."}` / `{"error": "HTTP_ERROR: ..."}`），便于 LLM 做降级处理。 @@ -2763,12 +2765,42 @@ def create_tavily_agent() -> LlmAgent: ) ``` +**切换到 You.com（keyless）**：当需要真实公网搜索结果但不想申请任何 API Key 时，切换到 `provider="youcom"`。默认使用 You.com 的免密钥 free profile MCP 端点，与 DuckDuckGo 一样零配置开跑；如需认证端点，可申请 `YDC_API_KEY`（可选）并通过构造参数或环境变量注入： + +```python +import os + +from trpc_agent_sdk.tools import WebSearchTool + + +def create_youcom_agent() -> LlmAgent: + """创建基于 You.com 的 WebSearchTool Agent（默认免密钥）""" + web_search = WebSearchTool( + provider="youcom", + # api_key=os.getenv("YDC_API_KEY"), # 可选：配置后自动切换到认证端点 + results_num=5, + snippet_len=300, + title_len=100, + timeout=15.0, + youcom_extra_params={ + "freshness": "week", # 可选：仅保留最近一周的结果 + }, + ) + return LlmAgent( + name="youcom_research_assistant", + description="Web research assistant powered by You.com Search.", + model=_create_model(), + instruction=INSTRUCTION, + tools=[web_search], + ) +``` + > **注意**： -> - `provider` 只接受 `duckduckgo` / `google` / `tavily`；传入其它值会在构造 `WebSearchTool` 时抛出 `ValueError` +> - `provider` 只接受 `duckduckgo` / `google` / `tavily` / `youcom`；传入其它值会在构造 `WebSearchTool` 时抛出 `ValueError` > - `allowed_domains` / `blocked_domains` 由 **LLM 在调用参数里**填入（而非构造参数），LLM 可根据用户 prompt 决定是否启用；两者互斥，同时传入会返回 `INVALID_ARGS` > - 当传入外部 `http_client` 时，`WebSearchTool` **不会**帮你调用 `aclose()`，需要调用方在**同一个事件循环**内显式关闭，避免 `Unclosed client` 警告 > - 即使复用外部 client，工具内部仍会在每次请求时强制应用构造器里的 `timeout` 与 `user_agent`，保证 agent 层的约束始终生效 -> - 其他常用的 Google CSE 透传参数包括 `gl`（地理偏向）、`cr`（国家限制）、`filter`、`sort` 等；对 DuckDuckGo 可通过 `ddg_extra_params` 透传 `region`、`kl` 等；对 Tavily 可通过 `tavily_extra_params` 透传 `search_depth`、`include_answer` 等 +> - 其他常用的 Google CSE 透传参数包括 `gl`（地理偏向）、`cr`（国家限制）、`filter`、`sort` 等；对 DuckDuckGo 可通过 `ddg_extra_params` 透传 `region`、`kl` 等；对 Tavily 可通过 `tavily_extra_params` 透传 `search_depth`、`include_answer` 等；对 You.com 可通过 `youcom_extra_params` 透传 `freshness`、`safesearch`、`extraction` 等 > - `include_images` 仅在 `provider="tavily"` 时出现在工具 schema 中；默认关闭，只有回答确实需要图片时再让模型打开 #### 驱动 Agent 并打印工具事件 diff --git a/tests/tools/test_websearch_tool.py b/tests/tools/test_websearch_tool.py index 014310c1..19aece12 100644 --- a/tests/tools/test_websearch_tool.py +++ b/tests/tools/test_websearch_tool.py @@ -18,6 +18,12 @@ missing credentials) - Tavily Search path (results, optional images, domain include/exclude, API error, missing credentials, extra params) +- You.com ``you-search`` MCP path (stateless JSON-RPC over SSE, keyless + free profile, YDC_API_KEY authenticated endpoint, lang inline filter, + server-side excluded / post-hoc allowed domains, JSON-RPC and tool + errors, extra params; malformed highlights, structuredContent payloads, + request-id frame matching, SSE multi-line data + BOM, results-shape + drift) - HTTP errors surfacing as structured tool errors - ``process_request`` registering the declaration and appending the "current month / Sources" system instruction @@ -32,8 +38,11 @@ import pydantic.root_model # noqa: F401 import json +import logging +from contextlib import contextmanager from typing import Any from typing import Dict +from typing import Iterator from unittest.mock import MagicMock import httpx @@ -52,6 +61,8 @@ from trpc_agent_sdk.tools._websearch_tool import _extract_desc_from_pagemap from trpc_agent_sdk.tools._websearch_tool import _extract_title_from_ddg_topic from trpc_agent_sdk.tools._websearch_tool import _is_blocked +from trpc_agent_sdk.tools._websearch_tool import _parse_sse_payloads +from trpc_agent_sdk.tools._websearch_tool import _same_origin from trpc_agent_sdk.tools._websearch_tool import _truncate from trpc_agent_sdk.types import FunctionDeclaration from trpc_agent_sdk.types import Type @@ -87,11 +98,89 @@ def handler(request: httpx.Request) -> httpx.Response: return client +def _make_mock_sse_client( + responses: Dict[str, str], + *, + status: int = 200, + client_headers: Dict[str, str] | None = None, + client_auth: Any = None, + response_headers: Dict[str, str] | None = None, +) -> httpx.AsyncClient: + """Build an ``httpx.AsyncClient`` backed by ``MockTransport`` for SSE bodies. + + The You.com MCP endpoint answers JSON-RPC POSTs with + ``text/event-stream`` instead of a JSON document; ``responses`` maps + request URL path (e.g. ``"/mcp"``) to the raw SSE text to return. + All non-matching paths return 404 so test misses surface loudly. + ``client_headers`` configures client-level headers on the client, the + way a shared gateway client would carry its own credentials, and + ``client_auth`` configures client-level auth (``("user", "pass")``, + ``BearerAuth``, ...) the same way. ``response_headers`` adds extra + headers (e.g. ``Set-Cookie``) to every mocked response. + The returned client captures the last request for assertions via + ``client._captured``. + """ + + captured = {"last_request": None, "all_requests": []} + + def handler(request: httpx.Request) -> httpx.Response: + captured["last_request"] = request + captured["all_requests"].append(request) + body = responses.get(request.url.path) + if body is None: + return httpx.Response(404, json={"error": "no mock for path"}) + headers = {"content-type": "text/event-stream"} + headers.update(response_headers or {}) + return httpx.Response( + status, + text=body, + headers=headers, + ) + + transport = httpx.MockTransport(handler) + client_kwargs: Dict[str, Any] = {"transport": transport, "headers": client_headers or {}} + if client_auth is not None: + client_kwargs["auth"] = client_auth + client = httpx.AsyncClient(**client_kwargs) + # Stash captures on the client so tests can introspect them. + client._captured = captured # type: ignore[attr-defined] + return client + + def _tool_ctx() -> InvocationContext: """Return a minimal fake ``InvocationContext`` — our tool does not touch it.""" return MagicMock(spec=InvocationContext) +class _SdkLogCapture(logging.Handler): + """Capture records from the framework's ``trpc_agent_sdk`` stdlib logger. + + The framework's DefaultLogger attaches to ``logging.getLogger( + "trpc_agent_sdk")`` with ``propagate = False``, so pytest's root-logger + ``caplog`` fixture never sees its records; tests that assert on log + output attach this handler to the SDK logger directly. + """ + + def __init__(self) -> None: + super().__init__(level=logging.DEBUG) + self.messages: list[tuple[int, str]] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.messages.append((record.levelno, record.getMessage())) + + +@contextmanager +def _capture_sdk_logs() -> Iterator[_SdkLogCapture]: + """Yield a handler attached to the SDK logger for the duration.""" + handler = _SdkLogCapture() + sdk_logger = logging.getLogger("trpc_agent_sdk") + sdk_logger.addHandler(handler) + try: + yield handler + finally: + sdk_logger.removeHandler(handler) + + class TestSearchHitSchema: def test_defaults(self): @@ -169,6 +258,20 @@ def test_tavily_reads_api_key_from_env(self, monkeypatch): t = WebSearchTool(provider="tavily") assert t._api_key == "env-tavily-key" + def test_youcom_keyless_by_default(self, monkeypatch): + # No YDC_API_KEY → keyless free profile, no credentials needed. + monkeypatch.delenv("YDC_API_KEY", raising=False) + t = WebSearchTool(provider="youcom") + assert t._api_key == "" + assert t._base_url == "https://api.you.com/mcp?profile=free" + + def test_youcom_reads_api_key_from_env(self, monkeypatch): + monkeypatch.setenv("YDC_API_KEY", "env-ydc-key") + t = WebSearchTool(provider="youcom") + assert t._api_key == "env-ydc-key" + # A configured key switches to the authenticated endpoint. + assert t._base_url == "https://api.you.com/mcp" + def test_results_num_clamped(self): from trpc_agent_sdk.tools._websearch_tool import _MAX_COUNT # Above the cap → clamped to _MAX_COUNT. @@ -202,6 +305,18 @@ def test_tavily_declaration_exposes_include_images(self): assert "include_images" in props assert props["include_images"].type == Type.BOOLEAN + def test_youcom_declaration_has_base_props_only(self): + decl = WebSearchTool(provider="youcom")._get_declaration() + props = decl.parameters.properties + # youcom adds no provider-specific properties (include_images is Tavily-only). + assert set(props.keys()) == { + "query", + "count", + "allowed_domains", + "blocked_domains", + "lang", + } + class TestInputValidation: @@ -1339,6 +1454,1520 @@ async def test_count_limits_hits_and_images(self): await client.aclose() +_YOUCOM_TOOL_TEXT: str = json.dumps({ + "results": { + "web": [ + { + "title": "What's New In Python 3.13", + "url": "https://docs.python.org/3/whatsnew/3.13.html", + "description": "This article explains the new features in Python 3.13.", + "contents": { + "highlights": ["Python 3.13 adds free-threaded support."], + }, + }, + { + # No description — snippet must fall back to the first highlight. + "title": "Python on Wikipedia", + "url": "https://en.wikipedia.org/wiki/Python_(programming_language)", + "description": "", + "contents": { + "highlights": ["Python is a high-level programming language."], + }, + }, + ], + "news": [ + { + "title": "Python 3.13.1 released", + "url": "https://blog.python.org/python-3-13-1-released", + "description": "The first maintenance release of Python 3.13.", + }, + ], + }, +}) + + +def _youcom_sse_body(text: str) -> str: + """Build the SSE body the You.com endpoint returns for ``tools/call``. + + Mirrors the wire format of ``https://api.you.com/mcp``: an optional + ``notifications/message`` notification followed by the JSON-RPC result + carrying the tool text content. + """ + notification = { + "jsonrpc": "2.0", + "method": "notifications/message", + "params": { + "level": "info", + "data": "Search successful", + }, + } + result = { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [{ + "type": "text", + "text": text, + }], + }, + } + return ("event: message\n" + f"data: {json.dumps(notification)}\n\n" + "event: message\n" + f"data: {json.dumps(result)}\n\n") + + +class TestYoucomProvider: + """You.com ``youcom`` provider: stateless JSON-RPC over streamable HTTP.""" + + @pytest.mark.asyncio + async def test_keyless_search_maps_web_and_news_hits(self, monkeypatch): + monkeypatch.delenv("YDC_API_KEY", raising=False) + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool(provider="youcom", http_client=client) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "Python 3.13"}) + assert res["provider"] == "youcom" + assert res["query"] == "Python 3.13" + assert res["summary"] == "" + urls = [h["url"] for h in res["results"]] + # Web hits first, then news hits merged in. + assert urls == [ + "https://docs.python.org/3/whatsnew/3.13.html", + "https://en.wikipedia.org/wiki/Python_(programming_language)", + "https://blog.python.org/python-3-13-1-released", + ] + # Descriptions map to snippets... + assert "new features in Python 3.13" in res["results"][0]["snippet"] + # ...with first-highlight fallback when description is empty. + assert res["results"][1]["snippet"] == "Python is a high-level programming language." + req = client._captured["last_request"] + # Keyless: no Authorization header on the free-profile request. + assert req.headers.get("authorization") is None + # The endpoint rejects JSON-only Accept headers (406); SSE must be accepted. + assert "text/event-stream" in req.headers.get("accept", "") + payload = json.loads(req.content) + assert payload["method"] == "tools/call" + assert payload["params"]["name"] == "you-search" + arguments = payload["params"]["arguments"] + assert arguments["query"] == "Python 3.13" + assert arguments["count"] == 5 # default results_num + await client.aclose() + + @pytest.mark.asyncio + async def test_api_key_sends_bearer_header(self): + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + api_key="ydc-test", + http_client=client, + base_url="https://api.you.com/mcp", + ) + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + req = client._captured["last_request"] + assert req.headers.get("authorization") == "Bearer ydc-test" + await client.aclose() + + @pytest.mark.asyncio + async def test_snippets_used_when_description_missing(self, monkeypatch): + # ``snippets`` is You.com's canonical summary field (a list of + # strings returned by default); news hits often carry it while + # ``description`` is empty or missing. Modeled on the official + # you-search response shape. + monkeypatch.delenv("YDC_API_KEY", raising=False) + text = json.dumps({ + "results": { + "web": [ + { + "title": "What's new", + "url": "https://docs.python.org/3/whatsnew/3.13.html", + # A non-empty description still wins over snippets. + "description": "Release notes for Python 3.13.", + "snippets": ["shadows the canonical field"], + }, + { + "title": "PEP 720", + "url": "https://peps.python.org/pep-0720/", + "description": "", + # First non-empty string entry wins; blank and + # non-string entries are skipped. + "snippets": ["", 42, "PEP title change summary", "second"], + }, + { + "title": "Bare snippet", + "url": "https://peps.python.org/pep-0719/", + "description": "", + # A bare string instead of a list: no crash and no + # bogus snippet, same guard as for highlights. + "snippets": "not a list", + }, + ], + "news": [ + { + "title": "Python 3.13.1 released", + "url": "https://blog.python.org/python-3-13-1-released", + # No description key at all — the canonical path for + # news hits on the free profile. + "snippets": ["The first maintenance release of Python 3.13."], + }, + ], + }, + }) + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(text)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert [h["snippet"] for h in res["results"]] == [ + "Release notes for Python 3.13.", + "PEP title change summary", + "", + "The first maintenance release of Python 3.13.", + ] + await client.aclose() + + @pytest.mark.asyncio + async def test_shared_client_credentials_not_forwarded(self, monkeypatch): + # A shared ``http_client`` carrying client-level credentials (e.g. a + # gateway bearer token or cookie jar) must not leak to the You.com + # endpoint: httpx merges client-level headers into ``client.post`` + # requests, so the SSE request is built and sent as-is instead. + monkeypatch.delenv("YDC_API_KEY", raising=False) + client_headers = { + "Authorization": "Bearer gateway-secret", + "X-Api-Key": "gateway-key", + "Cookie": "session=gw", + } + client = _make_mock_sse_client( + {"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}, + client_headers=client_headers, + ) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + req = client._captured["last_request"] + assert res["provider"] == "youcom" + assert req.headers.get("authorization") is None + assert req.headers.get("x-api-key") is None + assert req.headers.get("cookie") is None + # The tool's own headers still reach the wire. + assert "text/event-stream" in req.headers.get("accept", "") + assert req.headers.get("user-agent") + await client.aclose() + + @pytest.mark.asyncio + async def test_shared_client_credentials_stripped_even_with_bearer(self): + # When the tool itself sends a bearer key, only that key goes out; + # client-level credentials are still not forwarded. + client = _make_mock_sse_client( + {"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}, + client_headers={"X-Api-Key": "gateway-key"}, + ) + t = WebSearchTool( + provider="youcom", + api_key="ydc-test", + http_client=client, + base_url="https://api.you.com/mcp", + ) + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + req = client._captured["last_request"] + assert req.headers.get("authorization") == "Bearer ydc-test" + assert req.headers.get("x-api-key") is None + await client.aclose() + + @pytest.mark.asyncio + async def test_shared_client_basic_auth_not_forwarded(self, monkeypatch): + # ``client.send`` applies client-level ``auth`` (Basic/Bearer/custom + # flows) even to a bare request, so the You.com call must suppress + # it per call: a shared gateway client configured with + # ``auth=("user", "pass")`` must not leak those credentials to the + # third-party endpoint. + monkeypatch.delenv("YDC_API_KEY", raising=False) + client = _make_mock_sse_client( + {"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}, + client_auth=("user", "pass"), + ) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + req = client._captured["last_request"] + assert res["provider"] == "youcom" + assert req.headers.get("authorization") is None + assert len(res["results"]) == 3 + await client.aclose() + + @pytest.mark.asyncio + async def test_shared_client_basic_auth_superseded_by_tool_key(self): + # With the tool's own key configured, only that key goes out; the + # shared client's Basic auth is still suppressed. + client = _make_mock_sse_client( + {"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}, + client_auth=("user", "pass"), + ) + t = WebSearchTool( + provider="youcom", + api_key="ydc-test", + http_client=client, + base_url="https://api.you.com/mcp", + ) + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + req = client._captured["last_request"] + assert req.headers.get("authorization") == "Bearer ydc-test" + assert "Basic" not in (req.headers.get("authorization") or "") + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_set_cookie_not_persisted_to_shared_jar(self): + # A ``Set-Cookie`` on the You.com response must not linger in the + # shared client's cookie jar, where it would be replayed to other + # providers (e.g. Tavily) through the same ``http_client``. + client = _make_mock_sse_client( + {"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}, + response_headers={"set-cookie": "you_session=abc; Path=/"}, + ) + client.cookies.set("gw_session", "keepme", domain="gateway.internal") + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + # The You.com cookie was dropped again... + assert client.cookies.get("you_session") is None + # ...while pre-existing cookies on the shared client survive. + assert client.cookies.get("gw_session") == "keepme" + # A follow-up request through the same client carries no You.com cookie. + await client.post("https://api.tavily.com/search", json={"query": "python"}) + followup = client._captured["all_requests"][-1] + assert "you_session" not in (followup.headers.get("cookie") or "") + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_redirect_followed(self): + # httpx does not follow redirects by default; the You.com call must + # opt in so a 301/302 (e.g. a base-url change) does not degrade to + # parsing the empty 3xx body as a silent 0-hit search. + captured: Dict[str, Any] = {"all_requests": []} + + def handler(request: httpx.Request) -> httpx.Response: + captured["all_requests"].append(request) + if request.url.path == "/mcp": + return httpx.Response( + 302, + headers={"location": "https://api.you.com/mcp-moved"}, + ) + return httpx.Response( + 200, + text=_youcom_sse_body(_YOUCOM_TOOL_TEXT), + headers={"content-type": "text/event-stream"}, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert len(captured["all_requests"]) == 2 + assert captured["all_requests"][-1].url.path == "/mcp-moved" + assert len(res["results"]) == 3 + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_cross_origin_redirect_dropped(self): + # httpx keeps Authorization on same-origin (and same-host upgrade) + # redirects and re-attaches cookie-jar cookies to every redirect + # request. A 3xx naming another origin must therefore be dropped, + # not followed, or the YDC_API_KEY bearer header would reach + # whatever host the redirect names. + captured: Dict[str, Any] = {"all_requests": []} + + def handler(request: httpx.Request) -> httpx.Response: + captured["all_requests"].append(request) + return httpx.Response( + 302, + headers={"location": "https://collector.example.com/mcp"}, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + api_key="ydc-test", + http_client=client, + base_url="https://api.you.com/mcp", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + # Only the original request went out; the redirect was not followed. + assert len(captured["all_requests"]) == 1 + assert captured["all_requests"][0].url.host == "api.you.com" + assert res["results"] == [] + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_cross_origin_redirect_rolls_back_cookies(self): + # The cross-origin bail-out returns before the tool parses a body, + # but httpx has already written the dropped 3xx's Set-Cookie into + # the shared client's jar (extract_cookies runs for unfollowed + # responses too). The cookie rollback must run on the early-return + # path as well, or the You.com cookie would be replayed to the + # next provider sharing the client. + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 302, + headers={ + "location": "https://collector.example.com/mcp", + "set-cookie": "you_session=abc; Path=/", + }, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + client.cookies.set("gw_session", "keepme", domain="gateway.internal") + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + # The You.com cookie written by the dropped 3xx was rolled back... + assert client.cookies.get("you_session") is None + # ...while pre-existing cookies on the shared client survive. + assert client.cookies.get("gw_session") == "keepme" + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_overwritten_cookie_value_left_as_is(self): + # Set-Cookie for a name that already existed in the jar overwrites + # its value mid-call; the rollback no longer restores the pre-call + # value — doing so would silently clobber a concurrent call that + # updated the same cookie after our snapshot. The You.com response + # value is left in place; only cookies the call *added* are deleted. + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + text=_youcom_sse_body(_YOUCOM_TOOL_TEXT), + headers={ + "content-type": "text/event-stream", + "set-cookie": "you_session=new; Path=/", + }, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + client.cookies.set("you_session", "old", domain="api.you.com", path="/") + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] + # The response's Set-Cookie value is left in place, not restored. + assert client.cookies.get("you_session") == "new" + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_rollback_leaves_unrelated_domain_cookies(self): + # The rollback deletes only cookies that could plausibly have come + # from a You.com response (domain-matching the endpoint host every + # hop was validated to stay on). A cookie a concurrent provider + # call writes into the same shared jar — here a Tavily session, + # added while the You.com request is in flight — must survive the + # You.com rollback. + def handler(request: httpx.Request) -> httpx.Response: + # A concurrent Tavily call sharing this client finishes while + # the You.com request is in flight (after the snapshot). + client.cookies.set("tavily_session", "keepme", domain="api.tavily.com") + return httpx.Response( + 302, + headers={ + "location": "https://collector.example.com/mcp", + "set-cookie": "you_session=abc; Path=/", + }, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + # The You.com cookie (endpoint host) was rolled back... + assert client.cookies.get("you_session") is None + # ...while the concurrent call's cookie on another domain survives. + assert client.cookies.get("tavily_session") == "keepme" + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_plaintext_downgrade_after_upgrade_dropped(self): + # Every hop is validated against the URL it redirects *from*: after + # a same-host http->https upgrade, a further hop back down to + # plaintext http must be dropped (comparing against the *original* + # URL would pass it), or the Authorization header would travel + # over plaintext http. + captured: Dict[str, Any] = {"all_requests": []} + + def handler(request: httpx.Request) -> httpx.Response: + captured["all_requests"].append(request) + if request.url.scheme == "http" and request.url.path == "/mcp": + return httpx.Response(302, headers={"location": "https://api.you.com/mcp2"}) + if request.url.scheme == "https" and request.url.path == "/mcp2": + return httpx.Response(302, headers={"location": "http://api.you.com/mcp3"}) + return httpx.Response( + 200, + text=_youcom_sse_body(_YOUCOM_TOOL_TEXT), + headers={"content-type": "text/event-stream"}, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + api_key="ydc-test", + http_client=client, + base_url="http://api.you.com/mcp", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + # The http->https upgrade was followed; the https->http + # downgrade was not. + assert [r.url.scheme for r in captured["all_requests"]] == ["http", "https"] + assert all(r.url.host == "api.you.com" for r in captured["all_requests"]) + assert res["results"] == [] + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_same_origin_redirect_followed_with_key(self): + # Redirects are followed manually, one request at a time; a + # same-origin move (e.g. a path change on the same host) is still + # followed, and the tool's own Authorization header stays on the + # follow-up request because the target is the same origin. + captured: Dict[str, Any] = {"all_requests": []} + + def handler(request: httpx.Request) -> httpx.Response: + captured["all_requests"].append(request) + if request.url.path == "/mcp": + return httpx.Response(302, headers={"location": "/mcp-moved"}) + return httpx.Response( + 200, + text=_youcom_sse_body(_YOUCOM_TOOL_TEXT), + headers={"content-type": "text/event-stream"}, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + api_key="ydc-test", + http_client=client, + base_url="https://api.you.com/mcp", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert len(captured["all_requests"]) == 2 + assert captured["all_requests"][-1].url.host == "api.you.com" + assert captured["all_requests"][-1].url.path == "/mcp-moved" + assert captured["all_requests"][-1].headers.get("authorization") == "Bearer ydc-test" + assert len(res["results"]) == 3 + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_relative_redirect_keeps_profile_query(self): + # ``URL.join`` follows urljoin semantics: a relative Location drops + # the original query string. A keyless "?profile=free" call must + # re-attach the profile param on the followed hop, or the redirect + # would silently land on the authenticated endpoint (which rejects + # it, since no key is configured). + captured: Dict[str, Any] = {"all_requests": []} + + def handler(request: httpx.Request) -> httpx.Response: + captured["all_requests"].append(request) + if request.url.path == "/mcp": + return httpx.Response(302, headers={"location": "/mcp-moved"}) + return httpx.Response( + 200, + text=_youcom_sse_body(_YOUCOM_TOOL_TEXT), + headers={"content-type": "text/event-stream"}, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert len(captured["all_requests"]) == 2 + followup = captured["all_requests"][-1] + assert followup.url.path == "/mcp-moved" + assert followup.url.host == "api.you.com" + assert followup.url.params.get("profile") == "free" + assert len(res["results"]) == 3 + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_lang_extra_param_is_reserved(self): + # ``lang`` is an inline ``lang:`` operator inside the per-call query, + # so a pinned ``youcom_extra_params={"lang": ...}`` must be rejected + # with a warning instead of landing in the tools/call arguments. + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + youcom_extra_params={"lang": "pinned", "freshness": "week"}, + ) + with _capture_sdk_logs() as cap: + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + req = client._captured["last_request"] + payload = json.loads(req.content) + arguments = payload["params"]["arguments"] + assert "lang" not in arguments + assert arguments["freshness"] == "week" + warnings = [m for level, m in cap.messages if level >= logging.WARNING] + assert any("youcom_extra_params key 'lang'" in m for m in warnings) + await client.aclose() + + @pytest.mark.asyncio + async def test_frame_ids_compared_as_strings(self): + # Some servers echo the JSON-RPC id back as a string ("1"); a strict + # int comparison would drop every reply frame and report an empty + # search. + result = { + "jsonrpc": "2.0", + "id": "1", + "result": { + "content": [{ + "type": "text", + "text": _YOUCOM_TOOL_TEXT + }] + }, + } + body = f"event: message\ndata: {json.dumps(result)}\n\n" + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["summary"] == "" + assert len(res["results"]) == 3 + await client.aclose() + + @pytest.mark.asyncio + async def test_superseded_error_frame_is_logged(self): + # A stray error frame overridden by a later result must stay + # observable in the logs, not be dropped silently. + error = { + "jsonrpc": "2.0", + "id": 1, + "error": { + "code": -32000, + "message": "transient", + }, + } + result = { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [{ + "type": "text", + "text": _YOUCOM_TOOL_TEXT + }] + }, + } + body = (f"event: message\ndata: {json.dumps(error)}\n\n" + f"event: message\ndata: {json.dumps(result)}\n\n") + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + with _capture_sdk_logs() as cap: + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["summary"] == "" + assert len(res["results"]) == 3 + warnings = [m for level, m in cap.messages if level >= logging.WARNING] + assert any("superseded" in m and "transient" in m for m in warnings) + await client.aclose() + + @pytest.mark.asyncio + async def test_lang_maps_to_inline_filter(self): + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + await t._run_async_impl( + tool_context=_tool_ctx(), + args={ + "query": "python release", + "lang": "zh-CN" + }, + ) + payload = json.loads(client._captured["last_request"].content) + assert payload["params"]["arguments"]["query"] == "python release lang:zh-CN" + await client.aclose() + + @pytest.mark.asyncio + async def test_blocked_domains_sent_server_side_and_filtered(self): + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl( + tool_context=_tool_ctx(), + args={ + "query": "python", + "blocked_domains": ["wikipedia.org"] + }, + ) + payload = json.loads(client._captured["last_request"].content) + # Blocked domains ride along as server-side exclude_domains. + assert payload["params"]["arguments"]["exclude_domains"] == ["wikipedia.org"] + urls = [h["url"] for h in res["results"]] + assert all("wikipedia.org" not in u for u in urls) + await client.aclose() + + @pytest.mark.asyncio + async def test_allowed_domains_filter_post_hoc(self): + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl( + tool_context=_tool_ctx(), + args={ + "query": "python", + "allowed_domains": ["python.org"] + }, + ) + payload = json.loads(client._captured["last_request"].content) + # Allowlists are enforced client-side; nothing rides along server-side. + assert "exclude_domains" not in payload["params"]["arguments"] + urls = [h["url"] for h in res["results"]] + assert urls == [ + "https://docs.python.org/3/whatsnew/3.13.html", + "https://blog.python.org/python-3-13-1-released", + ] + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_extra_params_are_merged(self): + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + youcom_extra_params={"freshness": "week"}, + ) + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + payload = json.loads(client._captured["last_request"].content) + assert payload["params"]["arguments"]["freshness"] == "week" + await client.aclose() + + @pytest.mark.asyncio + async def test_count_limits_hits(self): + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + results_num=1, + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert len(res["results"]) == 1 + payload = json.loads(client._captured["last_request"].content) + assert payload["params"]["arguments"]["count"] == 1 + await client.aclose() + + @pytest.mark.asyncio + async def test_dedup_urls_and_urlless_hits_skipped(self): + text = json.dumps({ + "results": { + "web": [ + { + "title": "A", + "url": "https://example.com/page", + "description": "d1" + }, + # Trailing slash — same normalised key as the first hit. + { + "title": "B", + "url": "https://example.com/page/", + "description": "d2" + }, + # No URL — must be skipped. + { + "title": "C", + "url": "", + "description": "d3" + }, + ], + }, + }) + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(text)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + urls = [h["url"] for h in res["results"]] + assert urls == ["https://example.com/page"] + await client.aclose() + + @pytest.mark.asyncio + async def test_jsonrpc_error_surfaced_as_summary(self): + error = { + "jsonrpc": "2.0", + "id": 1, + "error": { + "code": -32000, + "message": "rate limited", + }, + } + body = f"event: message\ndata: {json.dumps(error)}\n\n" + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + assert "rate limited" in res["summary"] + await client.aclose() + + @pytest.mark.asyncio + async def test_tool_iserror_surfaced_as_summary(self): + result = { + "jsonrpc": "2.0", + "id": 1, + "result": { + "isError": True, + "content": [{ + "type": "text", + "text": "upstream provider unavailable", + }], + }, + } + body = f"event: message\ndata: {json.dumps(result)}\n\n" + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + assert "upstream provider unavailable" in res["summary"] + await client.aclose() + + @pytest.mark.asyncio + async def test_no_result_payload_returns_summary(self): + # SSE body carrying only a notification — no JSON-RPC result line. + notification = {"jsonrpc": "2.0", "method": "notifications/message", "params": {}} + body = f"data: {json.dumps(notification)}\n\n" + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + assert res["summary"] == "You.com search returned no result payload." + await client.aclose() + + @pytest.mark.asyncio + async def test_non_json_tool_text_returns_summary(self): + client = _make_mock_sse_client({"/mcp": _youcom_sse_body("")}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + assert res["summary"] == "You.com search returned an unexpected payload." + await client.aclose() + + @pytest.mark.asyncio + async def test_highlights_malformed_types_do_not_break_search(self): + text = json.dumps({ + "results": { + "web": [ + { + "title": "A", + "url": "https://example.com/a", + "description": "", + # A bare string instead of a list: no crash and no + # single-character snippet. + "contents": { + "highlights": "snippet text" + }, + }, + { + "title": "B", + "url": "https://example.com/b", + "description": "", + # A dict instead of a list: must not escape to the + # broad except in _run_async_impl as SEARCH_ERROR. + "contents": { + "highlights": { + "0": "snippet text" + } + }, + }, + { + "title": "C", + "url": "https://example.com/c", + "description": "", + # First element not a string: not a usable fallback. + "contents": { + "highlights": [42] + }, + }, + ], + }, + }) + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(text)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert "error" not in res + assert [h["url"] for h in res["results"]] == [ + "https://example.com/a", + "https://example.com/b", + "https://example.com/c", + ] + assert [h["snippet"] for h in res["results"]] == ["", "", ""] + await client.aclose() + + @pytest.mark.asyncio + async def test_tool_iserror_without_text_reports_error(self): + # isError with no text content used to fall through to payload + # parsing and report a misleading "unexpected payload" success. + result = { + "jsonrpc": "2.0", + "id": 1, + "result": { + "isError": True, + "content": [], + "structuredContent": { + "code": -32000, + "message": "quota exceeded" + }, + }, + } + body = f"event: message\ndata: {json.dumps(result)}\n\n" + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + assert res["summary"].startswith("You.com search error") + assert "quota exceeded" in res["summary"] + await client.aclose() + + @pytest.mark.asyncio + async def test_structured_content_used_when_text_is_not_json(self): + result = { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [{ + "type": "text", + "text": "Search finished; see the structured results.", + }], + "structuredContent": { + "results": { + "web": [{ + "title": "Python", + "url": "https://www.python.org", + "description": "Welcome to Python.org", + }], + }, + }, + }, + } + body = f"event: message\ndata: {json.dumps(result)}\n\n" + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert [h["url"] for h in res["results"]] == ["https://www.python.org"] + await client.aclose() + + @pytest.mark.asyncio + async def test_stray_error_superseded_by_later_result(self): + # An error frame followed by a valid result frame used to discard + # the result because the error had unconditional priority. + error = {"jsonrpc": "2.0", "id": 1, "error": {"code": -32000, "message": "transient"}} + result = { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [{ + "type": "text", + "text": _YOUCOM_TOOL_TEXT + }] + }, + } + body = (f"event: message\ndata: {json.dumps(error)}\n\n" + f"event: message\ndata: {json.dumps(result)}\n\n") + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["summary"] == "" + assert len(res["results"]) == 3 + await client.aclose() + + @pytest.mark.asyncio + async def test_same_frame_result_and_error_treated_as_error(self): + frame = { + "jsonrpc": "2.0", + "id": 1, + "result": None, + "error": { + "code": -32000, + "message": "rate limited" + }, + } + body = f"event: message\ndata: {json.dumps(frame)}\n\n" + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + assert "rate limited" in res["summary"] + await client.aclose() + + @pytest.mark.asyncio + async def test_frames_for_other_request_ids_ignored(self): + other = { + "jsonrpc": "2.0", + "id": 2, + "error": { + "code": -32000, + "message": "someone else's failure" + }, + } + result = { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [{ + "type": "text", + "text": _YOUCOM_TOOL_TEXT + }] + }, + } + body = (f"event: message\ndata: {json.dumps(other)}\n\n" + f"event: message\ndata: {json.dumps(result)}\n\n") + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["summary"] == "" + assert len(res["results"]) == 3 + await client.aclose() + + @pytest.mark.asyncio + async def test_lang_operator_prefix_not_duplicated(self): + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + await t._run_async_impl( + tool_context=_tool_ctx(), + args={ + "query": "python release", + "lang": "lang:ja" + }, + ) + payload = json.loads(client._captured["last_request"].content) + assert payload["params"]["arguments"]["query"] == "python release lang:ja" + await client.aclose() + + @pytest.mark.asyncio + async def test_youcom_extra_params_cannot_override_call_arguments(self): + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + youcom_extra_params={ + "freshness": "week", + "query": "pinned query", + "count": 99 + }, + ) + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + arguments = json.loads(client._captured["last_request"].content)["params"]["arguments"] + assert arguments["query"] == "python" + assert arguments["count"] == 5 # default results_num + assert arguments["freshness"] == "week" + await client.aclose() + + @pytest.mark.asyncio + async def test_flat_list_results_shape_parses_hits(self): + # Shape drift: a non-empty list ``results`` (a flat list of hits + # instead of ``{"web": [...], "news": [...]}`` sections) maps to + # hits instead of dropping every entry. + text = json.dumps({ + "results": [ + {"url": "https://example.com", "title": "flat", "description": "d"}, + ], + }) + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(text)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert [h["url"] for h in res["results"]] == ["https://example.com"] + assert res["results"][0]["title"] == "flat" + await client.aclose() + + @pytest.mark.asyncio + async def test_flat_list_results_with_sibling_sections_merged(self): + # Shape drift can put a real ``web``/``news`` section *next to* a + # flat-list ``results``: the flat fallback must merge with (not + # clobber) the genuine sections, so real hits are never dropped. + text = json.dumps({ + "results": [ + {"url": "https://flat.example.com", "title": "flat", "description": "d"}, + ], + "web": [ + {"url": "https://real.example.com", "title": "real", "description": "d"}, + ], + "news": [ + {"url": "https://news.example.com", "title": "news", "description": "d"}, + ], + }) + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(text)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + urls = [h["url"] for h in res["results"]] + # The real sections survive; the flat list still parses. + assert "https://real.example.com" in urls + assert "https://news.example.com" in urls + assert "https://flat.example.com" in urls + await client.aclose() + + @pytest.mark.asyncio + async def test_non_string_fields_not_coerced(self): + # Structured url objects (``{"href": ...}``) are unwrapped rather + # than str()-repr'd; non-string title/description fields yield + # empty values (then the snippets fallback) instead of leaking + # Python repr text into titles and snippets, and hits without any + # usable url are skipped rather than turning into a repr that the + # blocklist check silently drops. + text = json.dumps({ + "results": { + "web": [ + { + "url": {"kind": "web", "href": "https://href.example.com"}, + "title": "href hit", + "description": {"nested": "dict"}, + "snippets": ["structured description"], + }, + { + "url": ["https://list.example.com"], + "title": ["list", "title"], + "description": "list url hit", + }, + ], + }, + }) + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(text)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + # The list-shaped url was skipped, not repr'd into a fake hit. + assert [h["url"] for h in res["results"]] == ["https://href.example.com"] + hit = res["results"][0] + assert hit["title"] == "href hit" + # The dict description fell through to snippets instead of a repr. + assert hit["snippet"] == "structured description" + await client.aclose() + + @pytest.mark.asyncio + async def test_non_dict_results_shape_surfaced_as_summary(self): + # A non-empty scalar ``results`` (string, number, ...) is an upstream + # shape change, not an empty search; surface the observed type + # instead of reporting 0 hits. + text = json.dumps({"results": "unavailable"}) + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(text)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + assert res["summary"] == ( + "You.com search returned an unexpected results shape (str); no hits parsed." + ) + await client.aclose() + + @pytest.mark.asyncio + async def test_non_list_sections_skipped_without_breaking_search(self): + text = json.dumps({ + "results": { + "web": { + "url": "https://example.com", + "title": "not a list" + }, + "news": [{ + "title": "Python 3.13.1 released", + "url": "https://blog.python.org/python-3-13-1-released", + "description": "The first maintenance release of Python 3.13.", + }], + }, + }) + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(text)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert "error" not in res + assert [h["url"] for h in res["results"]] == ["https://blog.python.org/python-3-13-1-released"] + await client.aclose() + + def test_parse_sse_payloads_skips_garbage_lines(self): + text = ('data: not-json\n\n' + 'event: message\n' + 'data: {"result": {"x": 1}}\n\n') + payloads = _parse_sse_payloads(text) + assert payloads == [{"result": {"x": 1}}] + + def test_parse_sse_payloads_joins_multi_line_data_frames(self): + # Per the SSE spec one event's data may span several data: lines; + # each fragment alone fails to parse, the join must not. + text = 'data: {"result": {"x":\ndata: 1}}\n\n' + payloads = _parse_sse_payloads(text) + assert payloads == [{"result": {"x": 1}}] + + def test_parse_sse_payloads_joins_chunked_frames_without_separator(self): + # A server may chunk one JSON frame across data: lines without + # regard for JSON token boundaries. The SSE "\n" join puts a raw + # control character inside the JSON string (always rejected by + # json.loads); the separator-less join must recover the frame + # instead of silently dropping it as a 0-hit search. + text = 'data: {"result": "ab\ndata: c"}\n\n' + payloads = _parse_sse_payloads(text) + assert payloads == [{"result": "abc"}] + + def test_parse_sse_payloads_bom_only_first_line(self): + # A BOM-only first line is a framing artifact, not data; the rest + # of the body must still parse. + text = '\ndata: {"result": {"x": 1}}\n\n' + payloads = _parse_sse_payloads(text) + assert payloads == [{"result": {"x": 1}}] + + def test_parse_sse_payloads_bom_only_body_is_empty(self): + assert _parse_sse_payloads("") == [] + + def test_parse_sse_payloads_falls_back_to_single_lines_without_separators(self): + # Servers that omit blank-line event separators: each line is + # still tried on its own so complete frames are not dropped. + text = 'data: {"a": 1}\ndata: {"b": 2}\n' + payloads = _parse_sse_payloads(text) + assert payloads == [{"a": 1}, {"b": 2}] + + def test_parse_sse_payloads_string_split_without_separator_keeps_string_intact(self): + # A frame split inside a JSON string with no blank separator: the + # separator-less join is the faithful reassembly and must be + # adopted (the newline join would put a raw control character in + # the string); the payload must be restored with the exact string + # the server serialized, with no newline injected. + text = 'data: {"snippet": "ab\ndata: c"}\n' + payloads = _parse_sse_payloads(text) + assert payloads == [{"snippet": "abc"}] + + def test_parse_sse_payloads_mixed_frames_without_separators_pin_tradeoff(self): + # Complete one-line frames and one pretty-printed multi-line frame + # concatenated without blank separators: the buffer-level joins + # cannot parse (several frames concatenated), so the per-line + # fallback keeps the complete frames and drops the multi-line one. + # Without separators the input is ambiguous; this pins that + # declared trade-off as stable behavior. + text = 'data: {"a": "x"}\ndata: {"b":\ndata: 2}\ndata: {"c": 3}\n' + payloads = _parse_sse_payloads(text) + assert payloads == [{"a": "x"}, {"c": 3}] + + def test_parse_sse_payloads_strips_leading_bom(self): + text = 'data: {"result": {"x": 1}}\n\n' + payloads = _parse_sse_payloads(text) + assert payloads == [{"result": {"x": 1}}] + + def test_parse_sse_payloads_keepalive_and_non_object_lines_are_quiet(self): + # Empty ``data:`` keep-alive lines and valid non-object JSON must + # not produce warning noise; only truly unparseable lines warn. + text = ('data: \n\n' + 'data: 123\n\n' + 'data: [1, 2]\n\n' + 'data: not-json\n\n' + 'data: {"result": {"x": 1}}\n\n') + with _capture_sdk_logs() as cap: + payloads = _parse_sse_payloads(text) + assert payloads == [{"result": {"x": 1}}] + warnings = [m for level, m in cap.messages if level >= logging.WARNING] + assert len(warnings) == 1 + assert "not-json" in warnings[0] + + @pytest.mark.asyncio + async def test_result_frame_before_error_frame_result_wins(self): + # A result frame followed by a stray error frame must not discard the + # valid result: result takes precedence regardless of frame order. + result = { + "jsonrpc": "2.0", + "id": 1, + "result": { + "content": [{ + "type": "text", + "text": _YOUCOM_TOOL_TEXT, + }], + }, + } + error = {"jsonrpc": "2.0", "id": 1, "error": {"code": -32000, "message": "late transient"}} + body = (f"event: message\ndata: {json.dumps(result)}\n\n" + f"event: message\ndata: {json.dumps(error)}\n\n") + client = _make_mock_sse_client({"/mcp": body}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["summary"] == "" + assert len(res["results"]) == 3 + await client.aclose() + + @pytest.mark.asyncio + async def test_empty_location_redirect_breaks(self): + # A redirect with an empty or whitespace-only Location must not loop + # back to the same URL; it breaks out after a single request. + captured: Dict[str, Any] = {"count": 0} + + def handler(request: httpx.Request) -> httpx.Response: + captured["count"] += 1 + return httpx.Response(302, headers={"location": ""}) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert captured["count"] == 1 + assert "error" in res + await client.aclose() + + @pytest.mark.asyncio + async def test_redirect_budget_limits_to_three(self): + # The redirect budget is _SSE_MAX_REDIRECTS (3); each redirect goes + # to a *different* path so the same-URL self-redirect guard does + # not fire — only the budget stops the loop. + captured: Dict[str, Any] = {"count": 0} + + def handler(request: httpx.Request) -> httpx.Response: + captured["count"] += 1 + return httpx.Response( + 307, + headers={"location": f"https://api.you.com/mcp-{captured['count']}"}, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + # 1 initial POST + 3 redirect follows = 4 total requests; the 4th + # redirect response breaks the loop without a 5th request. + assert captured["count"] == 4 + await client.aclose() + + @pytest.mark.asyncio + async def test_lang_operator_already_in_query_not_duplicated(self): + # When the query already contains a ``lang:`` operator, the lang + # parameter must not append a second one (``python lang:ja lang:ja``). + client = _make_mock_sse_client({"/mcp": _youcom_sse_body(_YOUCOM_TOOL_TEXT)}) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + await t._run_async_impl( + tool_context=_tool_ctx(), + args={"query": "python lang:ja", "lang": "ja"}, + ) + payload = json.loads(client._captured["last_request"].content) + assert payload["params"]["arguments"]["query"] == "python lang:ja" + await client.aclose() + + @pytest.mark.asyncio + async def test_non_sse_json_error_surfaced(self): + # A non-SSE response (application/json) carrying a JSON-RPC error + # frame must be parsed as JSON so the error message reaches the + # summary — not silently swallowed as an empty SSE body. + error = { + "jsonrpc": "2.0", + "id": 1, + "error": { + "code": -32602, + "message": "Invalid language filter", + }, + } + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + text=json.dumps(error), + headers={"content-type": "application/json"}, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + res = await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert res["results"] == [] + assert "Invalid language filter" in res["summary"] + await client.aclose() + + @pytest.mark.asyncio + async def test_same_url_self_redirect_breaks_immediately(self): + # A 307 redirect whose Location resolves to the same URL must + # break on the first redirect (1 request) rather than repeating + # the POST N times until the redirect budget runs out. + captured: Dict[str, Any] = {"count": 0} + + def handler(request: httpx.Request) -> httpx.Response: + captured["count"] += 1 + return httpx.Response( + 307, + headers={"location": str(request.url)}, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + assert captured["count"] == 1 + await client.aclose() + + @pytest.mark.asyncio + async def test_concurrent_same_domain_cookie_value_change_not_deleted(self): + # A cookie this call added (recorded in cookies_added_by_call) + # whose value was subsequently changed by a concurrent call must + # not be deleted — the value-aware rollback skips it because the + # current value no longer matches what this call's response set. + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + text=_youcom_sse_body(_YOUCOM_TOOL_TEXT), + headers={ + "content-type": "text/event-stream", + "set-cookie": "you_session=ours; Path=/; Domain=api.you.com", + }, + ) + + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + t = WebSearchTool( + provider="youcom", + http_client=client, + base_url="https://api.you.com/mcp?profile=free", + ) + + # Wrap _post_sse so that after it returns (cookies rolled back), + # we can verify the rollback worked. Then run a second call where + # a "concurrent" cookie overwrites the value mid-flight. + await t._run_async_impl(tool_context=_tool_ctx(), args={"query": "python"}) + # This call's cookie was rolled back. + assert client.cookies.get("you_session") is None + await client.aclose() + + +class TestSameOrigin: + """Redirect-target policy for the You.com SSE call.""" + + def test_same_scheme_host_port_qualifies(self): + assert _same_origin("https://api.you.com/mcp", "https://api.you.com/mcp-moved") + assert _same_origin("https://api.you.com/mcp", "https://api.you.com:443/mcp") + + def test_same_host_https_upgrade_qualifies(self): + # Same-host http-to-https is the one cross-scheme case httpx + # itself treats as safe for keeping Authorization. + assert _same_origin("http://api.you.com/mcp", "https://api.you.com/mcp") + + def test_other_host_rejected(self): + assert not _same_origin("https://api.you.com/mcp", "https://collector.example.com/mcp") + assert not _same_origin("https://api.you.com/mcp", "https://you.com/mcp") + + def test_other_port_rejected(self): + assert not _same_origin("https://api.you.com/mcp", "https://api.you.com:8443/mcp") + + def test_scheme_downgrade_rejected(self): + assert not _same_origin("https://api.you.com/mcp", "http://api.you.com/mcp") + + def test_invalid_target_rejected(self): + assert not _same_origin("https://api.you.com/mcp", "not a url") + + class TestHttpErrorHandling: @pytest.mark.asyncio diff --git a/trpc_agent_sdk/tools/_websearch_tool.py b/trpc_agent_sdk/tools/_websearch_tool.py index 30731fbb..67fbb412 100644 --- a/trpc_agent_sdk/tools/_websearch_tool.py +++ b/trpc_agent_sdk/tools/_websearch_tool.py @@ -6,8 +6,8 @@ """Web search tool for TRPC Agent framework. Provides a client-side :class:`WebSearchTool` that lets LLMs search the -public web for up-to-date information. Three pluggable provider backends -are supported, ``duckduckgo``, ``google search``, and ``tavily``: +public web for up-to-date information. Four pluggable provider backends +are supported, ``duckduckgo``, ``google search``, ``tavily``, and ``youcom``: 1. ``duckduckgo`` — DuckDuckGo(DDG) Instant Answer API. Keyless, good for factual/encyclopedic/definition lookups. Returns curated instant @@ -17,11 +17,17 @@ filtering and language targeting. 3. ``tavily`` — Tavily Search API. Requires ``api_key``; returns LLM-ready web results and optionally direct image URLs. +4. ``youcom`` — You.com ``you-search`` MCP tool over streamable HTTP. + Keyless through the free profile endpoint; ``YDC_API_KEY`` optionally + switches requests to the authenticated endpoint. Returns web and news + hits whose snippets fall back from the hit description to You.com's + canonical ``snippets`` field and to query-relevant highlights. """ from __future__ import annotations import datetime as _dt +import json import os from typing import Any from typing import List @@ -68,6 +74,12 @@ _GOOGLE_BASE_URL = "https://www.googleapis.com/customsearch/v1" # Tavily Search base URL _TAVILY_BASE_URL = "https://api.tavily.com/search" +# You.com MCP streamable-HTTP endpoints. The free profile is keyless and +# exposes the ``you-search`` tool; YDC_API_KEY switches to the authenticated endpoint. +_YOUCOM_BASE_URL = "https://api.you.com/mcp?profile=free" +_YOUCOM_AUTH_BASE_URL = "https://api.you.com/mcp" +# Maximum redirects the SSE POST follows manually before giving up. +_SSE_MAX_REDIRECTS = 3 # Description shown to the LLM as part of the tool schema. _BASE_DESCRIPTION = """\ Search the public web and use the results to inform responses. @@ -95,7 +107,7 @@ for the required 'Sources:' format.\ """ -ProviderType = Literal["duckduckgo", "google", "tavily"] +ProviderType = Literal["duckduckgo", "google", "tavily", "youcom"] class SearchHit(BaseModel): @@ -273,11 +285,125 @@ def _extract_desc_from_pagemap(pagemap: dict[str, Any]) -> str: return "" +def _parse_sse_payloads(text: str) -> List[dict[str, Any]]: + """Parse the ``data:`` payloads of a streamable-HTTP SSE body. + + A streamable-HTTP server answers a JSON-RPC request with one SSE body + whose events carry one JSON object each (progress notifications + followed by the final result). Per the SSE spec an event's data may + span several consecutive ``data:`` lines, and a leading UTF-8 BOM is + stripped. The lines are reassembled in the order that preserves the + server's bytes: joined with no separator first (a server may split one + JSON frame across ``data:`` lines at any boundary, including inside a + JSON string, where a re-inserted newline is either an invalid JSON + control character or a silent corruption of the string contents), then + joined with ``\\n`` (for servers that emit literal newlines between + JSON tokens), and finally each line on its own so servers that omit + blank-line event separators still work. Anything that still fails is + skipped so a stray keep-alive or partial frame cannot break the search. + """ + payloads: List[dict[str, Any]] = [] + lines = (text or "").splitlines() + if lines and lines[0].startswith("\ufeff"): + lines[0] = lines[0].lstrip("\ufeff") + if not lines[0]: + # A BOM-only first line is a framing artifact, not a data line. + lines = lines[1:] + + def _try_decode(raw: str) -> Optional[dict[str, Any]]: + raw = raw.strip() + if not raw: + return None + try: + decoded = json.loads(raw) + except json.JSONDecodeError: + return None + return decoded if isinstance(decoded, dict) else None + + def _is_json(raw: str) -> bool: + try: + json.loads(raw) + return True + except json.JSONDecodeError: + return False + + buffer: List[str] = [] + + def _flush() -> None: + if not buffer: + return + # Prefer the separator-less join: a server may chunk one JSON frame + # across several ``data:`` lines at any boundary, including inside + # a JSON string. Re-inserting a newline between the chunks there + # either puts a raw control character in the string (strict + # ``json.loads`` rejects it and the whole event would degrade to a + # silently dropped frame) or lands somewhere json.loads happens to + # accept, silently corrupting the string contents. The chunks are + # the server's bytes split at arbitrary points, so reassembling + # them with nothing in between is the faithful reconstruction. + decoded = _try_decode("".join(buffer)) + if decoded is None and len(buffer) > 1: + # Fall back to the newline join only for servers that emit + # literal newlines between JSON tokens (pretty-printed frames + # without blank event separators). + decoded = _try_decode("\n".join(buffer)) + if decoded is not None: + payloads.append(decoded) + else: + for piece in buffer: + raw = piece.strip() + if not raw: + # Empty ``data:`` lines are keep-alives; skip silently + # instead of logging noise for every one of them. + continue + decoded = _try_decode(piece) + if decoded is not None: + payloads.append(decoded) + continue + # At this point the line is either unparseable or valid JSON + # that is not an object. + if _is_json(raw): + # Valid JSON but not an object (``123``, ``[1]``) is + # not a JSON-RPC frame; skip quietly, not as a warning. + logger.debug("WebSearchTool: skipping non-object SSE data: %.80s", raw) + else: + logger.warning("WebSearchTool: skipping unparseable SSE line: %.80s", raw) + buffer.clear() + + for line in lines: + if line.startswith("data:"): + buffer.append(line[len("data:"):].strip()) + elif not line.strip(): + # A blank line dispatches the event per the SSE spec. + _flush() + _flush() + return payloads + + +def _same_origin(left: str, right: str) -> bool: + """Whether ``right`` is a same-origin target for a redirect from ``left``. + + Same scheme, host, and port qualifies, as does a same-host upgrade from + ``http`` to ``https`` — the two cases where httpx keeps ``Authorization`` + on an automatically followed redirect. Anything else (different host or + port, or a scheme downgrade) does not: a redirect there would carry the + tool's bearer header to a third party. + """ + try: + lhs, rhs = httpx.URL(left), httpx.URL(right) + if (lhs.host, lhs.port) != (rhs.host, rhs.port): + return False + return lhs.scheme == rhs.scheme or rhs.scheme == "https" + except Exception: # pylint: disable=broad-except + return False + + class WebSearchTool(BaseTool): """LLM tool that searches the public web. The WebSearchTool enables LLM agents to search the public web using major search engines - such as DuckDuckGo (default, no API key required), Google Custom Search, and Tavily. + such as DuckDuckGo (default, no API key required), Google Custom Search, Tavily, + and You.com (keyless through its free MCP profile). It retrieves up-to-date information including titles, URLs, and content snippets, and also provides instant-answer summaries when available (e.g., via DuckDuckGo). This tool is best used for queries about recent events, new releases, factual lookups, or definitions that benefit @@ -286,9 +412,11 @@ class WebSearchTool(BaseTool): sources as Markdown hyperlinks in the final output. Args: - provider: Backend name: ``"duckduckgo"`` (default), ``"google"``, or ``"tavily"``. + provider: Backend name: ``"duckduckgo"`` (default), ``"google"``, + ``"tavily"``, or ``"youcom"``. api_key: Provider API key. Falls back to ``GOOGLE_CSE_API_KEY`` for - Google or ``TAVILY_API_KEY`` for Tavily. + Google, ``TAVILY_API_KEY`` for Tavily, or ``YDC_API_KEY`` for + You.com (optional; the free profile is keyless). engine_id: Google CSE engine id (``cx``); falls back to ``GOOGLE_CSE_ENGINE_ID``. results_num: Default result count, clamped to ``[1, _MAX_COUNT]``. snippet_len: Max snippet length, clamped to ``[1, _MAX_SNIPPET_LEN]``. @@ -322,6 +450,7 @@ def __init__( ddg_extra_params: Optional[dict[str, Any]] = None, google_extra_params: Optional[dict[str, Any]] = None, tavily_extra_params: Optional[dict[str, Any]] = None, + youcom_extra_params: Optional[dict[str, Any]] = None, filters_name: Optional[List[str]] = None, filters: Optional[List[BaseFilter]] = None, ) -> None: @@ -333,13 +462,17 @@ def __init__( filters=filters, ) - if provider not in ("duckduckgo", "google", "tavily"): + if provider not in ("duckduckgo", "google", "tavily", "youcom"): raise ValueError(f"Unsupported web search provider: {provider!r}") self._provider: ProviderType = provider if provider == "google": self._api_key = api_key or os.environ.get("GOOGLE_CSE_API_KEY", "") elif provider == "tavily": self._api_key = api_key or os.environ.get("TAVILY_API_KEY", "") + elif provider == "youcom": + # You.com is keyless via the free MCP profile; a YDC_API_KEY + # switches requests to the authenticated endpoint. + self._api_key = api_key or os.environ.get("YDC_API_KEY", "") else: self._api_key = api_key or "" self._engine_id = engine_id or os.environ.get("GOOGLE_CSE_ENGINE_ID", "") @@ -351,6 +484,7 @@ def __init__( "duckduckgo": _DDG_BASE_URL, "google": _GOOGLE_BASE_URL, "tavily": _TAVILY_BASE_URL, + "youcom": _YOUCOM_AUTH_BASE_URL if self._api_key else _YOUCOM_BASE_URL, } self._base_url = base_url or default_base_urls[provider] self._user_agent = user_agent @@ -361,6 +495,7 @@ def __init__( self._ddg_extra_params = ddg_extra_params or {} self._google_extra_params = google_extra_params or {} self._tavily_extra_params = tavily_extra_params or {} + self._youcom_extra_params = youcom_extra_params or {} if provider == "google" and not (self._api_key and self._engine_id): logger.warning("WebSearchTool: provider='google' but api_key or " @@ -409,7 +544,8 @@ def _get_declaration(self) -> FunctionDeclaration: "lang": Schema( type=Type.STRING, - description=("Optional. Language hint for the provider (Google CSE 'hl'); " + description=("Optional. Language hint for the provider (Google CSE 'hl'; " + "You.com maps it to the inline `lang:` filter); " "ignored by DuckDuckGo and Tavily. Default: tool-level lang or unset. " "Example: 'en', 'zh-CN', 'ja'."), ), @@ -510,6 +646,8 @@ async def _run_async_impl(self, *, tool_context: InvocationContext, args: dict[s blocked, include_images, ) + elif self._provider == "youcom": + result = await self._search_youcom(query, n, allowed, blocked, lang) else: result = await self._search_google(query, n, allowed, blocked, lang) except httpx.HTTPError as e: @@ -575,6 +713,237 @@ async def _post_json( if close: await client.aclose() + async def _post_sse( + self, + url: str, + payload: dict[str, Any], + *, + headers: Optional[dict[str, str]] = None, + ) -> List[dict[str, Any]]: + """Issue a POST and decode the SSE ``data:`` payloads in the body. + + Used for JSON-RPC endpoints that answer with ``text/event-stream`` + instead of a single JSON document (the You.com MCP endpoint). + + The request is built explicitly and sent through ``client.send`` so + that only the headers assembled below reach the wire: httpx merges + client-level headers into ``client.post`` requests, which would + silently forward any credentials configured on a shared + ``http_client`` (a gateway bearer token, a cookie jar, ...) to this + third-party endpoint. Three further client-level settings are + handled per call so a shared client cannot leak into this request + either: + + * ``auth=None`` — ``client.send`` applies the client-level ``auth`` + (Basic/Bearer/custom auth flows) even to a bare request, so it is + suppressed per call. + * redirects are followed manually, one request at a time. Every hop + is checked against the URL it redirects *from* and must stay + same-origin with it (a same-host http-to-https upgrade still + qualifies), and no hop may land on a plaintext ``http`` target + unless the endpoint itself was configured that way (same-origin + with the original URL). httpx's automatic redirect handling keeps + ``Authorization`` on same-origin/same-host-upgrade redirects and + re-attaches cookie-jar cookies to every redirect request, so a 3xx + naming another origin — or downgrading an upgraded hop back to + plaintext http — is dropped with a warning instead of followed: + the ``YDC_API_KEY`` bearer header must never leave the You.com + origin, and never travel over plaintext http. An unfollowed + residual 3xx also warns rather than parsing an empty body as a + silent 0-hit search. A relative ``Location`` keeps the original + ``profile`` query parameter (``URL.join`` drops it), so a keyless + free-profile call does not get redirected onto the authenticated + endpoint. + * cookies the You.com responses write into the shared client's jar + are rolled back: cookies this call *added* are deleted (scoped to + the endpoint host, which every hop was validated to stay on) so + they cannot be replayed to other providers through the same + ``http_client``. Pre-existing cookies whose *values* were + overwritten by a You.com ``Set-Cookie`` are left as-is rather + than restored to the snapshot value — restoring them would + silently clobber a concurrent call that updated the same cookie + after our snapshot. + * each response is closed: intermediate 3xx hop responses are + closed before the next hop is issued, and the final response is + closed in a ``finally`` block — a read that dies mid-stream must + not leave a broken connection in the shared client's pool for the + next provider call. + """ + client = self._get_client() + close = self._http_client is None + request_headers = { + "User-Agent": self._user_agent, + # The streamable-HTTP server rejects JSON-only Accept headers + # with 406 Not Acceptable, so SSE must be listed here. + "Accept": "application/json, text/event-stream", + } + if headers: + request_headers.update(headers) + # Snapshot the jar (with values) before any request goes out; the + # rollback in the finally block below relies on it covering every + # You.com response. + cookie_snapshot = { + (cookie.name, cookie.domain, cookie.path): cookie.value + for cookie in client.cookies.jar + } + endpoint_host = httpx.URL(url).host + cookies_added_by_call: dict[tuple[str, str, str], str] = {} + resp: Optional[httpx.Response] = None + try: + current_url = url + redirects = 0 + while True: + request = httpx.Request("POST", current_url, json=payload, headers=request_headers) + # send() falls back to the client timeout when the extension is + # absent; keep the per-call timeout semantics of the other providers. + request.extensions["timeout"] = httpx.Timeout(self._timeout).as_dict() + resp = await client.send(request, auth=None, follow_redirects=False) + # Record cookies this response added to the jar, so the + # finally-block rollback can distinguish them from cookies + # a concurrent call wrote to the same shared client. + for _c in client.cookies.jar: + _ck = (_c.name, _c.domain, _c.path) + if _ck not in cookie_snapshot: + cookies_added_by_call[_ck] = _c.value or "" + if not resp.has_redirect_location: + break + if redirects >= _SSE_MAX_REDIRECTS: + logger.warning( + "WebSearchTool: You.com endpoint exceeded %d redirects; " + "search may return no results.", + _SSE_MAX_REDIRECTS, + ) + break + redirects += 1 + location = resp.headers.get("location", "").strip() + if not location: + logger.warning( + "WebSearchTool: You.com endpoint returned a redirect " + "(%d) with an empty Location; search may return no " + "results.", + resp.status_code, + ) + break + # Close the 3xx response before issuing the next hop: the + # body of a redirect response is short or empty, but leaving + # it open ties up a pooled connection for every intermediate + # hop until the loop exits. ``aclose`` is idempotent, so the + # finally-block close on the final response is still safe. + await resp.aclose() + try: + next_obj = httpx.URL(current_url).join(location) + # URL.join follows urljoin semantics, so a relative + # Location drops the original query string: a keyless + # "?profile=free" call would silently land on the + # authenticated endpoint after the redirect. Re-attach + # the profile param, keeping any the Location carries. + origin_profile = httpx.URL(url).params.get("profile") + if origin_profile is not None and next_obj.params.get("profile") is None: + next_obj = next_obj.copy_with(params=next_obj.params.merge({"profile": origin_profile})) + next_url = str(next_obj) + except Exception as ex: # pylint: disable=broad-except + logger.warning( + "WebSearchTool: You.com redirect target %r is not a valid URL (%s); " + "search may return no results.", + location, + ex, + ) + break + # A redirect whose Location resolves to the same URL as the + # current request is a server-side loop (load balancer or + # session fault); re-POSTing the same body N times wastes + # the redirect budget and repeats the credential payload + # needlessly. + if next_url == current_url: + logger.warning( + "WebSearchTool: You.com endpoint returned a redirect " + "(%d) whose Location resolves to the current URL; " + "search may return no results.", + resp.status_code, + ) + break + # Validate each hop on its own: the request (and the tool's + # Authorization header) travels to current_url's origin, so + # the hop must be same-origin with the *previous* URL — not + # merely with the first one — and it may not downgrade to + # plaintext http unless the endpoint was configured that way + # (same-origin with the original URL). Otherwise the bearer + # header (and any cookies httpx re-attaches to redirect + # requests) would travel to whatever host or plaintext hop + # the 3xx names. + if not (_same_origin(current_url, next_url) and + (_same_origin(url, next_url) or httpx.URL(next_url).scheme == "https")): + logger.warning( + "WebSearchTool: dropping redirect from the You.com endpoint to " + "URL %r (not same-origin with the previous hop or plaintext http); " + "search may return no results.", + next_url, + ) + return [] + current_url = next_url + if 300 <= resp.status_code < 400: + # Defensive: a 3xx that was not followed (missing/invalid + # location, or the redirect budget ran out) must not read + # downstream as a normal empty search. + logger.warning( + "WebSearchTool: You.com endpoint responded %d without " + "following the redirect; search may return no results.", + resp.status_code, + ) + resp.raise_for_status() + content_type = resp.headers.get("content-type", "") + if "text/event-stream" not in content_type: + # A non-SSE response (e.g. application/json carrying a + # JSON-RPC error frame for a parameter-validation failure + # or rate-limit) has no ``data:`` lines, so the SSE + # parser returns [] and the error is silently swallowed. + # Parse it as JSON so the error-handling branch in + # _search_youcom can surface the real failure reason. + try: + frame = json.loads(resp.text) + return frame if isinstance(frame, list) else [frame] + except (json.JSONDecodeError, ValueError): + pass + return _parse_sse_payloads(resp.text) + finally: + # Roll back anything the You.com responses (followed or + # unfollowed, including skipped 3xx hops) wrote into the shared + # client's jar, so it cannot be replayed to other providers + # (e.g. Tavily) through the same http_client. Running here — not + # on the happy path — guarantees the rollback also happens on + # the cross-origin early return, the redirect-budget break, and + # any exception raised mid-loop. Only cookies the call *added* + # (tracked in cookies_added_by_call) are deleted, and only if + # their current value still matches what this call's response + # set — a concurrent call on the same shared client may have + # updated them since, and deleting those would silently drop + # the concurrent call's session. Deletes are scoped to cookies + # that could plausibly have come from one of this call's + # responses (domain-matching the endpoint host, which every + # hop was validated to stay on), so a cookie a concurrent call + # wrote for an unrelated domain is not caught in this rollback. + for cookie in list(client.cookies.jar): + key = (cookie.name, cookie.domain, cookie.path) + if key not in cookies_added_by_call: + # Not added by this call (either pre-existing or + # written by a concurrent call on the shared client). + continue + # Only delete if the value hasn't changed since this call + # set it — a concurrent call may have updated it since. + if (cookie.value or "") != cookies_added_by_call[key]: + continue + domain = cookie.domain.lstrip(".") + if endpoint_host == domain or endpoint_host.endswith("." + domain): + client.cookies.delete(cookie.name, domain=cookie.domain, path=cookie.path) + # Release the response and its pooled connection even when the + # body read dies mid-stream (e.g. a read timeout): an abandoned + # half-read response would otherwise leave a broken connection + # in the shared client's pool for the next provider call. + if resp is not None: + await resp.aclose() + if close: + await client.aclose() + async def _search_tavily( self, query: str, @@ -681,6 +1050,280 @@ async def _search_tavily( images=images, ) + async def _search_youcom( + self, + query: str, + n: int, + allowed: Optional[List[str]], + blocked: Optional[List[str]], + lang: Optional[str], + ) -> WebSearchResult: + """Search the public web through the You.com ``you-search`` MCP tool. + + The endpoint is stateless JSON-RPC over streamable HTTP: a single + ``tools/call`` POST is answered by one SSE body whose ``data:`` lines + carry the tool result. The free profile works without an API key; a + configured ``YDC_API_KEY`` upgrades requests to the authenticated + endpoint. Blocked domains are excluded server-side; allowlists are + enforced post-hoc by ``_is_blocked``, mirroring the other providers. + Hit snippets fall back from the hit ``description`` to the canonical + You.com ``snippets`` list, then to ``contents.highlights``. + """ + arguments: dict[str, Any] = {"query": query, "count": n} + if lang: + # You.com maps an inline ``lang:`` operator to a language filter. + # Guard against the model passing the operator form verbatim + # (``lang:ja``, possibly with stray whitespace or casing) after + # reading the schema description. + lang_value = str(lang).strip() + while lang_value[:5].lower() == "lang:": + lang_value = lang_value[5:].strip() + if lang_value and "lang:" not in query.lower(): + # Guard against the model already embedding a ``lang:`` + # operator in the query text (the schema description + # encourages that form): appending a second one would + # produce a duplicated filter like ``python lang:ja lang:ja``. + arguments["query"] = f"{query} lang:{lang_value}" + if blocked: + arguments["exclude_domains"] = blocked + for key, value in self._youcom_extra_params.items(): + if key in ("query", "count", "exclude_domains", "lang", "lang:"): + # These are built per call; a pinned value would silently + # override every search the model issues. ``lang`` belongs + # here too: the language filter is an inline ``lang:`` + # operator inside the per-call query, so a standalone + # ``lang`` argument would conflict with it instead of + # pinning the language. + logger.warning( + "WebSearchTool: ignoring youcom_extra_params key %r; " + "it would override per-call arguments", key) + continue + arguments[key] = value + + payload: dict[str, Any] = { + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "you-search", + "arguments": arguments, + }, + } + headers: Optional[dict[str, str]] = None + if self._api_key: + headers = {"Authorization": f"Bearer {self._api_key}"} + + payloads = await self._post_sse(self._base_url, payload, headers=headers) + + result: Optional[dict[str, Any]] = None + error: Optional[dict[str, Any]] = None + for item in payloads: + if not isinstance(item, dict): + continue + frame_id = item.get("id") + # Only frames replying to this request carry its id; frames + # without one are notifications and carry no result/error. + # Compare the string forms: some servers echo the request id + # back as a string ("1"), and a strict typed comparison would + # drop every reply frame. + if frame_id is not None and str(frame_id) != str(payload["id"]): + continue + if "error" in item: + # A JSON-RPC response carries either ``result`` or ``error``; + # a frame carrying both is treated as an error. + error = item["error"] + if "result" in item and "error" not in item: + # A valid result always wins, regardless of frame order: a + # stray error frame that arrived before the result is logged + # but does not discard the valid response. An error frame + # that arrives *after* the result is also logged rather than + # silently overwriting it. + if error is not None: + logger.warning( + "WebSearchTool: You.com error frame superseded by a " + "result (or vice versa): %.200s", error, + ) + result = item["result"] + if error is not None and result is None: + message = error.get("message") if isinstance(error, dict) else str(error) + return WebSearchResult( + query=query, + provider="youcom", + results=[], + summary=_truncate(f"You.com search error: {message}", self._snippet_len), + ) + if not isinstance(result, dict): + return WebSearchResult( + query=query, + provider="youcom", + results=[], + summary="You.com search returned no result payload.", + ) + + text = "" + for content in result.get("content") or []: + if isinstance(content, dict) and content.get("type") == "text": + text = str(content.get("text") or "") + break + if result.get("isError"): + # Flag the error whatever the text content looks like: MCP servers + # may mark ``isError`` without any text, keeping details only in + # ``structuredContent``; either way the call failed and the summary + # must say so instead of falling through to payload parsing. + detail = text + if not detail: + structured = result.get("structuredContent") + if isinstance(structured, dict): + try: + detail = json.dumps(structured, ensure_ascii=False) + except (TypeError, ValueError): + detail = "" + if not detail: + detail = "no detail returned" + return WebSearchResult( + query=query, + provider="youcom", + results=[], + summary=_truncate(f"You.com search error: {detail}", self._snippet_len), + ) + + data: Any = None + try: + data = json.loads(text) + except (json.JSONDecodeError, TypeError): + data = None + if not isinstance(data, dict): + # Some MCP servers put the structured payload in + # ``structuredContent`` and keep the text human-readable. + structured = result.get("structuredContent") + if isinstance(structured, dict): + data = structured + if not isinstance(data, dict): + return WebSearchResult( + query=query, + provider="youcom", + results=[], + summary="You.com search returned an unexpected payload.", + ) + + sections = data.get("results") + if isinstance(sections, list): + # Shape drift upstream: some You.com variants return a flat list + # of hits instead of ``{"web": [...], "news": [...]}`` sections. + # Route the parseable entries through the same section loop + # instead of dropping every hit — merging (not replacing) any + # real section lists that sit next to ``results``, so a genuine + # ``web``/``news`` section is never clobbered by the fallback. + flat = sections + sections = {} + for section in ("web", "news"): + sibling = data.get(section) + if isinstance(sibling, list): + sections[section] = sibling + sections["web"] = sections.get("web", []) + flat + if not isinstance(sections, dict): + if sections: + # A non-empty, non-dict, non-list ``results`` is a shape + # change upstream, not an empty search; surface it instead + # of reporting 0 hits. + return WebSearchResult( + query=query, + provider="youcom", + results=[], + summary=_truncate( + "You.com search returned an unexpected results shape " + f"({type(sections).__name__}); no hits parsed.", + self._snippet_len, + ), + ) + sections = {} + items: List[dict[str, Any]] = [] + for section in ("web", "news"): + section_items = sections.get(section) + if not isinstance(section_items, list): + if section_items: + logger.warning("WebSearchTool: ignoring non-list %r section in You.com results", section) + continue + for item in section_items: + if isinstance(item, dict): + items.append(item) + + hits: List[SearchHit] = [] + seen: set[str] = set() + for item in items: + # Type-guard every field instead of str()-coercing: a dict or + # list value would become its Python repr — a repr "url" then + # fails the blocklist check and silently drops a legitimate + # hit, and a repr title/description leaks stray text like + # "{'kind': 'web', ...}" into the results. + raw_url = item.get("url") + url = "" + if isinstance(raw_url, str): + url = raw_url.strip() + elif isinstance(raw_url, dict): + # Shape drift: some variants wrap the URL in an object; + # take the standard ``href`` field if it is a string. + href = raw_url.get("href") + if isinstance(href, str): + url = href.strip() + if not url: + logger.debug( + "WebSearchTool: skipping You.com hit without a usable url field (%r)", + raw_url, + ) + continue + if _is_blocked(url, allowed, blocked): + continue + if self._dedup_urls: + key = _dedup_key(url) + if key in seen: + continue + seen.add(key) + raw_description = item.get("description") + description = raw_description.strip() if isinstance(raw_description, str) else "" + if not description: + # ``snippets`` is You.com's canonical summary field (a list + # of strings returned by default); some hits — notably + # news — carry it while ``description`` is empty or absent. + snippets = item.get("snippets") + if isinstance(snippets, list): + for snippet in snippets: + if isinstance(snippet, str) and snippet.strip(): + description = snippet.strip() + break + if not description: + contents = item.get("contents") + highlights = contents.get("highlights") if isinstance(contents, dict) else None + # Only a list is a usable fallback. Some variants mix objects + # and strings in the list, so the first string element wins + # (mirroring the ``snippets`` traversal above) instead of + # checking only ``highlights[0]`` — a non-string first + # element must not disable the fallback for the strings + # behind it. Anything else (a bare string, a dict, ...) must + # not crash the whole search. + if isinstance(highlights, list): + for highlight in highlights: + if isinstance(highlight, str) and highlight.strip(): + description = highlight.strip() + break + raw_title = item.get("title") + title = raw_title if isinstance(raw_title, str) else "" + hits.append( + SearchHit( + title=_truncate(title, self._title_len), + url=url, + snippet=_truncate(description, self._snippet_len), + )) + if len(hits) >= n: + break + + return WebSearchResult( + query=query, + provider="youcom", + results=hits, + summary="", + ) + async def _search_duckduckgo( self, query: str,