Coverage for src/lilbee/server/routes/crawl.py: 100%

1"""Crawl route handler.""" 

2 

3from __future__ import annotations 

4 

5import anyio 

6from litestar import post 

7from litestar.exceptions import ValidationException 

8from litestar.response import Stream 

9 

10from lilbee.server import handlers 

11from lilbee.server.handlers.sse import SSE_MEDIA_TYPE 

12from lilbee.server.models import CrawlRequest 

13 

14 

15@post("/api/crawl", media_type=SSE_MEDIA_TYPE) 

16async def crawl_route(data: CrawlRequest) -> Stream: 

17 """Crawl a URL with streaming SSE progress events.""" 

18 from lilbee.crawler import require_valid_crawl_url 

19 

20 try: 

21 # URL validation resolves the host (blocking DNS), so it runs off the loop, 

22 # mirroring the MCP crawl tool so neither async transport stalls the loop. 

23 await anyio.to_thread.run_sync(require_valid_crawl_url, data.url) 

24 except ValueError as exc: 

25 raise ValidationException(str(exc)) from exc 

26 gen = handlers.crawl_stream( 

27 url=data.url, 

28 depth=data.depth, 

29 max_pages=data.max_pages, 

30 render_mode=data.render_mode, 

31 include_subdomains=data.include_subdomains, 

32 ) 

33 return Stream(gen, media_type=SSE_MEDIA_TYPE)