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