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

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

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

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

python

运行

from pydantic import BaseModel, Field, AnyUrlfrom typing import Annotatedclass 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, HTTPErrorfrom typing import Tuplefrom mcp.shared.exceptions import McpErrorfrom .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}\") # 智能判断内容类型 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 Protegofrom urllib.parse import urlparse, urlunparseasync 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 markdownifyimport readabilipy.simple_jsondef 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 Serverfrom mcp.types import Tool, TextContentasync 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 工具开发的硬核技术解析！