Crawl4AI MCP Server Guide: Setup for Cursor & Claude (2026 Update)

pip install mcp

Claude Desktop → MCP Client → MCP Server (your code) → Crawl4AI → Target Website
     ↓                    ↓                    ↓               ↓            ↓
 User request       MCP protocol         Tool handler    Async crawler   HTTP request
                     (JSON-RPC)

#!/usr/bin/env python3
"""
Crawl4AI MCP Server
Exposes web scraping tools to Claude Desktop via Model Context Protocol
"""

from crawl4ai import AsyncWebCrawler
from crawl4ai.extraction_strategy import LLMExtractionStrategy
from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Resource,
    Tool,
    TextContent,
    ImageContent,
    EmbeddedResource,
)
from typing import Any, Optional
import json
import asyncio

app = Server("crawl4ai-server")

@app.list_tools()
async def handle_list_tools() -> list[Tool]:
    """
    List available scraping tools.
    Claude Desktop uses this to discover what operations are available.
    """
    return [
        Tool(
            name="scrape_url",
            description="Scrape a single URL and return content as markdown",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL to scrape (must include http:// or https://)"
                    },
                    "wait_for_selector": {
                        "type": "string",
                        "description": "Optional CSS selector to wait for before scraping"
                    }
                },
                "required": ["url"]
            }
        ),
        Tool(
            name="extract_with_css",
            description="Extract specific elements using CSS selectors",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "URL to scrape"
                    },
                    "selector": {
                        "type": "string",
                        "description": "CSS selector for elements to extract"
                    },
                    "attribute": {
                        "type": "string",
                        "description": "Attribute to extract (text, href, src, etc.)",
                        "default": "text"
                    }
                },
                "required": ["url", "selector"]
            }
        ),
        Tool(
            name="crawl_multiple",
            description="Crawl multiple pages from a website",
            inputSchema={
                "type": "object",
                "properties": {
                    "url": {
                        "type": "string",
                        "description": "Starting URL"
                    },
                    "max_pages": {
                        "type": "integer",
                        "description": "Maximum number of pages to crawl",
                        "default": 5
                    }
                },
                "required": ["url"]
            }
        )
    ]

@app.call_tool()
async def handle_call_tool(name: str, arguments: dict | None) -> list[TextContent | ImageContent | EmbeddedResource]:
    """
    Handle tool calls from Claude Desktop.
    This is where the actual scraping happens.
    """
    if arguments is None:
        arguments = {}

    try:
        if name == "scrape_url":
            return await scrape_url(arguments.get("url"), arguments.get("wait_for_selector"))
        elif name == "extract_with_css":
            return await extract_with_css(
                arguments.get("url"),
                arguments.get("selector"),
                arguments.get("attribute", "text")
            )
        elif name == "crawl_multiple":
            return await crawl_multiple(
                arguments.get("url"),
                arguments.get("max_pages", 5)
            )
        else:
            raise ValueError(f"Unknown tool: {name}")
    except Exception as e:
        return [TextContent(
            type="text",
            text=f"Error during scraping: {str(e)}\n\nTip: Check if the URL is accessible and respects robots.txt"
        )]

async def scrape_url(url: str, wait_for_selector: Optional[str] = None) -> list[TextContent]:
    """Scrape a single URL and return markdown content"""
    if not url:
        return [TextContent(type="text", text="Error: URL is required")]

    async with AsyncWebCrawler(verbose=True) as crawler:
        result = await crawler.arun(
            url=url,
            word_count_threshold=10,
            exclude_external_links=False,
            wait_for_selector=wait_for_selector
        )

        if result.success:
            markdown_content = f"""# Scraped Content from {url}

{result.markdown}

---

## Metadata
- **Status**: Success
- **Word Count**: {result.extracted_content_word_count if hasattr(result, 'extracted_content_word_count') else 'N/A'}
- **Links Found**: {len(result.links.get('internal', [])) + len(result.links.get('external', [])) if hasattr(result, 'links') else 'N/A'}
- **Screenshot**: {'Available' else 'Not available'}
"""
            return [TextContent(type="text", text=markdown_content)]
        else:
            return [TextContent(
                type="text",
                text=f"Failed to scrape {url}\n\nError: {result.error_message if hasattr(result, 'error_message') else 'Unknown error'}"
            )]

async def extract_with_css(url: str, selector: str, attribute: str = "text") -> list[TextContent]:
    """Extract specific elements using CSS selectors"""
    if not url or not selector:
        return [TextContent(type="text", text="Error: URL and selector are required")]

    async with AsyncWebCrawler(verbose=True) as crawler:
        # Use CSS extraction strategy
        result = await crawler.arun(
            url=url,
            css_selector=selector,
            bypass_cache=True
        )

        if result.success:
            extracted_data = result.extracted_content if hasattr(result, 'extracted_content') else result.markdown
            content = f"""# CSS Extraction Results from {url}

**Selector**: `{selector}`
**Attribute**: `{attribute}`

## Extracted Data
```
{extracted_data}
```

---

## Metadata
- **Status**: Success
- **Elements Found**: {len(result.extracted_content) if isinstance(result.extracted_content, list) else '1' if hasattr(result, 'extracted_content') else 'N/A'}
"""
            return [TextContent(type="text", text=content)]
        else:
            return [TextContent(
                type="text",
                text=f"CSS extraction failed for {url}\n\nError: {result.error_message if hasattr(result, 'error_message') else 'Unknown error'}"
            )]

async def crawl_multiple(url: str, max_pages: int = 5) -> list[TextContent]:
    """Crawl multiple pages from a starting URL"""
    if not url:
        return [TextContent(type="text", text="Error: URL is required")]

    results = []
    async with AsyncWebCrawler(verbose=True) as crawler:
        # Basic crawl implementation
        result = await crawler.arun(url=url, word_count_threshold=10)

        if result.success:
            results.append(f"## Page 1: {url}\n{result.markdown}\n")

            # Get internal links for further crawling
            if hasattr(result, 'links'):
                internal_links = result.links.get('internal', [])[:max_pages-1]

                for i, link in enumerate(internal_links, 2):
                    if i > max_pages:
                        break
                    try:
                        page_result = await crawler.arun(url=link, word_count_threshold=10)
                        if page_result.success:
                            results.append(f"## Page {i}: {link}\n{page_result.markdown}\n")
                    except Exception as e:
                        results.append(f"## Page {i}: {link}\nError: {str(e)}\n")

        content = f"""# Multi-Page Crawl Results from {url}

**Pages Crawled**: {len(results)}

{chr(10).join(results)}

---

## Metadata
- **Total Pages**: {len(results)}
- **Starting URL**: {url}
- **Max Pages Limit**: {max_pages}
"""
        return [TextContent(type="text", text=content)]

async def main():
    """Run the MCP server"""
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="crawl4ai-server",
                server_version="1.0.0",
                capabilities=app.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={}
                )
            )
        )

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

{
  "mcpServers": {
    "crawl4ai": {
      "command": "python3",
      "args": ["/absolute/path/to/your/mcp_server.py"],
      "env": {
        "PYTHONPATH": "/absolute/path/to/your/project"
      }
    }
  }
}