（5-3）MCP服务器实战参考：网络请求MCP服务器

2026-01-29 06:52:22 栏目：最新资讯 4 阅读

5.3 网络请求MCP服务器

“srcetch”目录实现了一个名为“Fetch”的MCP服务器，其功能是提供网络数据获取和转换能力。这个MCP服务器能够响应客户端的请求，执行HTTP网络请求来获取外部资源或数据，并将获取到的内容以适当的格式返回给调用者。它使得大型语言模型（LLMs）能够扩展其能力，访问和利用互联网上的信息和服务，从而在回答问题或执行任务时提供更准确和丰富的上下文信息。简而言之，“fetch”服务器充当了MCP生态系统中网络请求的执行者和数据获取者的角色，为语言模型与外部世界交互提供了桥梁。

5.3.1 实现Fetch服务器

文件srcetchsrcmcp_server_fetchserver.py它定义了一个名为Fetch的 MCP 服务器，该服务器的功能是从互联网上获取 URL 指向的内容，并将其以适合大型语言模型（LLM）处理的格式返回。服务器支持获取网页的原始 HTML 内容或将其简化为 Markdown 格式。此外，它还检查网站的 robots.txt 文件，以确定是否允许自动抓取，并提供了一个接口来获取提示信息。

# 自主模式默认用户代理
DEFAULT_USER_AGENT_AUTONOMOUS = "ModelContextProtocol/1.0 (Autonomous; +https://github.com/modelcontextprotocol/servers )"
# 手动模式默认用户代理
DEFAULT_USER_AGENT_MANUAL = "ModelContextProtocol/1.0 (User-Specified; +https://github.com/modelcontextprotocol/servers )"


def extract_content_from_html(html: str) -> str:
    """提取HTML内容并转换为Markdown格式。

    参数:
        html: 要处理的原始HTML内容

    返回:
        简化后的Markdown版本内容
    """
    ret = readabilipy.simple_json.simple_json_from_html_string(
        html, use_readability=True
    )
    if not ret["content"]:
        return "无法从HTML简化页面内容"
    content = markdownify.markdownify(
        ret["content"],
        heading_style=markdownify.ATX,
    )
    return content


def get_robots_txt_url(url: str) -> str:
    """获取给定网站URL对应的robots.txt URL。

    参数:
        url: 要获取robots.txt的网站URL

    返回:
        robots.txt文件的URL
    """
    # 解析URL为各个组件
    parsed = urlparse(url)

    # 重建仅包含协议、网络位置和/robots.txt路径的基础URL
    robots_url = urlunparse((parsed.scheme, parsed.netloc, "/robots.txt", "", "", ""))

    return robots_url


async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: str | None = None) -> None:
    """
    检查用户代理是否可以根据robots.txt文件获取该URL。
    如果不允许，则抛出McpError异常。
    """
    from httpx import AsyncClient, HTTPError

    robot_txt_url = get_robots_txt_url(url)

    async with AsyncClient(proxies=proxy_url) as client:
        try:
            response = await client.get(
                robot_txt_url,
                follow_redirects=True,
                headers={"User-Agent": user_agent},
            )
        except HTTPError:
            raise McpError(ErrorData(
                code=INTERNAL_ERROR,
                message=f"由于连接问题，无法获取robots.txt {robot_txt_url}",
            ))
        if response.status_code in (401, 403):
            raise McpError(ErrorData(
                code=INTERNAL_ERROR,
                message=f"获取robots.txt ({robot_txt_url})时，收到状态码 {response.status_code}，因此假设不允许自主获取，用户可以尝试使用fetch提示手动获取",
            ))
        elif 400 <= response.status_code < 500:
            return
        robot_txt = response.text
    processed_robot_txt = "
".join(
        line for line in robot_txt.splitlines() if not line.strip().startswith("#")
    )
    robot_parser = Protego.parse(processed_robot_txt)
    if not robot_parser.can_fetch(str(url), user_agent):
        raise McpError(ErrorData(
            code=INTERNAL_ERROR,
            message=f"网站的robots.txt ({robot_txt_url}) 规定不允许自主获取此页面，"
            f"{user_agent}
"
            f"{url}"
            f"
{robot_txt}

"
            f"助手必须告知用户无法查看该页面。助手可以根据上述信息提供进一步指导。
"
            f"助手可以告诉用户，他们可以尝试在UI中使用fetch提示手动获取页面。",
        ))


