pyfetcher (Pr1m8)

费奇基特

用于自主获取、抓取和内容获取的代理网络基础设施。

赋予AI代理在网络上获取、抓取、提取和下载任何内容的能力——具有逼真的浏览器指纹、结构化输出以及由Postgres和MinIO支持的完整抓取-抓取下载管道。

        

______________________________________________________________________

为什么用fetchkit？

问题：人工智能代理需要与网络交互——获取页面、提取数据、下载文件——但现有的工具不是为自主操作而设计的。它们缺乏结构化的输出、逼真的浏览器指纹和流水线编排。

fetchkit解决了这个问题 通过提供：

1. MCP服务器 --任何AI代理（Claude、LangChain、LangGraph）都可以直接调用的16个工具。结构化的Pydantic输出，而不是原始HTML。
2. 真实的浏览器身份 --11个配置文件，具有一致的UA+客户端提示+安全获取-\*标头。通过curl_cffi进行TLS指纹识别。Cloudflare旁路。
3. 完整管道 --事件驱动的抓取->抓取->下载，由Postgres作业队列和MinIO对象存储支持。
4. 深度下载器集成 --yt-dlp和gallery-dl Python API，带有进度挂钩和元数据提取功能。

pip install 'fetchkit[mcp]'     # AI agent integration
pip install 'fetchkit[full]'    # Everything

亮点

pip install fetchkit                   # Core: fetch, scrape, headers
pip install 'fetchkit[mcp]'            # + MCP server for AI agents
pip install 'fetchkit[pipeline]'       # + Postgres job queue + MinIO storage
pip install 'fetchkit[full]'           # Everything including yt-dlp, Playwright, etc.

使用逼真的浏览器标题进行获取

from pyfetcher import fetch

response = fetch("https://example.com")
print(response.status_code, response.ok)
# Sends Chrome-like headers with Client Hints, Sec-Fetch-*, UA rotation automatically

刮掉任何东西

from pyfetcher.scrape import extract_links, extract_text, extract_readable_text

links = extract_links(html, base_url="https://example.com")  # all links with internal/external tags
titles = extract_text(html, "h1")                             # CSS selector extraction
article = extract_readable_text(html)                         # strips scripts, nav, ads

4个HTTP后端——为作业选择合适的后端

from pyfetcher import FetchRequest, fetch

response = fetch("https://example.com")                                        # httpx (default, HTTP/2)
response = fetch(FetchRequest(url="https://example.com", backend="aiohttp"))   # aiohttp (pure async)
response = fetch(FetchRequest(url="https://example.com", backend="curl_cffi")) # TLS fingerprinting
response = fetch(FetchRequest(url="https://example.com", backend="cloudscraper")) # Cloudflare bypass

使用yt-dlp和gallery dl下载媒体

from pyfetcher.downloaders.ytdlp import YtdlpDownloader
from pyfetcher.downloaders.gallerydl import GalleryDlDownloader

# Video/audio with progress tracking
yt = YtdlpDownloader()
info = await yt.extract_info("https://youtube.com/watch?v=...")    # metadata only
results = await yt.download("https://youtube.com/watch?v=...",     # full download
    output_dir="./media", progress_callback=lambda p: print(p.status))

# Image galleries (170+ supported sites)
gdl = GalleryDlDownloader()
results = await gdl.download("https://imgur.com/gallery/...", output_dir="./images")

MCP服务器——赋予AI代理网络超能力

pyfetcher-mcp                  # stdio for Claude Desktop / Claude Code
pyfetcher-mcp --http 8000      # HTTP for LangChain / remote agents

# LangChain integration
from langchain_mcp_adapters import MultiServerMCPClient
client = MultiServerMCPClient({"pyfetcher": {"transport": "http", "url": "http://localhost:8000/mcp"}})
tools = await client.get_tools()  # 16 structured tools ready for any agent

特性

核心库

基础设施（可选附加）

安装

pip install fetchkit

所有可选附加功能：

