"""Crawl4AI Firecrawl-compatible API server.
Drops in as a local replacement for Firecrawl by serving the same API
endpoints on a configurable port (default 11235).
Endpoints:
POST /scrape — scrape a single URL, return markdown
POST /v1/scrape — same (v1 compat)
POST /search — web search via DuckDuckGo (no API key needed)
POST /v1/search — same
GET /health — health check
Config via env:
CRAWL4AI_PORT — listen port (default 11235)
CRAWL4AI_HOST — bind address (default 127.0.0.1)
CRAWL4AI_API_KEY — optional API key for auth (no default)
"""
from __future__ import annotations
import asyncio
import json
import logging
import os
import sys
from contextlib import asynccontextmanager
from typing import Any, Dict, List, Optional
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
# Wire Crawl4AI venv onto sys.path so we can import it even when
# this server is launched from outside the venv.
_VENV = os.path.dirname(os.path.abspath(__file__))
_SITE_PKGS = os.path.join(_VENV, "lib", "python3.11", "site-packages")
if _SITE_PKGS not in sys.path:
sys.path.insert(0, _SITE_PKGS)
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, CacheMode
logger = logging.getLogger("crawl4ai-server")
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
)
# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------
_crawler: Optional[AsyncWebCrawler] = None
async def _get_crawler() -> AsyncWebCrawler:
"""Lazily initialize the async crawler singleton."""
global _crawler
if _crawler is None:
_crawler = AsyncWebCrawler()
await _crawler.__aenter__()
logger.info("Crawl4AI crawler initialized")
return _crawler
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Startup / shutdown hook."""
logger.info("Crawl4AI server starting on port %s", os.getenv("CRAWL4AI_PORT", "11235"))
yield
if _crawler is not None:
await _crawler.__aexit__(None, None, None)
logger.info("Crawl4AI server shutting down")
app = FastAPI(title="Crawl4AI Server", version="1.0.0", lifespan=lifespan)
# ---------------------------------------------------------------------------
# Auth
# ---------------------------------------------------------------------------
def _check_auth(request: Request) -> None:
required_key = os.getenv("CRAWL4AI_API_KEY", "").strip()
if not required_key:
return # No auth required
auth = request.headers.get("authorization", "")
api_key = request.headers.get("x-api-key", "")
if not auth.startswith("Bearer "):
auth = ""
provided = auth.removeprefix("Bearer ").strip() or api_key.strip()
if provided != required_key:
raise HTTPException(status_code=401, detail="Invalid API key")
# ---------------------------------------------------------------------------
# Scraping
# ---------------------------------------------------------------------------
async def _scrape_url(url: str, **kwargs: Any) -> Dict[str, Any]:
"""Scrape a single URL and return markdown."""
crawler = await _get_crawler()
run_config = CrawlerRunConfig(
cache_mode=CacheMode.BYPASS,
)
result = await crawler.arun(url, config=run_config)
markdown_text = str(result.markdown) if result.markdown else ""
html_text = result.html if result.html else ""
meta = result.metadata or {}
return {
"success": True,
"data": {
"markdown": markdown_text,
"html": html_text,
"metadata": {
"title": meta.get("title", "") or "",
"description": meta.get("description", "") or "",
"language": meta.get("language", "") or "",
"url": result.url or url,
},
},
}
@app.post("/scrape")
@app.post("/v1/scrape")
async def scrape(request: Request) -> JSONResponse:
_check_auth(request)
body = await request.json()
url = body.get("url", "")
if not url:
raise HTTPException(status_code=400, detail="url is required")
try:
result = await _scrape_url(url)
return JSONResponse(content=result)
except Exception as exc:
logger.error("Scrape error for %s: %s", url, exc)
return JSONResponse(
status_code=500,
content={"success": False, "error": str(exc)},
)
# ---------------------------------------------------------------------------
# Search (via DuckDuckGo — no API key)
# ---------------------------------------------------------------------------
async def _search_query(query: str, limit: int = 5) -> Dict[str, Any]:
"""Search via DuckDuckGo using the ddgs library."""
from ddgs import DDGS
# DDGS is flaky with rate limiting — retry up to 3 times with backoff.
import time
max_attempts = 3
for attempt in range(max_attempts):
try:
with DDGS() as ddgs:
results = list(ddgs.text(query, max_results=limit))
if results:
web_results = []
for i, r in enumerate(results):
web_results.append({
"title": r.get("title", ""),
"url": r.get("href", ""),
"description": r.get("body", ""),
"position": i + 1,
})
return {
"success": True,
"data": {"web": web_results},
}
logger.warning("DDGS returned 0 results for '%s' (attempt %d/%d)", query, attempt+1, max_attempts)
if attempt < max_attempts - 1:
time.sleep(2 * (attempt + 1)) # 2s, 4s, 6s backoff
continue
except Exception as exc:
logger.error("Search error for '%s' (attempt %d/%d): %s", query, attempt+1, max_attempts, exc)
if attempt < max_attempts - 1:
time.sleep(2 * (attempt + 1))
continue
raise
return {
"success": True,
"data": {"web": []},
}
@app.post("/search")
@app.post("/v1/search")
async def search(request: Request) -> JSONResponse:
_check_auth(request)
body = await request.json()
query = body.get("query", body.get("q", ""))
limit = body.get("limit", 5)
if not query:
raise HTTPException(status_code=400, detail="query is required")
result = await _search_query(query, limit=limit)
status = 200 if result.get("success") else 500
return JSONResponse(content=result, status_code=status)
# ---------------------------------------------------------------------------
# Health
# ---------------------------------------------------------------------------
@app.get("/health")
@app.get("/v1/health")
async def health() -> JSONResponse:
return JSONResponse(content={"status": "ok", "version": "1.0.0"})
if __name__ == "__main__":
import uvicorn
port = int(os.getenv("CRAWL4AI_PORT", "11235"))
host = os.getenv("CRAWL4AI_HOST", "127.0.0.1")
uvicorn.run(
"server:app",
host=host,
port=port,
log_level="info",
)