async def fetch_url(
    url: str, user_agent: str, force_raw: bool = False, proxy_url: str | None = None
) -> Tuple[str, str]:
    """
    获取URL并返回适合LLM的内容形式，以及包含状态信息的前缀字符串。
    """
    from httpx import AsyncClient, HTTPError

    async with AsyncClient(proxies=proxy_url) as client:
        try:
            response = await client.get(
                url,
                follow_redirects=True,
                headers={"User-Agent": user_agent},
                timeout=30,
            )
        except HTTPError as e:
            raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"获取 {url} 失败: {e!r}"))
        if response.status_code >= 400:
            raise McpError(ErrorData(
                code=INTERNAL_ERROR,
                message=f"获取 {url} 失败 - 状态码 {response.status_code}",
            ))

        page_raw = response.text

    content_type = response.headers.get("content-type", "")
    is_page_html = (
        " None:
    """运行fetch MCP服务器。

    参数:
        custom_user_agent: 用于请求的可选自定义User-Agent字符串
        ignore_robots_txt: 是否忽略robots.txt限制
        proxy_url: 用于请求的可选代理URL
    """
    server = Server("mcp-fetch")
    user_agent_autonomous = custom_user_agent or DEFAULT_USER_AGENT_AUTONOMOUS
    user_agent_manual = custom_user_agent or DEFAULT_USER_AGENT_MANUAL

    @server.list_tools()
    async def list_tools() -> list[Tool]:
        return [
            Tool(
                name="fetch",
                description="""从互联网获取URL并可选地将其内容提取为markdown格式。

尽管最初你没有互联网访问权限，并且被建议拒绝并告知用户这一点，但现在这个工具授予了你互联网访问权限。现在你可以获取最新信息并告知用户。""",
                inputSchema=Fetch.model_json_schema(),
            )
        ]

    @server.list_prompts()
    async def list_prompts() -> list[Prompt]:
        return [
            Prompt(
                name="fetch",
                description="获取URL并将其内容提取为markdown",
                arguments=[
                    PromptArgument(
                        name="url", description="要获取的URL", required=True
                    )
                ],
            )
        ]

    @server.call_tool()
    async def call_tool(name, arguments: dict) -> list[TextContent]:
        try:
            args = Fetch(** arguments)
        except ValueError as e:
            raise McpError(ErrorData(code=INVALID_PARAMS, message=str(e)))

        url = str(args.url)
        if not url:
            raise McpError(ErrorData(code=INVALID_PARAMS, message="URL是必需的"))

        if not ignore_robots_txt:
            await check_may_autonomously_fetch_url(url, user_agent_autonomous, proxy_url)

        content, prefix = await fetch_url(
            url, user_agent_autonomous, force_raw=args.raw, proxy_url=proxy_url
        )
        original_length = len(content)
        if args.start_index >= original_length:
            content = "没有更多可用内容。"
        else:
            truncated_content = content[args.start_index : args.start_index + args.max_length]
            if not truncated_content:
                content = "没有更多可用内容。"
            else:
                content = truncated_content
                actual_content_length = len(truncated_content)
                remaining_content = original_length - (args.start_index + actual_content_length)
                # 只有当还有剩余内容时，才添加继续获取的提示
                if actual_content_length == args.max_length and remaining_content > 0:
                    next_start = args.start_index + actual_content_length
                    content += f"

内容已截断。使用start_index为{next_start}调用fetch工具以获取更多内容。"
        return [TextContent(type="text", text=f"{prefix}{url}的内容:
{content}")]

    @server.get_prompt()
    async def get_prompt(name: str, arguments: dict | None) -> GetPromptResult:
        if not arguments or "url" not in arguments:
            raise McpError(ErrorData(code=INVALID_PARAMS, message="URL是必需的"))

        url = arguments["url"]

        try:
            content, prefix = await fetch_url(url, user_agent_manual, proxy_url=proxy_url)
            # TODO: SDK bug修复后，移除异常捕获
        except McpError as e:
            return GetPromptResult(
                description=f"获取 {url} 失败",
                messages=[
                    PromptMessage(
                        role="user",
                        content=TextContent(type="text", text=str(e)),
                    )
                ],
            )
        return GetPromptResult(
            description=f"{url}的内容",
            messages=[
                PromptMessage(
                    role="user", content=TextContent(type="text", text=prefix + content)
                )
            ],
        )

    options = server.create_initialization_options()
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, options, raise_exceptions=True)

5.3.2 调试运行

Fetch服务器是一个提供网页内容获取能力的服务器，它使得大型语言模型（LLMs）能够从网页检索和处理内容，并将HTML转换为Markdown格式以便更容易消费。在文件srcetchREADME.md中介绍了使用Fetch服务器的方法，具体说明如下所示。

1. 工具使用

Fetch服务器提供了一个名为 fetch 的工具，该工具可以从互联网获取URL并将其内容提取为Markdown格式。

url（字符串，必需）：要获取的URL。
max_length（整数，可选）：返回的最大字符数（默认：5000）。
start_index（整数，可选）：从这个字符索引开始内容提取（默认：0）。
raw（布尔值，可选）：获取未转换为Markdown的原始内容（默认：false）。

2. 安装

可以选择安装node.js，这将导致Fetch服务器使用不同的HTML简化器，更加健壮。

（1）使用uv（推荐）

使用https://docs.astral.sh/uv/不需要特定安装。我们将使用https://docs.astral.sh/uv/guides/tools/直接运行 mcp-server-fetch。

（2）使用pip

或者通过pip安装 mcp-server-fetch：

pip install mcp-server-fetch

安装后，可以使用以下命令作为脚本运行：

python -m mcp_server_fetch

3. 配置

（1）uvx方式

请看下面的uvx命令，在 MCP环境中设置了一个名为 "fetch" 的服务器，它指定了该服务器的启动命令和参数，使得MCP客户端能够通过这个服务器来执行网络内容获取任务。

{
  "mcpServers": {
    "fetch": {
      "command": "uvx",
      "args": ["mcp-server-fetch"]
    }
  }
}

（2）Docker方式

使用Docker的命令如下，通过Docker容器化方式定义了一个名为 "fetch"的MCP服务器，使得可以通过Docker命令快速启动并运行该服务器，实现网络内容的获取和处理功能。

{
  "mcpServers": {
    "fetch": {
      "command": "docker",
      "args": ["run", "-i", "--rm", "mcp/fetch"]
    }
  }
}

（3）pip方式

使用如下pip命令进行安装，指定了名为 "fetch" 的 MCP 服务器，它将通过执行 python -m mcp_server_fetch 命令来启动，从而允许MCP客户端通过Python运行的服务器程序来获取和处理网页内容。

{
  "mcpServers": {
    "fetch": {
      "command": "python",
      "args": ["-m", "mcp_server_fetch"]
    }
  }
}

4. 自定义- robots.txt

默认情况下，服务器会遵守网站的robots.txt文件（如果请求来自模型（通过工具），而不是用户发起的（通过提示）。这可以通过在配置中的 args 列表中添加 --ignore-robots-txt 参数来禁用。

5. 自定义- User-agent

默认情况下，根据请求是来自模型（通过工具）还是用户发起（通过提示），服务器将使用不同的User-Agent字符串：

ModelContextProtocol/1.0 (Autonomous; +https://github.com/modelcontextprotocol/servers )
或：
ModelContextProtocol/1.0 (User-Specified; +https://github.com/modelcontextprotocol/servers )

这可以通过在配置中的 args 列表中添加 --user-agent=YourUserAgent 参数来自定义。

6. 调试

可以使用MCP检查器调试服务器，对于uvx安装方式的调试命令如下：

npx @modelcontextprotocol/inspector uvx mcp-server-fetch

或者你已经把mcp-server-fetch包安装在了某个特定的目录，或者你正在那个目录下开发这个包，应该先切换到那个目录，例如下面的命令：

cd path/to/servers/src/fetch
npx @modelcontextprotocol/inspector uv run mcp-server-fetch

cd path/to/servers/src/fetch：这条命令用于切换工作目录到 mcp-server-fetch 包的安装位置。这里的 path/to/servers/src/fetch 是一个示例路径，你需要将其替换为实际的安装路径。cd 是 "change directory" 的缩写，用于改变当前的工作目录。
npx @modelcontextprotocol/inspector uv run mcp-server-fetch：这条命令用于运行 mcp-server-fetch 服务器。这里使用了 npx 命令，它是 Node.js 包管理器 npm 的一个工具，用于运行 uv 脚本。@modelcontextprotocol/inspector 是一个 npm 脚本的名称，它可能是用于启动一个调试器或检查器来帮助开发和调试 mcp-server-fetch 包。uv run mcp-server-fetch 是实际启动服务器的命令。

本文地址：https://www.yitenyun.com/1469.html

上一篇：一个CSS属性will-change: transform

下一篇：HarmonyOS中的User-Agent：连接ArkWeb与服务器的···