URL 链接抓取 MCP

一个干净的 Model Context Protocol (MCP) 实现，使 Claude 或任何 LLM 能够从 URL 中获取内容。

功能特性

• 从任意 URL 获取内容
• 支持多种内容类型（HTML、JSON、文本、图片）
• 控制请求参数（头信息、超时时间）
• 清晰的错误处理
• 支持交互式的 CLI 操作

目录结构

项目根目录/
├── mcp_urlfetch/                # 核心代码模块
│   ├── __init__.py             # 包初始化文件
│   ├── client.py               # 客户端 API 实现
│   ├── server.py               # MCP 服务器实现
│   └── tools.py                # 工具函数定义
├── examples/                   # 示例脚本
│   ├── quick_test.py           # 快速测试脚本
│   ├── simple_usage.py         # 简单用法示例
│   └── interactive_client.py  # 交互式 CLI 客户端
└── tests/                      # 测试用例
    ├── direct_test.py          # 直接测试 URL 抓取功能
    └── mcp_server_test.py      # MCP 服务器功能测试

安装说明

安装依赖库

pip install "requests>=2.31.0" "python-multipart>=0.3.5"

使用方法

快速上手

启动 MCP 服务器

python mcp_urlfetch/server.py --port 8000

调用工具函数

在 Python 代码中使用：

from mcp_urlfetch.client import AsyncMCPClient

async def main():
    async with AsyncMCPClient('localhost', 8000) as client:
        # 使用 fetch_url 工具
        result = await client.call_tool("fetch_url", {
            "url": "https://example.com"
        })
        
        # 使用 fetch_json 工具
        result = await client.call_tool("fetch_json", {
            "url": "https://api.example.com/data",
            "headers": {"Authorization": "Bearer token"}
        })
        
        # 使用 fetch_image 工具
        result = await client.call_tool("fetch_image", {
            "url": "https://example.com/image.jpg"
        })

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

核心实现

定义工具函数

在 mcp_urlfetch/tools.py 中定义：

from typing import Dict, Optional
from pydantic import Field
from mcp_core.context import Context
from mcp_core.tool import Tool
from mcp_core.utils import async_func_to_coroutine

@Tool(
    name="fetch_url",
    description="Fetch content from a URL and return it as text."
)
async def fetch_url(url: AnyUrl, headers: Optional[Dict[str, str]] = None, timeout: int = 10, ctx: Context = None):
    # 实现细节...

@Tool(
    name="fetch_image",
    description="Fetch an image from a URL and return it as an image."
)
async def fetch_image(url: AnyUrl, timeout: int = 10, ctx: Context = None):
    # 实现细节...

@Tool(
    name="fetch_json", 
    description="Fetch JSON from a URL, parse it, and return it formatted."
)
async def fetch_json(url: AnyUrl, headers: Optional[Dict[str, str]] = None, timeout: int = 10, ctx: Context = None):
    # 实现细节...

工具功能

fetch_url

从指定 URL 获取内容并返回文本。

参数

• url (必填)：要获取的 URL 地址。
• headers (可选)：附加请求头信息。
• timeout (可选)：设置请求超时时间，单位为秒，默认为 10 秒。

fetch_image

从指定 URL 获取图片并返回图片数据。

参数

• url (必填)：要获取图片的 URL 地址。
• timeout (可选)：设置请求超时时间，单位为秒，默认为 10 秒。

fetch_json

从指定 JSON 接口获取数据，并解析后返回格式化字符串。

参数

• url (必填)：要获取 JSON 数据的 URL 地址。
• headers (可选)：附加请求头信息。
• timeout (可选)：设置请求超时时间，单位为秒，默认为 10 秒。

示例用法

快速测试脚本 (examples/quick_test.py)

from mcp_urlfetch.client import AsyncMCPClient

async def main():
    async with AsyncMCPClient('localhost', 8000) as client:
        # 测试 fetch_url 工具
        result = await client.call_tool("fetch_url", {"url": "https://example.com"})
        print(f"URL 内容: {result}")
        
        # 测试 fetch_json 工具
        result = await client.call_tool("fetch_json", {
            "url": "https://api.example.com/data",
            "headers": {"Authorization": "Bearer token"}
        })
        print(f"JSON 数据: {result}")
        
        # 测试 fetch_image 工具
        result = await client.call_tool("fetch_image", {"url": "https://example.com/image.jpg"})
        print(f"图片数据已获取，大小：{len(result)} bytes")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

测试方法

运行测试用例

python -m pytest tests/

许可证

本项目使用 MIT License 开源协议。

贡献指南

欢迎 Fork 该仓库并提交 Pull Request！

参考文档

• MCP 标准协议
• Python 请求库
• [异步编程指南](https://docs.python.org/3/library asyncio.html)

项目版本: 1.0.0
文档最后更新时间: 2024年3月1日