| """Exa API client for immigration-focused web search."""
|
|
|
| from __future__ import annotations
|
|
|
| import os
|
| from functools import lru_cache
|
| from typing import Any
|
|
|
| from exa_py import Exa
|
|
|
| from apis.official_sources import classify_source, infer_official_domains
|
|
|
| HIGHLIGHT_CHAR_LIMIT = 2000
|
| SUMMARY_CHAR_LIMIT = 500
|
| DEFAULT_NUM_RESULTS = 8
|
| MAX_NUM_RESULTS = 15
|
|
|
|
|
| def _truncate(text: str | None, limit: int) -> str:
|
| if not text:
|
| return ""
|
| if len(text) <= limit:
|
| return text
|
| return text[:limit] + "\n… (truncated)"
|
|
|
|
|
| @lru_cache(maxsize=1)
|
| def _client() -> Exa:
|
| api_key = os.environ.get("EXA_API_KEY")
|
| if not api_key:
|
| raise RuntimeError("EXA_API_KEY is not set")
|
| return Exa(api_key=api_key)
|
|
|
|
|
| def _result_dict(result: Any) -> dict[str, Any]:
|
| highlights = result.highlights or []
|
| trimmed_highlights = [
|
| _truncate(highlight, HIGHLIGHT_CHAR_LIMIT) for highlight in highlights
|
| ]
|
| summary = getattr(result, "summary", None)
|
| source_quality = classify_source(result.url or "")
|
| return {
|
| "title": result.title or "",
|
| "url": result.url or "",
|
| "published_date": result.published_date,
|
| "score": result.score,
|
| "source_quality": source_quality,
|
| "highlights": trimmed_highlights,
|
| "summary": _truncate(summary, SUMMARY_CHAR_LIMIT) if summary else None,
|
| }
|
|
|
|
|
| def search_immigration(
|
| query: str,
|
| *,
|
| num_results: int = DEFAULT_NUM_RESULTS,
|
| country: str | None = None,
|
| include_domains: list[str] | None = None,
|
| ) -> dict[str, Any]:
|
| """
|
| Search the web for immigration-related information using Exa.
|
|
|
| Returns result titles, URLs, and query-relevant highlights so the agent
|
| can follow up with Firecrawl for full official page content.
|
|
|
| Args:
|
| query: Natural-language search query describing the ideal page.
|
| num_results: Number of results to return (capped at 15).
|
| country: Optional ISO-2 country code for geo-targeted results.
|
| include_domains: Optional list of domains to restrict results to.
|
| """
|
| normalized_query = (query or "").strip()
|
| if not normalized_query:
|
| return {"error": "query is required", "query": query}
|
|
|
| capped_results = max(1, min(num_results, MAX_NUM_RESULTS))
|
|
|
| try:
|
| kwargs: dict[str, Any] = {
|
| "num_results": capped_results,
|
| "type": "auto",
|
| "contents": {
|
| "highlights": {"maxCharacters": HIGHLIGHT_CHAR_LIMIT},
|
| },
|
| }
|
|
|
| if country:
|
| normalized_country = country.strip().upper()
|
| if normalized_country:
|
| kwargs["user_location"] = normalized_country
|
|
|
| if include_domains:
|
| kwargs["include_domains"] = include_domains
|
|
|
| official_domain_hints = infer_official_domains(normalized_query, country)
|
|
|
| response = _client().search(normalized_query, **kwargs)
|
| results = [_result_dict(result) for result in (response.results or [])]
|
| return {
|
| "query": normalized_query,
|
| "num_results": len(results),
|
| "official_domain_hints": official_domain_hints,
|
| "official_results": sum(
|
| 1
|
| for result in results
|
| if result.get("source_quality", {}).get("is_official")
|
| ),
|
| "results": results,
|
| }
|
| except Exception as exc:
|
| return {"error": str(exc), "query": normalized_query}
|
|
|