pip install 'fetchkit[tui]'            # Textual TUI
pip install 'fetchkit[curl]'           # curl_cffi TLS fingerprinting
pip install 'fetchkit[cloudscraper]'   # Cloudflare bypass
pip install 'fetchkit[db]'             # Postgres + SQLAlchemy + Alembic
pip install 'fetchkit[store]'          # MinIO/S3 object storage
pip install 'fetchkit[pipeline]'       # db + store (full pipeline)
pip install 'fetchkit[downloaders]'    # yt-dlp + gallery-dl
pip install 'fetchkit[extractors]'     # trafilatura, readability, html2text
pip install 'fetchkit[media]'          # Audio/video/image/PDF metadata
pip install 'fetchkit[browser]'        # Playwright + stealth
pip install 'fetchkit[feeds]'          # RSS/Atom feed parsing
pip install 'fetchkit[full]'           # Everything

快速开始

获取

from pyfetcher import fetch, afetch, FetchRequest
import asyncio

# Sync
response = fetch("https://example.com")
print(response.status_code, response.ok)

# Async
response = asyncio.run(afetch("https://example.com"))

浏览器配置文件和标题

from pyfetcher.headers.browser import BrowserHeaderProvider
from pyfetcher.headers.rotating import RotatingHeaderProvider
from pyfetcher.headers.ua import random_user_agent
from pyfetcher.fetch.service import FetchService

# Fixed profile (Chrome on Windows)
service = FetchService(header_provider=BrowserHeaderProvider("chrome_win"))

# Rotating profiles weighted by real-world market share
service = FetchService(header_provider=RotatingHeaderProvider())

# Just need a user-agent string?
ua = random_user_agent(browser="firefox", platform="macOS")

抓取

from pyfetcher.scrape import (
    extract_links, extract_text, extract_table,
    extract_forms, extract_readable_text,
)
from pyfetcher.scrape.robots import parse_robots_txt, is_allowed

# CSS selectors
titles = extract_text(html, "h1.title")
rows = extract_table(html, "table.data")

# Links with internal/external classification
links = extract_links(html, base_url=url, same_domain_only=True)

# Forms with field extraction
forms = extract_forms(html, base_url=url)
print(forms[0].action, forms[0].to_dict())

# Robots.txt
rules = parse_robots_txt(robots_content)
allowed = is_allowed(rules, "/admin", user_agent="MyBot")

价格限制取件

from pyfetcher.fetch.service import FetchService
from pyfetcher.ratelimit.limiter import DomainRateLimiter, RateLimitPolicy

limiter = DomainRateLimiter(
    default_policy=RateLimitPolicy(requests_per_second=2.0, burst=5),
    domain_policies={
        "api.example.com": RateLimitPolicy(requests_per_second=0.5),
    },
)
service = FetchService(rate_limiter=limiter)

内容提取

from pyfetcher.extractors.content import extract_article_text
from pyfetcher.extractors.convert import html_to_markdown, html_to_plaintext

# Article text (trafilatura with readability-lxml fallback)
article = extract_article_text(html, url="https://example.com/post")

# HTML -> Markdown
md = html_to_markdown(html)

yt-dlp&画廊-dl

from pyfetcher.downloaders.ytdlp import YtdlpDownloader
from pyfetcher.downloaders.gallerydl import GalleryDlDownloader

# yt-dlp with progress tracking
yt = YtdlpDownloader()
info = await yt.extract_info("https://youtube.com/watch?v=dQw4w9WgXcQ")
results = await yt.download(url, output_dir="./videos",
    progress_callback=lambda p: print(f"{p.status}: {p.percent}"))

# gallery-dl for image galleries (170+ supported sites)
gdl = GalleryDlDownloader()
results = await gdl.download("https://imgur.com/gallery/...", output_dir="./images")

命令行界面

# Fetch with any backend
pyfetcher fetch https://example.com
pyfetcher fetch https://example.com -o json -b curl_cffi

# Preview generated headers
pyfetcher headers --profile chrome_win
pyfetcher headers --browser firefox -o json
pyfetcher headers --list

# Scrape content
pyfetcher scrape https://example.com --css "h1"
pyfetcher scrape https://example.com --links -o json
pyfetcher scrape https://example.com --text
pyfetcher scrape https://example.com --meta

# Random user-agents
pyfetcher user-agent --browser chrome --count 5
pyfetcher user-agent --mobile

# Check robots.txt
pyfetcher robots https://example.com -p /admin

# Download files
pyfetcher download https://example.com/file.pdf ./file.pdf

管道

事件驱动的管道通过Postgres LISTEN/NOTIFY连接三个阶段：

