互联网搜索服务

MCP Internet Search v3.0 是一个支持多后端、页面内容抓取、缓存、结构化指标和全面输入验证的互联网搜索服务,专为AI代理优化。

  • by elrcosta-lab
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 elrcosta-lab/Tool-MCP-Search-Duckduckgo

使用说明

MCP互联网搜索v3.0

用于互联网搜索的MCP服务器,具有多后端支持、页面内容获取、缓存、结构化指标和全面的输入验证。针对AI代理消费进行了优化。

特征

  • 多后端搜索 --DuckDuckGo(默认),SearXNG(通过选择加入 SEARXNG_URL),勇敢搜索API(选择通过 BRAVE_API_KEY).自动回退并重试。
  • 3种搜索类型 — text (网络), news (文章带日期+来源), images (带缩略图+尺寸)。
  • 页面内容获取 — fetch_page 该工具通过BeautifulSoup提取干净的文本。URL上的SSRF保护。
  • 内联内容 — fetch_content=True 上 search_internet 返回结果旁边的整页文本。
  • 异步贯穿始终 — asyncio.to_thread() 对于DuckDuckGo同步呼叫, httpx.AsyncClient 对于HTTP。
  • 缓存 --在记忆中 cachetools.TTLCache (512个条目,5分钟TTL,可通过配置 CACHE_TTL/CACHE_MAX_SIZE).命中在<5ms内返回。
  • 输入验证 --查询长度限制(500)、空白规范化、参数边界、SSRF保护(通过DNS解析阻止私有/保留IP)。
  • 安全搜索 — safesearch 参数(off, moderate, strict).
  • 分页 — offset 用于分页结果的参数。
  • 结构化指标 --via structlog:持续时间、后端来源、缓存命中/未命中、错误率。
  • 错误分类 — error_type 字段帮助代理决定重试策略。
  • 一致的响应模式 --成功和错误的结构相同。
  • 地理区域 --60多个地区。默认 br-br.
  • 时间过滤 — time_filter='d' (天), 'w' (周), 'm' (月), 'y' (年)。
  • 使用回退重试 --指数退避(基数1,2倍乘数)可通过以下方式配置 max_retries/backoff_base.
  • 向后兼容性 — search_duckduckgo 该工具保留为已弃用的别名。
  • mypy——严格 --13个源文件中有0个类型错误。
  • 96%的测试覆盖率 --173个单元/集成/合同测试+14个针对真实API的E2E测试。

安装

# Using uv (recommended)
uv sync

# Or with pip
pip install -e .

用法

作为MCP服务器

{
  "mcpServers": {
    "internet-search": {
      "command": "uv",
      "args": ["run", "mcp-duckduckgo-search"],
      "cwd": "/path/to/internet_search"
    }
  }
}

直接执行

uv run mcp-duckduckgo-search

作为图书馆

import asyncio
from mcp_duckduckgo.search import InternetSearch

searcher = InternetSearch()

async def main():
    # General web search with inline content
    response = await searcher.search("Python asyncio", fetch_content=True)
    for result in response.results:
        print(f"{result.title}: {result.url}")
        if result.content:
            print(f"  Content: {result.content[:200]}...")

    # News
    response = await searcher.search("technology", search_type="news", time_filter="w")
    for result in response.results:
        print(f"[{result.date}] {result.source}: {result.title}")

    # Images
    response = await searcher.search("sunset landscape", search_type="images")
    for result in response.results:
        print(f"{result.title}: {result.thumbnail_url}")

    # Fetch page directly
    page = await searcher.fetch_page("https://example.com", max_chars=5000)
    print(f"Title: {page['title']}, Text: {page['text'][:200]}...")

asyncio.run(main())

工具

search_internet

参数 类型 默认值 说明
query str -- 搜索词。最多500个字符。
max_results int 5 结果数量(1-50)。
region str br-br 地理区域。60+个选项。
time_filter str "" d (天), w (周), m (月), y (年), "" (无)。
search_type str text text, news,或 images.
safesearch str moderate off, moderate, strict.
offset int 0 结果偏移(最大1000)。
fetch_content bool False 内联获取整页文本(最多5页,每页8000个字符)。

fetch_page

