diff --git a/.env.example b/.env.example index 778272bf3..3ba296669 100644 --- a/.env.example +++ b/.env.example @@ -2,6 +2,9 @@ OPENAI_API_KEY= ANTHROPIC_API_KEY= GOOGLE_API_KEY= TAVILY_API_KEY= +CRW_API_KEY= +# Optional: override the fastCRW base URL for self-hosting (default: https://fastcrw.com/api) +CRW_API_URL= LANGSMITH_API_KEY= LANGSMITH_PROJECT= LANGSMITH_TRACING= diff --git a/README.md b/README.md index 5bfa38ac5..1559bd553 100644 --- a/README.md +++ b/README.md @@ -74,7 +74,7 @@ Open Deep Research supports a wide range of LLM providers via the [init_chat_mod #### Search API :mag: -Open Deep Research supports a wide range of search tools. By default it uses the [Tavily](https://www.tavily.com/) search API. Has full MCP compatibility and work native web search for Anthropic and OpenAI. See the `search_api` and `mcp_config` fields in the [configuration.py](https://github.com/langchain-ai/open_deep_research/blob/main/src/open_deep_research/configuration.py) file for more details. This can be accessed via the LangGraph Studio UI. +Open Deep Research supports a wide range of search tools. By default it uses the [Tavily](https://www.tavily.com/) search API. It also supports [fastCRW](https://fastcrw.com/) (a Firecrawl-compatible web data engine that runs as a single binary; self-host or use the managed cloud), configured via the `CRW_API_KEY` env var (and optionally `CRW_API_URL` to point at a self-hosted server). Has full MCP compatibility and work native web search for Anthropic and OpenAI. See the `search_api` and `mcp_config` fields in the [configuration.py](https://github.com/langchain-ai/open_deep_research/blob/main/src/open_deep_research/configuration.py) file for more details. This can be accessed via the LangGraph Studio UI. #### Other diff --git a/src/open_deep_research/configuration.py b/src/open_deep_research/configuration.py index 1c5bac9e9..07b7f6b39 100644 --- a/src/open_deep_research/configuration.py +++ b/src/open_deep_research/configuration.py @@ -14,6 +14,7 @@ class SearchAPI(Enum): ANTHROPIC = "anthropic" OPENAI = "openai" TAVILY = "tavily" + CRW = "crw" NONE = "none" class MCPConfig(BaseModel): @@ -84,6 +85,7 @@ class Configuration(BaseModel): "description": "Search API to use for research. NOTE: Make sure your Researcher Model supports the selected search API.", "options": [ {"label": "Tavily", "value": SearchAPI.TAVILY.value}, + {"label": "fastCRW", "value": SearchAPI.CRW.value}, {"label": "OpenAI Native Web Search", "value": SearchAPI.OPENAI.value}, {"label": "Anthropic Native Web Search", "value": SearchAPI.ANTHROPIC.value}, {"label": "None", "value": SearchAPI.NONE.value} diff --git a/src/open_deep_research/utils.py b/src/open_deep_research/utils.py index 82ce304e2..87979423e 100644 --- a/src/open_deep_research/utils.py +++ b/src/open_deep_research/utils.py @@ -172,6 +172,167 @@ async def tavily_search_async( search_results = await asyncio.gather(*search_tasks) return search_results +########################## +# fastCRW Search Tool Utils +########################## +# fastCRW is a Firecrawl-compatible web data engine shipped as a single binary. +# Self-host (free, AGPL) or use the managed cloud at https://fastcrw.com. +# The /v1/search endpoint returns Firecrawl-compatible results, optionally with +# page markdown when the "markdown" format is requested. +CRW_SEARCH_DESCRIPTION = ( + "A search engine optimized for comprehensive, accurate, and trusted results. " + "Useful for when you need to answer questions about current events." +) +@tool(description=CRW_SEARCH_DESCRIPTION) +async def crw_search( + queries: List[str], + max_results: Annotated[int, InjectedToolArg] = 5, + config: RunnableConfig = None +) -> str: + """Fetch and summarize search results from the fastCRW search API. + + Args: + queries: List of search queries to execute + max_results: Maximum number of results to return per query + config: Runtime configuration for API keys and model settings + + Returns: + Formatted string containing summarized search results + """ + # Step 1: Execute search queries asynchronously + search_results = await crw_search_async( + queries, + max_results=max_results, + config=config + ) + + # Step 2: Deduplicate results by URL to avoid processing the same content multiple times + unique_results = {} + for response in search_results: + for result in response['results']: + url = result['url'] + if url not in unique_results: + unique_results[url] = {**result, "query": response['query']} + + # Step 3: Set up the summarization model with configuration + configurable = Configuration.from_runnable_config(config) + + # Character limit to stay within model token limits (configurable) + max_char_to_include = configurable.max_content_length + + # Initialize summarization model with retry logic + model_api_key = get_api_key_for_model(configurable.summarization_model, config) + summarization_model = init_chat_model( + model=configurable.summarization_model, + max_tokens=configurable.summarization_model_max_tokens, + api_key=model_api_key, + tags=["langsmith:nostream"] + ).with_structured_output(Summary).with_retry( + stop_after_attempt=configurable.max_structured_output_retries + ) + + # Step 4: Create summarization tasks (skip empty content) + async def noop(): + """No-op function for results without raw content.""" + return None + + summarization_tasks = [ + noop() if not result.get("raw_content") + else summarize_webpage( + summarization_model, + result['raw_content'][:max_char_to_include] + ) + for result in unique_results.values() + ] + + # Step 5: Execute all summarization tasks in parallel + summaries = await asyncio.gather(*summarization_tasks) + + # Step 6: Combine results with their summaries + summarized_results = { + url: { + 'title': result['title'], + 'content': result['content'] if summary is None else summary + } + for url, result, summary in zip( + unique_results.keys(), + unique_results.values(), + summaries + ) + } + + # Step 7: Format the final output + if not summarized_results: + return "No valid search results found. Please try different search queries or use a different search API." + + formatted_output = "Search results: \n\n" + for i, (url, result) in enumerate(summarized_results.items()): + formatted_output += f"\n\n--- SOURCE {i+1}: {result['title']} ---\n" + formatted_output += f"URL: {url}\n\n" + formatted_output += f"SUMMARY:\n{result['content']}\n\n" + formatted_output += "\n\n" + "-" * 80 + "\n" + + return formatted_output + +async def crw_search_async( + search_queries, + max_results: int = 5, + config: RunnableConfig = None +): + """Execute multiple fastCRW search queries asynchronously. + + Args: + search_queries: List of search query strings to execute + max_results: Maximum number of results per query + config: Runtime configuration for API key access + + Returns: + List of search result dictionaries normalized to a Tavily-like shape + """ + # Resolve the fastCRW base URL (cloud by default, self-host override) and key + base_url = get_crw_base_url(config).rstrip("/") + api_key = get_crw_api_key(config) + + # Self-hosted fastCRW may run without auth, so the key is optional + headers = {"Content-Type": "application/json"} + if api_key: + headers["Authorization"] = f"Bearer {api_key}" + + async def search_single(query: str): + """Execute a single fastCRW search query and normalize its results.""" + payload = { + "query": query, + "limit": max_results, + # Request markdown so results can be summarized like other providers + "scrapeOptions": {"formats": ["markdown"]}, + } + async with aiohttp.ClientSession() as session: + async with session.post( + f"{base_url}/v1/search", json=payload, headers=headers + ) as response: + response.raise_for_status() + data = await response.json() + + # fastCRW envelope: {success, error, data: [{title, url, description, markdown?}]} + if not data.get("success", True): + raise ToolException(data.get("error", "fastCRW search request failed")) + + results = [ + { + "title": item.get("title", ""), + "url": item.get("url", ""), + "content": item.get("description", ""), + "raw_content": item.get("markdown"), + } + for item in data.get("data", []) + ] + return {"query": query, "results": results} + + # Execute all search queries in parallel and return results + search_tasks = [search_single(query) for query in search_queries] + search_results = await asyncio.gather(*search_tasks) + return search_results + async def summarize_webpage(model: BaseChatModel, webpage_content: str) -> str: """Summarize webpage content using AI model with timeout protection. @@ -553,12 +714,22 @@ async def get_search_tool(search_api: SearchAPI): # Configure Tavily search tool with metadata search_tool = tavily_search search_tool.metadata = { - **(search_tool.metadata or {}), - "type": "search", + **(search_tool.metadata or {}), + "type": "search", "name": "web_search" } return [search_tool] - + + elif search_api == SearchAPI.CRW: + # Configure fastCRW search tool with metadata + search_tool = crw_search + search_tool.metadata = { + **(search_tool.metadata or {}), + "type": "search", + "name": "web_search" + } + return [search_tool] + elif search_api == SearchAPI.NONE: # No search functionality configured return [] @@ -923,3 +1094,32 @@ def get_tavily_api_key(config: RunnableConfig): return api_keys.get("TAVILY_API_KEY") else: return os.getenv("TAVILY_API_KEY") + +# Default fastCRW cloud base URL (note the /api suffix); override for self-host. +CRW_DEFAULT_BASE_URL = "https://fastcrw.com/api" + +def get_crw_api_key(config: RunnableConfig): + """Get fastCRW API key from environment or config. + + Optional for self-hosted fastCRW instances that run without auth. + """ + should_get_from_config = os.getenv("GET_API_KEYS_FROM_CONFIG", "false") + if should_get_from_config.lower() == "true": + api_keys = config.get("configurable", {}).get("apiKeys", {}) + if not api_keys: + return None + return api_keys.get("CRW_API_KEY") + else: + return os.getenv("CRW_API_KEY") + +def get_crw_base_url(config: RunnableConfig): + """Get fastCRW base URL from environment or config. + + Defaults to the managed cloud; set CRW_API_URL to point at a self-hosted server. + """ + should_get_from_config = os.getenv("GET_API_KEYS_FROM_CONFIG", "false") + if should_get_from_config.lower() == "true": + api_keys = config.get("configurable", {}).get("apiKeys", {}) + return (api_keys or {}).get("CRW_API_URL") or CRW_DEFAULT_BASE_URL + else: + return os.getenv("CRW_API_URL", CRW_DEFAULT_BASE_URL) diff --git a/tests/test_crw_search.py b/tests/test_crw_search.py new file mode 100644 index 000000000..6b6d13ea2 --- /dev/null +++ b/tests/test_crw_search.py @@ -0,0 +1,141 @@ +"""Unit tests for the fastCRW search provider (mocked HTTP, no network).""" + +import asyncio +from contextlib import asynccontextmanager +from unittest.mock import patch + +from open_deep_research.configuration import SearchAPI +from open_deep_research.utils import ( + CRW_DEFAULT_BASE_URL, + crw_search_async, + get_crw_api_key, + get_crw_base_url, + get_search_tool, +) + + +class _FakeResponse: + def __init__(self, payload): + self._payload = payload + + def raise_for_status(self): + return None + + async def json(self): + return self._payload + + +class _FakeSession: + """Minimal aiohttp.ClientSession stand-in capturing calls and returning a payload.""" + + def __init__(self, payload, recorder): + self._payload = payload + self._recorder = recorder + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + @asynccontextmanager + async def post(self, url, json=None, headers=None): + self._recorder.append({"url": url, "json": json, "headers": headers}) + yield _FakeResponse(self._payload) + + +def _make_session_factory(payload, recorder): + def factory(*args, **kwargs): + return _FakeSession(payload, recorder) + + return factory + + +def test_crw_search_async_normalizes_envelope(monkeypatch): + """crw_search_async maps the fastCRW envelope to a Tavily-like shape.""" + monkeypatch.setenv("CRW_API_KEY", "test-key") + monkeypatch.delenv("CRW_API_URL", raising=False) + monkeypatch.setenv("GET_API_KEYS_FROM_CONFIG", "false") + + payload = { + "success": True, + "data": [ + { + "title": "Example", + "url": "https://example.com", + "description": "An example result", + "markdown": "# Example\n\nBody content", + } + ], + } + recorder = [] + + with patch( + "open_deep_research.utils.aiohttp.ClientSession", + _make_session_factory(payload, recorder), + ): + results = asyncio.run( + crw_search_async(["hello world"], max_results=3, config=None) + ) + + assert len(results) == 1 + assert results[0]["query"] == "hello world" + assert results[0]["results"] == [ + { + "title": "Example", + "url": "https://example.com", + "content": "An example result", + "raw_content": "# Example\n\nBody content", + } + ] + + # Verify the request hit /v1/search with Bearer auth and the right body + assert len(recorder) == 1 + call = recorder[0] + assert call["url"] == f"{CRW_DEFAULT_BASE_URL}/v1/search" + assert call["json"]["query"] == "hello world" + assert call["json"]["limit"] == 3 + assert call["headers"]["Authorization"] == "Bearer test-key" + + +def test_crw_search_async_omits_auth_when_no_key(monkeypatch): + """Self-hosted fastCRW without a key sends no Authorization header.""" + monkeypatch.delenv("CRW_API_KEY", raising=False) + monkeypatch.setenv("CRW_API_URL", "http://localhost:3000") + monkeypatch.setenv("GET_API_KEYS_FROM_CONFIG", "false") + + payload = {"success": True, "data": []} + recorder = [] + + with patch( + "open_deep_research.utils.aiohttp.ClientSession", + _make_session_factory(payload, recorder), + ): + results = asyncio.run(crw_search_async(["q"], config=None)) + + assert results[0]["results"] == [] + call = recorder[0] + assert call["url"] == "http://localhost:3000/v1/search" + assert "Authorization" not in call["headers"] + + +def test_get_crw_api_key_from_env(monkeypatch): + """The fastCRW key is read from CRW_API_KEY by default.""" + monkeypatch.setenv("GET_API_KEYS_FROM_CONFIG", "false") + monkeypatch.setenv("CRW_API_KEY", "abc123") + assert get_crw_api_key(None) == "abc123" + + +def test_get_crw_base_url_defaults_to_cloud(monkeypatch): + """The base URL defaults to the managed cloud when unset.""" + monkeypatch.setenv("GET_API_KEYS_FROM_CONFIG", "false") + monkeypatch.delenv("CRW_API_URL", raising=False) + assert get_crw_base_url(None) == CRW_DEFAULT_BASE_URL + + +def test_get_search_tool_returns_crw_tool(): + """The dispatch returns the crw_search tool for SearchAPI.CRW.""" + tools = asyncio.run(get_search_tool(SearchAPI.CRW)) + assert len(tools) == 1 + assert tools[0].name == "crw_search" + assert tools[0].metadata["name"] == "web_search"