feat: search_infinity mcp version 0.1.0

server/mcp_server_askecho_search_infinity/README.md

@@ -1,8 +1,8 @@
-# 融合信息搜索 MCP Server
+# 联网搜索API MCP Server
 ## 版本信息
 v0.1.0
 ## 产品描述
-依托字节强大的搜索能力，提供适配大模型数据结构的联网搜索内容，助力提升大模型知识获取、时效性及回答准确性。
+火山引擎联网搜索API，提供网页与图片搜索能力，帮助大模型获取更准确、更新鲜的外部信息。
 ## 分类
 火山引擎云原生
 ## 标签
@@ -16,7 +16,7 @@ v0.1.0
 #### 类型
 saas
 #### 详细描述
-根据用户输入问题，提供基于联网搜索的大模型总结后回复内容
+根据用户输入问题，返回联网搜索结果，支持网页和图片搜索
 #### 调试所需的输入参数:
 输入：
 ```json
@@ -28,36 +28,50 @@ saas
         ],
         "properties": {
             "Query": {
-                "description": "用户搜索 query，1~100 个字符 (过长会截断)，不支持多词搜索",
+                "description": "用户搜索 query，1~100 个字符",
                 "type": "string"
             },
             "Count": {
-                "description": "返回条数，最多50条，不传默认10条",
+                "description": "返回条数；web 最多 50 条，image 最多 5 条，不传默认 10 条",
+                "type": "number"
+            },
+            "SearchType": {
+                "description": "搜索类型，仅支持 web 或 image，默认 web",
+                "type": "string"
+            },
+            "TimeRange": {
+                "description": "web 搜索时间范围，可选 OneDay/OneWeek/OneMonth/OneYear 或 YYYY-MM-DD..YYYY-MM-DD",
+                "type": "string"
+            },
+            "AuthLevel": {
+                "description": "权威等级过滤，0 为默认，1 为非常权威",
                 "type": "number"
             }
         }
     },
     "name": "web_search",
-    "description": "联网搜索能力调用"
+    "description": "联网搜索 API 调用"
 }
 ```
 输出：
 ```json
-联网搜索结果，结构参考文档的响应体部分 https://www.volcengine.com/docs/85508/1650263
+联网搜索结果，结构参考官方 API 文档 https://www.volcengine.com/docs/87772/2272953
 ```
 
 #### 最容易被唤起的 Prompt示例
 联网搜索北京周边游攻略
 ## 可适配平台
 Trae，Cursor，Python
 ## 服务开通链接 (整体产品)
-登录火山控制台，开通【融合信息检索】，服务开通链接：https://console.volcengine.com/ask-echo/web-search
+登录火山控制台，开通【联网搜索API】，服务开通链接：https://console.volcengine.com/search-infinity/web-search
+API Key 创建链接：https://console.volcengine.com/search-infinity/api-key
 ## 鉴权方式
 - API Key鉴权
 - 火山引擎的AKSK鉴权体系
 ## 安装部署
 ### 前置准备
-- Python 3.12+
+- Python 3.12 / 3.13
+- 当前不支持 Python 3.14 beta，`mcp` / `pydantic` 依赖链在该版本上仍存在兼容性问题
 - UV
   **Linux/macOS:**
 ```bash
@@ -84,7 +98,7 @@ uv run mcp-server-askecho-search-infinity -t streamable-http
 ```
 ## 部署
 ### UVX
-鉴权信息，火山引擎AK SK,与ASK_ECHO_SEARCH_INFINITY_API_KEY接入二选一即可
+鉴权信息，火山引擎 AK/SK 与 `ASK_ECHO_SEARCH_INFINITY_API_KEY` 二选一即可
 ```json
 {
   "mcpServers": {
@@ -105,4 +119,4 @@ uv run mcp-server-askecho-search-infinity -t streamable-http
 }
 ```
 ## License