参数 类型 默认值 说明
url str -- 完整URL(必须以开头 http:// 或 https://).
max_chars int 8000 最大字符数(100-50000)。

search_duckduckgo (已弃用)

传统工具。发射 DeprecationWarning.使用 search_internet 相反。

模式

输出(文本搜索)

{
  "query": "search terms",
  "results": [
    {
      "title": "Page Title",
      "snippet": "Description text...",
      "url": "https://example.com",
      "content": "Full page text...",
      "content_truncated": false
    }
  ],
  "count": 1,
  "error": null,
  "error_type": null,
  "timestamp": "2026-06-17T14:30:00+00:00",
  "truncated": false,
  "search_type": "text",
  "region": "br-br",
  "time_filter": null,
  "safesearch": "moderate",
  "offset": 0,
  "cached": false
}

输出(新闻搜索)

{
  "query": "Iran US war",
  "results": [
    {
      "title": "Iran responds to US peace proposal",
      "snippet": "The Ministry of Foreign Affairs...",
      "url": "https://news.example.com/article",
      "date": "2026-06-17T10:52:00+00:00",
      "source": "Gazeta do Povo",
      "image": "https://news.example.com/img.jpg",
      "content": "Full article text...",
      "content_truncated": false
    }
  ],
  "count": 1,
  "error": null,
  "error_type": null,
  "timestamp": "2026-06-17T14:30:00+00:00",
  "truncated": false,
  "search_type": "news",
  "region": "br-br",
  "time_filter": "w",
  "safesearch": "moderate",
  "offset": 0,
  "cached": false
}

输出(图像搜索)

{
  "query": "sunset landscape",
  "results": [
    {
      "title": "Beautiful Sunset",
      "thumbnail_url": "https://example.com/thumb.jpg",
      "source_url": "https://example.com/image.jpg",
      "width": 1920,
      "height": 1080,
      "source": "Example Photos"
    }
  ],
  "count": 1,
  "error": null,
  "error_type": null,
  "timestamp": "2026-06-17T14:30:00+00:00",
  "truncated": false,
  "search_type": "images",
  "region": "br-br",
  "time_filter": null,
  "safesearch": "moderate",
  "offset": 0,
  "cached": false
}

fetch_page 输出

{
  "url": "https://example.com",
  "title": "Example Domain",
  "text": "This domain is for use in illustrative examples...",
  "status_code": 200,
  "content_type": "text/html",
  "truncated": false,
  "error": null,
  "error_type": null
}

错误响应(相同架构)

{
  "query": "search terms",
  "results": [],
  "count": 0,
  "error": "All backends exhausted",
  "error_type": "all_backends_failed",
  "timestamp": "2026-06-17T14:30:00+00:00",
  "truncated": false,
  "search_type": "text",
  "region": "br-br",
  "time_filter": null,
  "safesearch": "moderate",
  "offset": 0,
  "cached": false
}

错误类型

error_type 含义 代理人行为
null 成功 过程结果
rate_limited 达到后端速率限制 等待并重试
network_error 连接/超时问题 延迟后重试
timeout 请求超时 重试或跳过
http_error 后端返回HTTP错误 日志和回退
invalid_url URL验证失败 修复URL
auth_error 身份验证失败(勇敢) 检查API密钥
parse_error 后端返回了无法解析的数据 日志和回退
validation_error 输入验证失败 修复输入
all_backends_failed 所有配置的后端均失败 不重试;记录并上报
unknown 其他错误 日志和回退

配置

环境变量

变量 默认值 描述
LOG_LEVEL 信息 日志记录级别
CACHE_TTL 300 缓存TTL(秒)
CACHE_MAX_SIZE 512 最大缓存条目数
FETCH_TIMEOUT 10 页面获取超时(秒)
SEARXNG_URL (空) SearXNG实例URL(启用SearXNG后端)
BRAVE_API_KEY (空) Brave Search API密钥(启用Brave后端)

多后端设置

# Enable SearXNG
export SEARXNG_URL="https://searxng.example.com"

# Enable Brave Search
export BRAVE_API_KEY="your-api-key-here"

后端回退顺序:DuckDuckGo→ SearXNG→ 勇敢。

建筑

InternetSearch (orchestrator)
├── Cache (TTLCache, per-param hashing)
├── Metrics (structlog, latency tracking)
├── PageFetcher (httpx + BeautifulSoup)
├── Backend[0]: DuckDuckGoBackend (sync → asyncio.to_thread)
├── Backend[1]: SearXNGBackend (httpx, requires SEARXNG_URL)
└── Backend[2]: BraveSearchBackend (httpx, requires BRAVE_API_KEY)

流动

  1. 验证 --输入已清理和验证(throws ValidationError 立即)
  2. 缓存检查 --所有参数的sha256哈希值;返回缓存 SearchResponse 如果击中
  3. 重试循环 --高达 max_retries 指数回退的尝试
  4. 后端迭代 --按顺序尝试每个配置的后端;跳过不支持的搜索类型
  5. 内容提取 --如果 fetch_content=True,通过以下方式并行获取最多5页 asyncio.gather
  6. 高速缓存组 --返回前已成功存储响应
  7. 后备方案 --如果所有后端都失败,则返回带有final的错误响应 error_type

项目结构

internet_search/
├── src/mcp_duckduckgo/
│   ├── __init__.py
│   ├── models.py         # SearchResult, NewsResult, ImageResult, PageContent, SearchResponse
│   ├── validators.py     # Input validation + SSRF protection via DNS resolution
│   ├── cache.py          # SearchCache (TTLCache, sha256 key, hit rate tracking)
│   ├── fetcher.py        # PageFetcher (httpx AsyncClient, BeautifulSoup, SSRF)
│   ├── metrics.py        # MetricsCollector (structlog, latency timer)
│   ├── search.py         # InternetSearch (orchestrator) + DuckDuckGoSearch alias
│   ├── server.py         # FastMCP server: search_internet, fetch_page, search_duckduckgo
│   └── backends/
│       ├── base.py       # SearchBackend ABC, BackendError, BackendResults
│       ├── duckduckgo.py # DuckDuckGo backend (DDGS wrapper)
│       ├── searxng.py    # SearXNG backend (httpx)
│       └── brave.py      # Brave Search backend (httpx)
├── tests/
│   ├── test_search.py    # InternetSearch orchestration, cache, retry, fallback
│   ├── test_server.py    # MCP tools: search_internet, fetch_page, deprecated alias
│   ├── test_models.py    # SearchResult, NewsResult, ImageResult, PageContent, SearchResponse
│   ├── test_validators.py # Query, max_results, region, time_filter, search_type, safesearch, offset, max_chars, URL (SSRF)
│   ├── test_cache.py     # TTLCache, key generation, hit rate, stats
│   ├── test_fetcher.py   # PageFetcher: success, stripping, truncation, HTTP errors, timeout, network errors, invalid URL, missing title, whitespace normalization
│   ├── test_backends.py  # All backends: protocol, parsing, HTTP mocking, error classification, images
│   ├── test_metrics.py   # MetricsCollector: record, error, sync/async timer
│   └── test_e2e.py       # 14 E2E tests against real DuckDuckGo and example.com
├── pyproject.toml
└── README.md

发展

测试阶段

# 1. Unit tests (models, validators, cache, metrics)
uv run pytest tests/test_models.py tests/test_validators.py tests/test_cache.py tests/test_metrics.py -v

# 2. Integration tests (backends, fetcher, search, server)
uv run pytest tests/test_backends.py tests/test_fetcher.py tests/test_search.py tests/test_server.py -v

# 3. Contract tests (schemas, error types, validation)
uv run pytest tests/ -k "schema or contract or to_dict or error_type or validation or truncated" -v

# 4. Regression (full suite)
uv run pytest tests/ -v

# 5. End-to-end (real DuckDuckGo API + example.com)
$env:RUN_E2E = "1"
uv run pytest tests/test_e2e.py -v

命令

# Install
uv sync

# Full test suite (excludes E2E)
uv run pytest tests/ --cov=mcp_duckduckgo --cov-report=term-missing --cov-fail-under=95

# Type check
uv run mypy src/

# Type check including E2E tests
uv run mypy src/ tests/test_e2e.py

总共187次测试

类别 测试 范围
单元 91 模型、验证器、缓存、指标
集成 82 后端、提取器、搜索、服务器
合约 18 模式、错误类型、输入验证
E2E 14 real DuckDuckGo API,example.com获取
总计 187 95%覆盖率,mypy-严格

许可证

麻省理工学院