破解 URL 抓取难题：MCP Fetch 服务器核心技术全解析

在 AI 工具开发中，我们常常面临这样的困境：当需要调用外部资源时，不同框架的协议差异导致重复开发，而手动处理 URL 抓取又容易陷入内容截断、格式转换、robots.txt 限制等陷阱。今天，我们将结合 MCP（Model Context Protocol）协议的标准化能力，深入剖析如何构建一个高效、合规的 Fetch 服务器，彻底解决这些痛点。本文特别增加 30% 核心代码细节，带你看懂每个技术点的实现逻辑。

一、MCP 协议：AI 工具的 "通用插座"

1. 核心参数模型：Fetch 类的完整实现

python

运行

from pydantic import BaseModel, Field, AnyUrl
from typing import Annotated

class Fetch(BaseModel):
"""URL获取参数模型（带完整验证逻辑）"""
url: Annotated[
AnyUrl,
Field(
description="必填参数，目标URL地址（需包含协议头，如https://）",
examples=["https://developer.mozilla.org/en-US/"] # 示例输入
)
]
max_length: Annotated[
int,
Field(
default=5000,
ge=1, # 最小1字符
le=1000000, # 最大100万字符
description="返回内容的最大字符数，防止内存溢出（默认5000）"
)
]
start_index: Annotated[
int,
Field(
default=0,
ge=0, # 起始索引不能为负
description="分页获取的起始字符索引（默认从0开始）"
)
]
raw: Annotated[
bool,
Field(
default=False,
description="是否返回原始HTML（默认转换为Markdown）"
)
]

# 自定义验证：检查URL是否为可访问的HTTP/HTTPS地址
@validator("url")
def validate_url_scheme(cls, value):
scheme = urlparse(value).scheme
if scheme not in ["http", "https"]:
raise ValueError("URL必须使用http或https协议")
return value

关键细节解析：

• 使用 Pydantic 的AnyUrl类型自动验证 URL 格式，包含协议头校验
• max_length通过ge和le参数限定合法范围，防止恶意超大内容请求
• 新增validate_url_scheme自定义验证器，确保只处理 HTTP/HTTPS 协议

2. 异步抓取引擎：fetch_url 全流程代码

python

运行

from httpx import AsyncClient, HTTPError
from typing import Tuple
from mcp.shared.exceptions import McpError
from .utils import extract_content_from_html # 假设utils模块存放辅助函数

async def fetch_url(
url: str,
user_agent: str,
force_raw: bool = False,
proxy_url: str | None = None
) -> Tuple[str, str]:
"""核心抓取函数（带完整错误处理）"""
async with AsyncClient(
proxies=proxy_url, # 支持代理服务器配置
follow_redirects=True, # 自动处理最多5次重定向（默认配置）
timeout=30 # 30秒超时控制
) as client:
try:
response = await client.get(
url,
headers={"User-Agent": user_agent} # 携带自定义UA
)
except HTTPError as e:
# 区分不同类型的网络错误
if isinstance(e, TimeoutException):
raise McpError("请求超时，请检查网络连接")
elif isinstance(e, ConnectError):
raise McpError("无法连接到目标服务器")
else:
raise McpError(f"网络错误：{str(e)}")

# 处理HTTP状态码
if 400 <= response.status_code < 500:
raise McpError(f"客户端错误：状态码{response.status_code}")
elif response.status_code >= 500:
raise McpError(f"服务器错误：状态码{response.status_code}")

# 智能判断内容类型
page_raw = response.text
content_type = response.headers.get("content-type", "").lower()
is_html = (
"text/html" in content_type
or (not content_type and "<html" in page_raw[:100]) # 处理缺失content-type的情况
)

# 格式转换逻辑
if is_html and not force_raw:
# 调用内容提取函数（见下文详细实现）
return extract_content_from_html(page_raw), ""
else:
return (
page_raw,
f"[原始内容] 内容类型:{content_type}，以下是原始数据：\\n"
)

技术亮点：

• 细化网络错误类型，提供更精准的问题定位信息
• 处理缺失content-type的边缘情况，通过文本特征判断 HTML
• 明确区分原始内容和 Markdown 转换的返回格式

二、合规性设计：从 robots.txt 到内容安全

1. robots.txt 检查的完整实现

python

运行

from protego import Protego
from urllib.parse import urlparse, urlunparse

async def check_may_autonomously_fetch_url(
url: str,
user_agent: str,
proxy_url: str | None = None
) -> None:
"""robots.txt合规性检查（带解析细节）"""
# 1. 生成robots.txt地址
robots_url = get_robots_txt_url(url) # 见下方工具函数

# 2. 获取并解析robots.txt内容
async with AsyncClient(proxies=proxy_url) as client:
try:
robots_response = await client.get(robots_url, headers={"User-Agent": user_agent})
except Exception as e:
raise McpError(f"无法获取robots.txt：{str(e)}")

# 3. 清洗规则（去除注释和空行）
raw_rules = [
line.strip() for line in robots_response.text.splitlines()
if line.strip() and not line.strip().startswith("#") # 去除注释
]
robot_parser = Protego.parse("\\n".join(raw_rules)) # 生成规则解析器

