🚀 URL 链接抓取 MCP

URL 链接抓取 MCP 是一个干净的 Model Context Protocol (MCP) 实现方案，它能让 Claude 或其他大语言模型（LLM）从 URL 中获取所需内容，为信息获取提供了便捷途径。

🚀 快速开始

安装依赖库

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())

✨ 主要特性

• 可从任意 URL 获取内容。
• 支持多种内容类型，包括 HTML、JSON、文本和图片。
• 能够控制请求参数，如头信息和超时时间。
• 具备清晰的错误处理机制。
• 支持交互式的 CLI 操作。

📦 安装指南

安装依赖库

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

💻 使用示例

基础用法

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())

高级用法

快速测试脚本 (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())

📚 详细文档

目录结构

项目根目录/
├── 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 服务器功能测试

工具功能

fetch_url

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

fetch_image

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

fetch_json

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

测试方法

运行测试用例

python -m pytest tests/

🔧 技术细节

定义工具函数

在 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):
    # 实现细节...

📄 许可证

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

👥 贡献指南

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

📚 参考文档

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

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