diff --git a/scrapling/spiders/__init__.py b/scrapling/spiders/__init__.py
new file mode 100644
index 0000000..ee3c00c
--- /dev/null
+++ b/scrapling/spiders/__init__.py
@@ -0,0 +1,20 @@
+from .spider import Spider, SessionConfigurationError, LogCounterHandler
+from .request import Request
+from .result import CrawlStats, CrawlResult
+from .engine import CrawlerEngine
+from .session import SessionManager
+from .scheduler import Scheduler
+from scrapling.engines.toolbelt.custom import Response
+
+__all__ = [
+    "Spider",
+    "SessionConfigurationError",
+    "LogCounterHandler",
+    "Request",
+    "CrawlerEngine",
+    "CrawlStats",
+    "CrawlResult",
+    "SessionManager",
+    "Scheduler",
+    "Response",
+]
diff --git a/scrapling/spiders/engine.py b/scrapling/spiders/engine.py
new file mode 100644
index 0000000..e4210e9
--- /dev/null
+++ b/scrapling/spiders/engine.py
@@ -0,0 +1,153 @@
+"""Crawler engine that orchestrates the crawling process."""
+
+import json
+
+import anyio
+from anyio import create_task_group, CapacityLimiter
+
+from scrapling.core.utils import log
+from scrapling.spiders.request import Request
+from scrapling.spiders.result import CrawlStats
+from scrapling.spiders.scheduler import Scheduler
+from scrapling.spiders.session import SessionManager
+from scrapling.core._types import Dict, TYPE_CHECKING, Any
+
+if TYPE_CHECKING:
+    from scrapling.spiders.spider import Spider
+
+
+def _dump(obj: Dict) -> str:
+    return json.dumps(obj, indent=4)
+
+
+class CrawlerEngine:
+    """Orchestrates the crawling process."""
+
+    def __init__(self, spider: "Spider", session_manager: SessionManager, scheduler: Scheduler | None = None):
+        self.spider = spider
+        self.session_manager = session_manager
+        self.scheduler = scheduler or Scheduler()
+        self.stats = CrawlStats()
+
+        self._global_limiter = CapacityLimiter(spider.concurrent_requests)
+        self._domain_limiters: dict[str, CapacityLimiter] = {}
+        self._allowed_domains: set[str] = spider.allowed_domains or set()
+
+        self._active_tasks: int = 0
+        self._running: bool = False
+        self._items: list[dict[str, Any]] = []
+
+    def _is_domain_allowed(self, request: Request) -> bool:
+        """Check if the request's domain is in allowed_domains."""
+        if not self._allowed_domains:
+            return True
+
+        domain = request.domain
+        for allowed in self._allowed_domains:
+            if domain == allowed or domain.endswith("." + allowed):
+                return True
+        return False
+
+    def _rate_limiter(self, domain: str) -> CapacityLimiter:
+        """Get or create a per-domain concurrency limiter if enabled, otherwise use the global limiter."""
+        if self.spider.concurrent_requests_per_domain:
+            if domain not in self._domain_limiters:
+                self._domain_limiters[domain] = CapacityLimiter(self.spider.concurrent_requests_per_domain)
+            return self._domain_limiters[domain]
+        return self._global_limiter
+
+    async def _process_request(self, request: Request) -> None:
+        """Download and process a single request."""
+        async with self._rate_limiter(request.domain):
+            if self.spider.download_delay:
+                await anyio.sleep(self.spider.download_delay)
+
+            if request._session_kwargs.get("proxy"):
+                self.stats.proxies.append(request._session_kwargs["proxy"])
+            if request._session_kwargs.get("proxies"):
+                self.stats.proxies.append(dict(request._session_kwargs["proxies"]))
+            try:
+                response = await self.session_manager.fetch(request)
+                self.stats.increment_requests_count(request.sid)
+                self.stats.increment_response_bytes(request.domain, len(response.body))
+                self.stats.increment_status(response.status)
+
+            except Exception as e:
+                self.stats.failed_requests_count += 1
+                await self.spider.on_error(request, e)
+                return
+
+        # Process response through callback
+        callback = request.callback if request.callback else self.spider.parse
+        try:
+            async for result in callback(response):
+                if isinstance(result, Request):
+                    if self._is_domain_allowed(result):
+                        await self.scheduler.enqueue(result)
+                    else:
+                        self.stats.offsite_requests_count += 1
+                        log.debug(f"Filtered offsite request to: {result.url}")
+                elif isinstance(result, dict):
+                    await self._handle_item(result)
+                    log.debug(f"Scraped from {str(response)}\n{result}")
+        except Exception as e:
+            await self.spider.on_error(request, e)
+
+    async def _handle_item(self, item: dict[str, Any]) -> None:
+        """Handle a scraped item. Override or extend for item pipelines."""
+        self.stats.items_scraped += 1
+        self._items.append(item)
+
+    async def _task_wrapper(self, request: Request) -> None:
+        """Wrapper to track active task count."""
+        try:
+            await self._process_request(request)
+        finally:
+            self._active_tasks -= 1
+
+    async def crawl(self) -> CrawlStats:
+        """Run the spider and return CrawlStats."""
+        self._running = True
+        self._items.clear()
+        self.stats = CrawlStats(start_time=anyio.current_time())
+
+        async with self.session_manager:
+            self.stats.concurrent_requests = self.spider.concurrent_requests
+            self.stats.concurrent_requests_per_domain = self.spider.concurrent_requests_per_domain
+            self.stats.download_delay = self.spider.download_delay
+            await self.spider.on_start()
+
+            try:
+                async for request in self.spider.start_requests():
+                    await self.scheduler.enqueue(request)
+
+                # Process queue
+                async with create_task_group() as tg:
+                    while self._running:
+                        if self.scheduler.is_empty:
+                            # Empty queue + no active tasks = done
+                            if self._active_tasks == 0:
+                                self._running = False
+                                log.debug("Spider idle")
+                                break
+
+                            # Brief wait for callbacks to enqueue new requests
+                            await anyio.sleep(0.05)
+                            continue
+
+                        request = await self.scheduler.dequeue()
+                        self._active_tasks += 1
+                        tg.start_soon(self._task_wrapper, request)
+
+            finally:
+                await self.spider.on_close()
+
+        self.stats.log_levels_counter = self.spider._log_counter.get_counts()
+        self.stats.end_time = anyio.current_time()
+        log.info(_dump(self.stats.to_dict()))
+        return self.stats
+
+    @property
+    def items(self) -> list[dict[str, Any]]:
+        """Access scraped items."""
+        return self._items
diff --git a/scrapling/spiders/request.py b/scrapling/spiders/request.py
new file mode 100644
index 0000000..495c73f
--- /dev/null
+++ b/scrapling/spiders/request.py
@@ -0,0 +1,59 @@
+from urllib.parse import urlparse
+
+from scrapling.engines.toolbelt.custom import Response
+from scrapling.core._types import Any, AsyncGenerator, Callable, Dict, Union
+
+
+class Request:
+    def __init__(
+        self,
+        url: str,
+        sid: str = "",
+        callback: Callable[[Response], AsyncGenerator[Union[Dict[str, Any], "Request", None], None]] | None = None,
+        priority: int = 0,
+        dont_filter: bool = False,
+        meta: dict[str, Any] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        self.url: str = url
+        self.sid: str = sid
+        self.callback = callback
+        self.priority: int = priority
+        self.dont_filter: bool = dont_filter
+        self.meta: dict[str, Any] = meta if meta else {}
+        self._session_kwargs = kwargs if kwargs else {}
+
+    @property
+    def domain(self) -> str:
+        return urlparse(self.url).netloc
+
+    @property
+    def _fp(self) -> str:
+        """Generate a unique fingerprint for deduplication."""
+        # TODO: Improve fingerprint
+        return f"{self.sid}:{self.url}"
+
+    def __repr__(self) -> str:
+        callback_name = getattr(self.callback, "__name__", None) or "None"
+        return f"<Request({self.url}) priority={self.priority} callback={callback_name}>"
+
+    def __str__(self) -> str:
+        return self.url
+
+    def __lt__(self, other: object) -> bool:
+        """Compare requests by priority"""
+        if not isinstance(other, Request):
+            return NotImplemented
+        return self.priority < other.priority
+
+    def __gt__(self, other: object) -> bool:
+        """Compare requests by priority"""
+        if not isinstance(other, Request):
+            return NotImplemented
+        return self.priority > other.priority
+
+    def __eq__(self, other: object) -> bool:
+        """Requests are equal if they have the same fingerprint."""
+        if not isinstance(other, Request):
+            return NotImplemented
+        return self._fp == other._fp
diff --git a/scrapling/spiders/result.py b/scrapling/spiders/result.py
new file mode 100644
index 0000000..d36b0ad
--- /dev/null
+++ b/scrapling/spiders/result.py
@@ -0,0 +1,82 @@
+from dataclasses import dataclass, field
+
+from scrapling.core._types import Any, Iterator, Dict, List, Tuple
+
+
+@dataclass
+class CrawlStats:
+    """Statistics for a crawl run."""
+
+    requests_count: int = 0
+    concurrent_requests: int = 0
+    concurrent_requests_per_domain: int = 0
+    failed_requests_count: int = 0
+    offsite_requests_count: int = 0
+    response_bytes: int = 0
+    items_scraped: int = 0
+    start_time: float = 0.0
+    end_time: float = 0.0
+    download_delay: float = 0.0
+    blocked_requests_count: int = 0
+    custom_stats: Dict = field(default_factory=dict)
+    response_status_count: Dict = field(default_factory=dict)
+    domains_response_bytes: Dict = field(default_factory=dict)
+    sessions_requests_count: Dict = field(default_factory=dict)
+    proxies: List[str | Dict | Tuple] = field(default_factory=list)
+    log_levels_counter: Dict = field(default_factory=dict)
+
+    @property
+    def elapsed_seconds(self) -> float:
+        return self.end_time - self.start_time
+
+    @property
+    def requests_per_second(self) -> float:
+        if self.elapsed_seconds == 0:
+            return 0.0
+        return self.requests_count / self.elapsed_seconds
+
+    def increment_status(self, status: int) -> None:
+        self.response_status_count[f"status_{status}"] = self.response_status_count.get(f"status_{status}", 0) + 1
+
+    def increment_response_bytes(self, domain: str, count: int) -> None:
+        self.response_bytes += count
+        self.domains_response_bytes[domain] = self.domains_response_bytes.get(domain, 0) + count
+
+    def increment_requests_count(self, sid: str) -> None:
+        self.requests_count += 1
+        self.sessions_requests_count[sid] = self.sessions_requests_count.get(sid, 0) + 1
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "items_scraped": self.items_scraped,
+            "elapsed_seconds": round(self.elapsed_seconds, 2),
+            "download_delay": round(self.download_delay, 2),
+            "concurrent_requests": self.concurrent_requests,
+            "concurrent_requests_per_domain": self.concurrent_requests_per_domain,
+            "requests_count": self.requests_count,
+            "requests_per_second": round(self.requests_per_second, 2),
+            "sessions_requests_count": self.sessions_requests_count,
+            "failed_requests_count": self.failed_requests_count,
+            "offsite_requests_count": self.offsite_requests_count,
+            "blocked_requests_count": self.blocked_requests_count,
+            "response_status_count": self.response_status_count,
+            "response_bytes": self.response_bytes,
+            "domains_response_bytes": self.domains_response_bytes,
+            "proxies": self.proxies,
+            "custom_stats": self.custom_stats,
+            "log_count": self.log_levels_counter,
+        }
+
+
+@dataclass
+class CrawlResult:
+    """Complete result from a spider run."""
+
+    stats: CrawlStats
+    items: list[dict[str, Any]]
+
+    def __len__(self) -> int:
+        return len(self.items)
+
+    def __iter__(self) -> Iterator[dict[str, Any]]:
+        return iter(self.items)
diff --git a/scrapling/spiders/scheduler.py b/scrapling/spiders/scheduler.py
new file mode 100644
index 0000000..2890e50
--- /dev/null
+++ b/scrapling/spiders/scheduler.py
@@ -0,0 +1,45 @@
+import asyncio
+from itertools import count
+
+from scrapling.core.utils import log
+from scrapling.spiders.request import Request
+
+
+class Scheduler:
+    """
+    Priority queue with URL deduplication. (heapq)
+
+    Higher priority requests are processed first.
+    Duplicate URLs are filtered unless dont_filter=True.
+    """
+
+    def __init__(self):
+        self._queue: asyncio.PriorityQueue[tuple[int, int, Request]] = asyncio.PriorityQueue()
+        self._seen: set[str] = set()
+        self._counter = count()
+
+    async def enqueue(self, request: Request) -> bool:
+        """Add a request to the queue."""
+        fingerprint = request._fp
+
+        if not request.dont_filter and fingerprint in self._seen:
+            log.debug("Dropped duplicate request: %s", request)
+            return False
+
+        self._seen.add(fingerprint)
+
+        # Negative priority so higher priority = dequeued first
+        await self._queue.put((-request.priority, next(self._counter), request))
+        return True
+
+    async def dequeue(self) -> Request:
+        """Get the next request to process."""
+        _, _, request = await self._queue.get()
+        return request
+
+    def __len__(self) -> int:
+        return self._queue.qsize()
+
+    @property
+    def is_empty(self) -> bool:
+        return self._queue.empty()
diff --git a/scrapling/spiders/session.py b/scrapling/spiders/session.py
new file mode 100644
index 0000000..452d12b
--- /dev/null
+++ b/scrapling/spiders/session.py
@@ -0,0 +1,129 @@
+from scrapling.spiders.request import Request
+from scrapling.engines.static import _ASyncSessionLogic
+from scrapling.engines.toolbelt.convertor import Response
+from scrapling.core._types import cast, SUPPORTED_HTTP_METHODS
+from scrapling.fetchers import AsyncDynamicSession, AsyncStealthySession, FetcherSession
+
+Session = FetcherSession | AsyncDynamicSession | AsyncStealthySession
+
+
+class SessionManager:
+    """Manages pre-configured session instances."""
+
+    def __init__(self):
+        self._sessions: dict[str, Session] = {}
+        self._default_session_id: str | None = None
+        self._started: bool = False
+
+    def add(self, session_id: str, session: Session, *, default: bool = False) -> "SessionManager":
+        """Register a session instance.
+
+        :param session_id: Name to reference this session in requests
+        :param session: Your pre-configured session instance
+        :param default: If True, this becomes the default session
+        """
+        if session_id in self._sessions:
+            raise ValueError(f"Session '{session_id}' already registered")
+
+        self._sessions[session_id] = session
+
+        if default or self._default_session_id is None:
+            self._default_session_id = session_id
+
+        return self
+
+    def remove(self, session_id: str) -> None:
+        """Removes a session.
+
+        :param session_id: ID of session to remove
+        """
+        _ = self.pop(session_id)
+
+    def pop(self, session_id: str) -> Session:
+        """Remove and returns a session.
+
+        :param session_id: ID of session to remove
+        """
+        if session_id not in self._sessions:
+            raise KeyError(f"Session '{session_id}' not found")
+
+        session = self._sessions.pop(session_id)
+
+        if session and self._default_session_id == session_id:
+            self._default_session_id = next(iter(self._sessions), None)
+
+        return session
+
+    @property
+    def default_session_id(self) -> str:
+        if self._default_session_id is None:
+            raise RuntimeError("No sessions registered")
+        return self._default_session_id
+
+    @property
+    def session_ids(self) -> list[str]:
+        return list(self._sessions.keys())
+
+    def get(self, session_id: str) -> Session:
+        if session_id not in self._sessions:
+            available = ", ".join(self._sessions.keys())
+            raise KeyError(f"Session '{session_id}' not found. Available: {available}")
+        return self._sessions[session_id]
+
+    async def start(self) -> None:
+        """Start all sessions that aren't already alive."""
+        if self._started:
+            return
+
+        for session in self._sessions.values():
+            if not session._is_alive:
+                await session.__aenter__()
+
+        self._started = True
+
+    async def close(self) -> None:
+        """Close all registered sessions."""
+        for session in self._sessions.values():
+            _ = await session.__aexit__(None, None, None)
+
+        self._started = False
+
+    async def fetch(self, request: Request) -> Response:
+        sid = request.sid if request.sid else self.default_session_id
+        session = self.get(sid)
+
+        if session:
+            if isinstance(session, FetcherSession):
+                session = session._client
+
+                if isinstance(session, _ASyncSessionLogic):
+                    response = await session._make_request(
+                        method=cast(SUPPORTED_HTTP_METHODS, request._session_kwargs.pop("method", "GET")),
+                        url=request.url,
+                        **request._session_kwargs,
+                    )
+                else:
+                    # Sync session or other types - shouldn't happen in async context
+                    raise TypeError(f"Session type {type(session)} not supported for async fetch")
+            else:
+                response = await session.fetch(url=request.url, **request._session_kwargs)
+
+            response.request = request
+            response.meta = request.meta
+            return response
+        raise RuntimeError("No session found with the request session id")
+
+    async def __aenter__(self) -> "SessionManager":
+        await self.start()
+        return self
+
+    async def __aexit__(self, *exc) -> None:
+        await self.close()
+
+    def __contains__(self, session_id: str) -> bool:
+        """Check if a session ID is registered."""
+        return session_id in self._sessions
+
+    def __len__(self) -> int:
+        """Number of registered sessions."""
+        return len(self._sessions)
diff --git a/scrapling/spiders/spider.py b/scrapling/spiders/spider.py
new file mode 100644
index 0000000..a111cd6
--- /dev/null
+++ b/scrapling/spiders/spider.py
@@ -0,0 +1,206 @@
+import logging
+from pathlib import Path
+from abc import ABC
+
+import anyio
+
+from scrapling.spiders.request import Request
+from scrapling.spiders.result import CrawlResult
+from scrapling.spiders.engine import CrawlerEngine
+from scrapling.spiders.session import SessionManager
+from scrapling.core.utils import set_logger, reset_logger
+from scrapling.core._types import Set, Any, Dict, Optional, TYPE_CHECKING, AsyncGenerator
+
+BLOCKED_CODES = {401, 403, 407, 429, 444, 500, 502, 503, 504}
+if TYPE_CHECKING:
+    from scrapling.engines.toolbelt.custom import Response
+
+
+class LogCounterHandler(logging.Handler):
+    """A logging handler that counts log messages by level."""
+
+    def __init__(self):
+        super().__init__()
+        self.counts = {
+            logging.DEBUG: 0,
+            logging.INFO: 0,
+            logging.WARNING: 0,
+            logging.ERROR: 0,
+            logging.CRITICAL: 0,
+        }
+
+    def emit(self, record: logging.LogRecord) -> None:
+        level = record.levelno
+        # Map to the closest standard level
+        if level >= logging.CRITICAL:
+            self.counts[logging.CRITICAL] += 1
+        elif level >= logging.ERROR:
+            self.counts[logging.ERROR] += 1
+        elif level >= logging.WARNING:
+            self.counts[logging.WARNING] += 1
+        elif level >= logging.INFO:
+            self.counts[logging.INFO] += 1
+        else:
+            self.counts[logging.DEBUG] += 1
+
+    def get_counts(self) -> Dict[str, int]:
+        """Return counts as a dictionary with string keys."""
+        return {
+            "debug": self.counts[logging.DEBUG],
+            "info": self.counts[logging.INFO],
+            "warning": self.counts[logging.WARNING],
+            "error": self.counts[logging.ERROR],
+            "critical": self.counts[logging.CRITICAL],
+        }
+
+
+class SessionConfigurationError(Exception):
+    """Raised when session configuration fails."""
+
+    pass
+
+
+class Spider(ABC):
+    """An abstract base class for creating web spiders.
+
+    Check the documentation website for more information.
+    """
+
+    name: Optional[str] = None
+    start_urls: list[str] = []
+    allowed_domains: Set[str] = set()
+
+    # Concurrency settings
+    concurrent_requests: int = 16
+    concurrent_requests_per_domain: int = 0
+    download_delay: float = 0.0
+
+    # Logging settings
+    logging_level: int = logging.DEBUG
+    log_file: Optional[str] = None
+
+    def __init__(self):
+        if self.name is None:
+            raise ValueError(f"{self.__class__.__name__} must have a name.")
+
+        self.logger = logging.getLogger(f"scrapling.spiders.{self.name}")
+        self.logger.setLevel(self.logging_level)
+        self.logger.handlers.clear()
+        self.logger.propagate = False  # Don't propagate to parent 'scrapling' logger
+
+        formatter = logging.Formatter(
+            fmt=f"[%(asctime)s]:({self.name}) %(levelname)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
+        )
+
+        # Add a log counter handler to track log counts by level
+        self._log_counter = LogCounterHandler()
+        self.logger.addHandler(self._log_counter)
+
+        console_handler = logging.StreamHandler()
+        console_handler.setFormatter(formatter)
+        self.logger.addHandler(console_handler)
+
+        if self.log_file:
+            Path(self.log_file).parent.mkdir(parents=True, exist_ok=True)
+            file_handler = logging.FileHandler(self.log_file)
+            file_handler.setFormatter(formatter)
+            self.logger.addHandler(file_handler)
+
+        self._session_manager = SessionManager()
+        try:
+            self.configure_sessions(self._session_manager)
+        except Exception as e:
+            raise SessionConfigurationError(f"Error in {self.__class__.__name__}.configure_sessions(): {e}") from e
+
+        if len(self._session_manager) == 0:
+            raise SessionConfigurationError(f"{self.__class__.__name__}.configure_sessions() did not add any sessions")
+
+        self.logger.info("Spider initialized")
+
+    async def start_requests(self) -> AsyncGenerator[Request, None]:
+        """Generate initial requests to start the crawl.
+
+        By default, this generates Request objects for each URL in `start_urls`
+        using the session manager's default session and `parse()` as callback.
+
+        Override this method for more control over initial requests
+        (e.g., to add custom headers, use different callbacks, etc.)
+        """
+        if not self.start_urls:
+            raise RuntimeError(
+                "Spider has no starting point, either set `start_urls` or override `start_requests` function."
+            )
+
+        for url in self.start_urls:
+            yield Request(url, sid=self._session_manager.default_session_id)
+
+    async def parse(self, response: "Response") -> AsyncGenerator[Dict[str, Any] | Request | None, None]:
+        """Default callback for processing responses"""
+        raise NotImplementedError(f"{self.__class__.__name__} must implement parse() method")
+        yield  # Make this a generator
+
+    async def on_start(self) -> None:
+        """Called before crawling starts. Override for setup logic."""
+        self.logger.debug("Starting spider")
+
+    async def on_close(self) -> None:
+        """Called after crawling finishes. Override for cleanup logic."""
+        self.logger.debug("Spider closed")
+
+    async def on_error(self, request: Request, error: Exception) -> None:
+        """
+        Handle request errors for all spider requests.
+
+        Override for custom error handling.
+        """
+        self.logger.error(error, exc_info=error)
+
+    @staticmethod
+    async def is_blocked(response: "Response") -> bool:
+        """Check if the response is blocked."""
+        # TODO
+        if response.status in BLOCKED_CODES:
+            return True
+        return False
+
+    def __repr__(self) -> str:
+        """String representation of the spider."""
+        return f"<{self.__class__.__name__} '{self.name}'>"
+
+    def configure_sessions(self, manager: SessionManager) -> None:
+        """Configure sessions for this spider.
+
+        Override this method to add custom sessions.
+        The default implementation creates a FetcherSession session.
+
+        The first session added becomes the default for `start_requests()` unless specified otherwise.
+
+        :param manager: SessionManager to configure
+        """
+        from scrapling.fetchers import FetcherSession
+
+        manager.add("default", FetcherSession())
+
+    async def __run(self) -> CrawlResult:
+        token = set_logger(self.logger)
+        try:
+            engine = CrawlerEngine(self, session_manager=self._session_manager)
+            stats = await engine.crawl()
+            return CrawlResult(stats=stats, items=engine.items)
+        finally:
+            reset_logger(token)
+            # Close any file handlers to release file resources.
+            if self.log_file:
+                for handler in self.logger.handlers:
+                    if isinstance(handler, logging.FileHandler):
+                        handler.close()
+
+    def start(self, backend_options: Dict[str, Any] | None = None) -> CrawlResult:
+        """Run the spider and return results.
+
+        This is the main entry point for running a spider.
+        Handles async execution internally via anyio.
+
+        :param backend_options: Asyncio backend options to be used with `anyio.run`
+        """
+        return anyio.run(self.__run, backend="asyncio", backend_options=backend_options)