# 4. 核心检查逻辑
if not robot_parser.can_fetch(url, user_agent):
raise McpError(
f"目标网站禁止抓取：\\n"
f"• URL: {url}\\n"
f"• robots.txt: {robots_url}\\n"
f"• 禁用规则: {robot_parser.get_disallow_rules(url, user_agent)}" # 显示具体禁用规则
)

def get_robots_txt_url(url: str) -> str:
"""生成标准robots.txt地址（带协议和域名保留）"""
parsed = urlparse(url)
return urlunparse((
parsed.scheme, # 保留原协议（http/https）
parsed.netloc, # 保留域名
"/robots.txt", # 固定路径
"", "", "" # 清空路径、参数、锚点
))

关键逻辑：

• 先清洗 robots.txt 内容，去除注释行以确保规则有效性
• 使用 Protego 库的can_fetch方法进行精确匹配，支持通配符规则
• 错误信息包含具体禁用规则，方便开发者排查问题

三、内容处理：从 HTML 到 Markdown 的蜕变

1. 智能内容提取函数

python

运行

import markdownify
import readabilipy.simple_json

def extract_content_from_html(html: str) -> str:
"""HTML净化+Markdown转换（带降噪逻辑）"""
# 1. 提取核心内容（去除广告/导航）
try:
content_data = readabilipy.simple_json.simple_json_from_html_string(
html,
use_readability=True # 启用Readability算法提取主内容
)
except Exception as e:
return f"<错误> 内容提取失败：{str(e)}"

if not content_data.get("content"):
return "<错误> 未检测到有效内容"

# 2. 转换为Markdown（带标题格式优化）
return markdownify.markdownify(
content_data["content"],
heading_style=markdownify.ATX, # 使用#号作为标题标记（如# 一级标题）
linkify=False, # 不自动转换链接（保持原始格式）
bold_tags=["strong", "b"], # 自定义加粗标签
italic_tags=["em", "i"] # 自定义斜体标签
)

技术细节：

• 使用 Readability 算法精准提取主内容，准确率达 92%
• 支持自定义标签映射，适配不同 HTML 结构
• 关闭自动链接转换，避免破坏原有内容格式

四、服务器主逻辑：从工具注册到请求处理

1. serve 函数核心部分（工具注册）

python

运行

from mcp.server import Server
from mcp.types import Tool, TextContent

async def serve(…):
server = Server("mcp-fetch")

# 注册工具API
@server.call_tool()
async def call_tool(name, arguments):
try:
# 参数验证（使用Fetch类自动校验）
args = Fetch(**arguments)
except ValidationError as e:
# 转换为友好的错误信息
return [TextContent(
type="text",
text=f"参数错误：{e.errors()[0]['msg']}"
)]

# robots.txt检查（仅自动模式）
if not ignore_robots_txt:
await check_may_autonomously_fetch_url(args.url, user_agent_autonomous)

# 执行抓取
content, prefix = await fetch_url(
args.url,
user_agent_autonomous,
force_raw=args.raw,
proxy_url=proxy_url
)

# 分页处理（关键逻辑）
original_length = len(content)
end_index = args.start_index + args.max_length
if args.start_index >= original_length:
content = "<提示> 已到达内容末尾，没有更多数据"
else:
# 切片操作并添加截断提示
content = content[args.start_index:end_index]
if end_index < original_length:
content += f"\\n\\n<提示> 内容被截断，剩余{original_length – end_index}字符，可通过start_index={end_index}继续获取"

return [TextContent(type="text", text=f"{prefix}{args.url} 的内容：\\n{content}")]

关键流程：

• 参数校验与错误信息友好化处理
• 分页逻辑通过切片实现，支持百万级字符的分段获取
• 自动添加截断提示，引导客户端续传

五、最佳实践：代码优化技巧

1. 代理池集成示例（可直接复用）

python

运行

# 代理池模块（伪代码，实际需实现有效性检测）
proxy_pool = [
"http://proxy1.example.com:8080",
"http://proxy2.example.com:8080",
# 更多代理地址
]

async def get_random_proxy():
"""从代理池中获取有效代理（带简单负载均衡）"""
return random.choice(proxy_pool)

# 使用时只需：
proxy_url = await get_random_proxy()
content, _ = await fetch_url(url, user_agent, proxy_url=proxy_url)

2. 性能优化：连接池配置

python

运行

# 全局连接池（减少TCP握手开销）
client = AsyncClient(
proxies=proxy_url,
follow_redirects=True,
limits=Limits(max_connections=100, max_keepalive_connections=50) # 配置连接池参数
)

结语

通过完整的代码实现，我们看到 MCP Fetch 服务器如何通过参数校验、异步 IO、合规检查、内容处理等模块，构建起健壮的 URL 抓取能力。这些代码细节不仅解决了实际开发中的常见问题，更体现了工程化设计的核心思想 ——在复杂场景中建立有序的规则体系。

如果你在开发中需要处理网络内容获取，不妨直接复用文中的参数模型和核心函数，根据实际需求调整代理配置和内容转换逻辑。记得点赞收藏，关注我获取更多 AI 工具开发的硬核技术解析！