Seeds / RSS / Sitemap
       |
  [Crawl Stage]  ──NOTIFY──>  [Scrape Stage]  ──NOTIFY──>  [Download Stage]
       |                             |                             |
       v                             v                             v
  pages table                 pages (enriched)              media_assets
  + new crawl jobs            + download jobs               + MinIO objects

设置

make infra-up     # Start Postgres + MinIO
make migrate      # Run Alembic migrations
make pipeline     # Start all workers

程序化

from pyfetcher.pipeline.runner import PipelineRunner
from pyfetcher.config import PyfetcherConfig

runner = PipelineRunner(PyfetcherConfig(
    crawl_concurrency=10,
    scrape_concurrency=20,
    download_concurrency=5,
))
await runner.start()

定制蜘蛛

from pyfetcher.crawler.spider import Spider, SpiderResult

spider = Spider(name="my-spider")

@spider.router.add(r"/blog/\d{4}/")
async def handle_post(url, response):
    return SpiderResult(
        discovered_urls=[...],
        items=[{"title": "...", "content": "..."}],
    )

MCP服务器（AI代理集成）

fetchkit船作为 MCP服务器，使其所有功能可供AI代理（Claude、LangChain、LangGraph和任何兼容MCP的客户端）使用。这将fetchkit变成 自主代理基础设施 --LLM可以获取、抓取、提取和下载，而无需自定义代码。

为什么选择MCP？

传统的抓取需要为每个站点编写代码。使用fetchkit的MCP服务器，AI代理可以：

• 自主研究课题 通过获取页面、提取内容和跟踪链接
• 审计网站 通过检查元数据、robots.txt、站点地图和页面结构
• 提取结构化数据 使用CSS选择器、表解析或文章提取从任何页面
• 下载媒体 具有进度跟踪和校验和验证功能
• 生成现实的请求 使用通过bot检测的浏览器配置文件

全部16个工具返回 结构化Pydantic模型 所以LLM得到的是干净的、类型化的数据，而不是原始的HTML。

快速开始

pip install 'fetchkit[mcp]'

# Run as stdio server (Claude Desktop / Claude Code)
pyfetcher-mcp

# Run as HTTP server (LangChain / remote agents)
pyfetcher-mcp --http 8000

# Or via Makefile
make mcp          # stdio
make mcp-http     # HTTP on port 8000

可用工具（16）

资源和提示

资源为上下文公开数据： pyfetcher://profiles, pyfetcher://backends, pyfetcher://version.

提示提供模板： web_research, site_audit, scrape_guide, compare_pages.

与LangChain一起使用

from langchain_mcp_adapters import MultiServerMCPClient

client = MultiServerMCPClient({
    "pyfetcher": {"transport": "http", "url": "http://localhost:8000/mcp"}
})
tools = await client.get_tools()  # 16 LangChain tools ready to use

# Build an agent
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(model, tools)

与Claude Desktop一起使用

添加 claude_desktop_config.json:

{
  "mcpServers": {
    "pyfetcher": {
      "command": "pyfetcher-mcp",
      "args": []
    }
  }
}

传输后端

|后端|同步|异步|流| TLS指纹| CF旁路|安装| | ------------ | :--: | :---: | :----: | :-------------: | :-------: | ---------------- | |httpx|Y|Y|Y-|-|-| _（核心）_ | |aiohttp|-|Y|Y|-|-| _（核心）_ | |curl_cffi|Y|Y|Y|Y|-| [curl] | |刮云器|Y|-|-|/|Y| [cloudscraper] |

发展

git clone https://github.com/pr1m8/pyfetcher.git
cd pyfetcher
make install-all              # pdm install with all deps
make test                     # 358 tests
make check                    # format + lint + test
make infra-up && make migrate # start Postgres + MinIO

生成文件目标

make help          Show all targets
make install-all   Install everything
make test          Run 358 tests
make test-cov      Tests with coverage report
make fmt           Format with trunk
make lint          Lint with trunk
make check         Format + lint + test
make infra-up      Start Postgres + MinIO
make infra-down    Stop infrastructure
make migrate       Run Alembic migrations
make pipeline      Run crawl->scrape->download
make build         Build wheel + sdist
make publish       Publish to PyPI
make docs          Build Sphinx docs
make clean         Remove build artifacts

文档

****

许可证

麻省理工学院