crawl4ai 使用记录
文档Crawler Result - Crawl4AI Documentation (v0.7.x)
部署
pip install -U crawl4ai
pip install nest_asyncio安装浏览器内核
playwright install补全缺少的库(ubuntu)
sudo apt update && sudo apt install -y \
libicu66 \
libxml2 \
libwebp6 \
libffi7arch
yay -S icu66 libxml2-2.9 libwebp6 libffi7修改有头模式
browser_conf = BrowserConfig(headless=False)
async with AsyncWebCrawler(config=browser_conf) as crawler:获取requset_header，需要配置
config = CrawlerRunConfig(
capture_network_requests=True,
capture_console_messages=True
)区分api_call
api_calls = [r for r in result.network_requests
if r.get("event_type") == "request" and "api" in r.get("url", "")]
if api_calls:
print(f"Detected {len(api_calls)} API calls:")
for call in api_calls[:3]: # Show first 3
print(f" - {call.get('method')} {call.get('url')}")获取response_headers，不需要配置
print(f"{result.response_headers}")拿到的是当前页面的response
import asyncio
import json
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig,BrowserConfig
async def main():
browser_conf = BrowserConfig(headless=False)
# Enable both network request capture and console message capture
config = CrawlerRunConfig(
capture_network_requests=True,
capture_console_messages=True
)
async with AsyncWebCrawler(config=browser_conf) as crawler:
result = await crawler.arun(
url="https://bing.com",
config=config
)
if result.success:
# Analyze network requests
if result.network_requests:
print(f"Captured {len(result.network_requests)} network events")
# Count request types
request_count = len([r for r in result.network_requests if r.get("event_type") == "request"])
response_count = len([r for r in result.network_requests if r.get("event_type") == "response"])
failed_count = len([r for r in result.network_requests if r.get("event_type") == "request_failed"])
print(f"Requests: {request_count}, Responses: {response_count}, Failed: {failed_count}")
# Find API calls
api_calls = [r for r in result.network_requests
if r.get("event_type") == "request" and "api" in r.get("url", "")]
if api_calls:
print(f"Detected {len(api_calls)} API calls:")
for call in api_calls[:3]: # Show first 3
print(f" - {call.get('method')} {call.get('url')}")
# Analyze console messages
if result.console_messages:
print(f"Captured {len(result.console_messages)} console messages")
# Group by type
message_types = {}
for msg in result.console_messages:
msg_type = msg.get("type", "unknown")
message_types[msg_type] = message_types.get(msg_type, 0) + 1
print("Message types:", message_types)
# Show errors (often the most important)
errors = [msg for msg in result.console_messages if msg.get("type") == "error"]
if errors:
print(f"Found {len(errors)} console errors:")
for err in errors[:2]: # Show first 2
print(f" - {err.get('text', '')[:100]}")
# Export all captured data to a file for detailed analysis
with open("network_capture.json", "w") as f:
json.dump({
"url": result.url,
"network_requests": result.network_requests or [],
"console_messages": result.console_messages or []
}, f, indent=2)
print("Exported detailed capture data to network_capture.json")
if __name__ == "__main__":
asyncio.run(main())
![[Pasted image 20251015183928.png]] 能成功打开网页并导航到网站并输出请求体 ![[Pasted image 20251015184116.png]] LLM配置用于过滤内容生成md，如果需要LLM指导爬虫，可以配置LLM，但我们需要的是api和路径，可以选择不生成md。
编写mcp工具
由于爬虫花的时间较长，设置了两个工具避免超时，下面是效果图，先使用start_crawl创建爬虫任务，然后使用get_crawl_resul查询执行结果
![[Pasted image 20251016115935.png]]
mcp = FastMCP()
# ---------- 内存任务存储 ----------
TASK_TTL = timedelta(minutes=10) # 10 分钟自动清理过期任务
_task_store: Dict[str, Dict] = {} # {task_id: {...}}
_gc_task = None # 全局变量存储垃圾回收任务
# ---------- 后台爬取逻辑 ----------
async def _crawl_task(task_id: str, url: str) -> None:
"""
在后台执行真正的 crawl4ai 抓取，完成后写入 task_store。
任何异常都会被捕获并标记为 failed。
"""
try:
browser_conf = BrowserConfig(
headless=False,
)
run_conf = CrawlerRunConfig(
capture_network_requests=True,
# wait_for="domcontentloaded", # 不等所有图片加载完
# page_timeout=30_000, # 10 秒超时
# delay_before_return_html=0,
)
async with AsyncWebCrawler(config=browser_conf) as crawler:
result = await crawler.arun(url=url, config=run_conf)
# 统计
net = result.network_requests or []
req_cnt = sum(1 for r in net if r.get("event_type") == "request")
resp_cnt = sum(1 for r in net if r.get("event_type") == "response")
fail_cnt = sum(1 for r in net if r.get("event_type") == "request_failed")
_task_store[task_id] = {
"status": "completed",
"result": {
"url": result.url,
"network_requests": net,
"total_requests": req_cnt,
"total_responses": resp_cnt,
"total_failed": fail_cnt,
},
"created_at": datetime.now(timezone.utc),
}
except Exception as exc:
_task_store[task_id] = {
"status": "failed",
"result": {"error": str(exc)},
"created_at": datetime.now(timezone.utc),
}
# ---------- 定期清理过期任务 ----------
async def _gc_tasks() -> None:
while True:
await asyncio.sleep(60) # 每分钟扫一次
cutoff = datetime.now(timezone.utc) - TASK_TTL
to_del = [tid for tid, t in _task_store.items() if t["created_at"] < cutoff]
for tid in to_del:
_task_store.pop(tid, None)
# ---------- MCP Tools ----------
@mcp.tool("start_crawl")
async def start_crawl(
url: str = Field(description="要爬取的网址"),
) -> str:
"""
启动异步爬取任务，立即返回 task_id，供后续轮询。
"""
# 启动垃圾回收任务（如果还没有启动）
global _gc_task
if _gc_task is None:
_gc_task = asyncio.create_task(_gc_tasks())
task_id = str(uuid.uuid4())
_task_store[task_id] = {
"status": "running",
"created_at": datetime.now(timezone.utc),
}
# 后台执行
asyncio.create_task(_crawl_task(task_id, url))
return task_id
@mcp.tool("get_crawl_result")
async def get_crawl_result(
task_id: str = Field(description="start_crawl 返回的任务 ID"),
) -> Dict:
"""
查询任务状态与结果。
返回示例
{
"status": "running" | "completed" | "failed" | "not_found",
"result": <具体数据或错误信息>
}
"""
task = _task_store.get(task_id)
if not task:
return {"status": "not_found"}
# 仅返回必要字段，避免 MCP 消息过大
return {"status": task["status"], "result": task.get("result")}
# ---------- 启动入口 ----------
def main():
mcp.run(transport="stdio")
if __name__ == "__main__":
main()