-volcengine/mcp-server is licensed under the [MIT License](https://github.com/volcengine/mcp-server/blob/main/LICENSE)
+volcengine/mcp-server is licensed under the [MIT License](https://github.com/volcengine/mcp-server/blob/main/LICENSE)

server/mcp_server_askecho_search_infinity/pyproject.toml

@@ -1,9 +1,9 @@
 [project]
 name = "mcp-server-askecho-search-infinity"
 version = "0.1.0"
-description = "Search Infinity MCP Server"
+description = "Web Search API MCP Server"
 readme = "README.md"
-requires-python = ">=3.12"
+requires-python = ">=3.12,<3.14"
 dependencies = [
     "mcp>=1.9.4",
     "aiohttp>=3.9.0",

server/mcp_server_askecho_search_infinity/src/mcp_server_askecho_search_infinity/api/api_key_auth.py

@@ -1,4 +1,3 @@
-from dataclasses import asdict
 import json
 import aiohttp
 from ..model import *
@@ -19,11 +18,11 @@ async def web_search_api_key_auth(api_key: str, req: WebSearchRequest, tool_name
             url=f"https://{Host}/search_api/web_search",
             headers=header,
             timeout=aiohttp.ClientTimeout(total=3000),
-            data=json.dumps(asdict(req))
+            data=json.dumps(req.to_payload())
         ) as response:
             # 在上下文内读取所有数据，避免连接关闭问题
             response.raise_for_status()  # 手动调用
             data = await response.json()
             return data
         return None
-    return None
+    return None

server/mcp_server_askecho_search_infinity/src/mcp_server_askecho_search_infinity/api/volcengine_auth.py

@@ -1,7 +1,6 @@
 import datetime
 import hashlib
 import hmac
-from dataclasses import asdict
 import json
 from urllib.parse import quote
 from ..model import *
@@ -10,7 +9,7 @@
 
 Service = "volc_torchlight_api"
 Version = "2025-01-01"
-Region = "cn-north-1"
+Region = "cn-beijing"
 Host = "mercury.volcengineapi.com"
 ContentType = "application/json"
 
@@ -20,19 +19,25 @@ async def web_search_volcengine_auth(ak: str, sk: str, req: WebSearchRequest, to
     headers = {
         "X-Traffic-Tag": f"ark_mcp_server_{tool_name}",
     }
-    return await volcengine_auth_request("POST", now, {}, headers, ak, sk, "WebSearch", json.dumps(asdict(req)))
+    return await volcengine_auth_request("POST", now, {}, headers, ak, sk, "WebSearch", json.dumps(req.to_payload()))
 
 
 def norm_query(params):
     query = ""
     for key in sorted(params.keys()):
-        if type(params[key]) == list:
-            for k in params[key]:
+        if isinstance(params[key], list):
+            for value in params[key]:
                 query = (
-                        query + quote(key, safe="-_.~") + "=" + quote(k, safe="-_.~") + "&"
+                        query + quote(key, safe="-_.~") + "=" + quote(value, safe="-_.~") + "&"
                 )
         else:
-            query = (query + quote(key, safe="-_.~") + "=" + quote(params[key], safe="-_.~") + "&")
+            query = (
+                query
+                + quote(key, safe="-_.~")
+                + "="
+                + quote(str(params[key]), safe="-_.~")
+                + "&"
+            )
     query = query[:-1]
     return query.replace("+", "%20")
 
@@ -72,21 +77,25 @@ async def volcengine_auth_request(method, date, query, header, ak, sk, action, b
         "X-Date": x_date,
         "Content-Type": request_param["content_type"],
     }
-    signed_headers_str = ";".join(
-        ["content-type", "host", "x-content-sha256", "x-date"]
-    )
+    signed_header_keys = ["content-type", "host", "x-content-sha256", "x-date"]
+    canonical_header_lines = [
+        "content-type:" + request_param["content_type"],
+        "host:" + request_param["host"],
+        "x-content-sha256:" + x_content_sha256,
+        "x-date:" + x_date,
+    ]
+    traffic_tag = header.get("X-Traffic-Tag")
+    if traffic_tag:
+        signed_header_keys.append("x-traffic-tag")
+        canonical_header_lines.append("x-traffic-tag:" + traffic_tag)
+    signed_header_keys.sort()
+    canonical_header_lines.sort()
+    signed_headers_str = ";".join(signed_header_keys)
     canonical_request_str = "\n".join(
         [request_param["method"].upper(),
          request_param["path"],
          norm_query(request_param["query"]),
-         "\n".join(
-             [
-                 "content-type:" + request_param["content_type"],
-                 "host:" + request_param["host"],
-                 "x-content-sha256:" + x_content_sha256,
-                 "x-date:" + x_date,
-             ]
-         ),
+         "\n".join(canonical_header_lines),
          "",
          signed_headers_str,
          x_content_sha256,
@@ -122,4 +131,4 @@ async def volcengine_auth_request(method, date, query, header, ak, sk, action, b
             data = await response.json()
             return data
         return None
-    return None
+    return None

server/mcp_server_askecho_search_infinity/src/mcp_server_askecho_search_infinity/model.py

@@ -1,6 +1,12 @@
+import datetime
+import re
 from dataclasses import dataclass
 from typing import Optional, List
 
+TIME_RANGE_SHORTCUTS = {"OneDay", "OneWeek", "OneMonth", "OneYear"}
+DATE_RANGE_PATTERN = re.compile(r"^(\d{4}-\d{2}-\d{2})\.\.(\d{4}-\d{2}-\d{2})$")
+SUPPORTED_SEARCH_TYPES = {"web", "image"}
+
 
 @dataclass
 class Error:
@@ -32,16 +38,22 @@ class WebSearchRequest:
     SearchType: str = "web"
     Count: int = 10
     Filter: Optional[dict] = None
-    NeedSummary: bool = True
-    TimeRange: str = ""
+    NeedSummary: Optional[bool] = None
+    TimeRange: Optional[str] = None
 
-    def __post_init__(self):
-        if self.Filter is None:
-            self.Filter = {
-                "NeedContent": True,
-                "NeedUrl": True,
-                "Sites": ""
-            }
+    def to_payload(self):
+        payload = {
+            "Query": self.Query,
+            "SearchType": self.SearchType,
+            "Count": self.Count,
+        }
+        if self.SearchType == "web":
+            payload["NeedSummary"] = True
+            if self.Filter:
+                payload["Filter"] = self.Filter
+            if self.TimeRange:
+                payload["TimeRange"] = self.TimeRange
+        return payload
 
 
 @dataclass
@@ -61,4 +73,67 @@ class SearchResult:
 
 @dataclass
 class WebSearchResponse:
-    results: List[SearchResult]
+    results: List[SearchResult]
+
+
+def validate_time_range(time_range: Optional[str]) -> Optional[str]:
+    if not time_range:
+        return None
+    if time_range in TIME_RANGE_SHORTCUTS:
+        return time_range
+
+    match = DATE_RANGE_PATTERN.match(time_range)
+    if not match:
+        raise ValueError(
+            "TimeRange 需为 OneDay/OneWeek/OneMonth/OneYear，或日期区间 YYYY-MM-DD..YYYY-MM-DD。"
+        )
+
+    start_text, end_text = match.groups()
+    try:
+        start_date = datetime.date.fromisoformat(start_text)
+        end_date = datetime.date.fromisoformat(end_text)
+    except ValueError as exc:
+        raise ValueError("TimeRange 中的日期需为有效的 YYYY-MM-DD。") from exc
+
+    if start_date > end_date:
+        raise ValueError("TimeRange 的开始日期不能晚于结束日期。")
+
+    return time_range
+
+
+def build_web_search_request(
+        query: str,
+        count: int = 10,
+        search_type: str = "web",
+        time_range: Optional[str] = None,
+        auth_level: int = 0,
+) -> WebSearchRequest:
+    normalized_query = (query or "").strip()
+    if not normalized_query:
+        raise ValueError("Query 不能为空。")
+    if len(normalized_query) > 100:
+        raise ValueError("Query 长度需为 1~100 个字符。")
+
+    if search_type not in SUPPORTED_SEARCH_TYPES:
+        raise ValueError("SearchType 仅支持 web 或 image。")
+
+    if count < 1:
+        raise ValueError("Count 需大于等于 1。")
+    max_count = 50 if search_type == "web" else 5
+    if count > max_count:
+        raise ValueError(f"{search_type} 类型最多返回 {max_count} 条。")
+
+    if auth_level not in {0, 1}:
+        raise ValueError("AuthLevel 仅支持 0 或 1。")
+
+    normalized_time_range = validate_time_range(time_range) if search_type == "web" else None
+    filters = {"AuthInfoLevel": auth_level} if search_type == "web" and auth_level > 0 else None
+
+    return WebSearchRequest(
+        Query=normalized_query,
+        SearchType=search_type,
+        Count=count,
+        Filter=filters,
+        NeedSummary=True if search_type == "web" else None,
+        TimeRange=normalized_time_range,
+    )