Skills › Scientific Agent Skills › exa-search › exa_search.py
exa_search.py
181 lines · skills/exa-search/scripts/exa_search.py at 5f2bc03
1#!/usr/bin/env python3
2# /// script
3# requires-python = ">=3.11"
4# dependencies = ["exa-py>=2.23.0,<3"]
5# ///
6"""Run an Exa web search and write results to JSON.
8Uses the Exa Python SDK. Auth via the EXA_API_KEY environment variable.
10Example:
11 uv run exa_search.py "transformer architectures" \\
12 --category publication \\
13 --text --highlights \\
14 -o results.json
15"""
16from __future__ import annotations
18import argparse
19import json
20import os
21import sys
22from dataclasses import asdict, dataclass, field
23from typing import Any
25try:
26 from exa_py import Exa
27except ImportError:
28 print(
29 "exa_py not installed. Run: uv pip install exa-py (or invoke with: uv run --with exa-py)",
30 file=sys.stderr,
31 )
32 sys.exit(2)
35EXA_INTEGRATION_HEADER = "k-dense-ai--scientific-agent-skills"
38@dataclass
39class SearchResult:
40 """Typed view of a single Exa search result for JSON export."""
42 title: str | None
43 url: str
44 id: str | None
45 author: str | None
46 published_date: str | None
47 score: float | None
48 text: str | None = None
49 highlights: list[str] = field(default_factory=list)
50 highlight_scores: list[float] = field(default_factory=list)
53def _split_csv(value: str | None) -> list[str] | None:
54 if not value:
55 return None
56 items = [item.strip() for item in value.split(",") if item.strip()]
57 return items or None
60def _build_contents(text: bool, highlights: bool) -> dict[str, Any] | None:
61 contents: dict[str, Any] = {}
62 if text:
63 contents["text"] = True
64 if highlights:
65 contents["highlights"] = True
66 return contents or None
69def _result_to_typed(item: Any) -> SearchResult:
70 highlights = list(getattr(item, "highlights", None) or [])
71 scores = list(getattr(item, "highlight_scores", None) or [])
72 return SearchResult(
73 title=getattr(item, "title", None),
74 url=getattr(item, "url", ""),
75 id=getattr(item, "id", None),
76 author=getattr(item, "author", None),
77 published_date=getattr(item, "published_date", None),
78 score=getattr(item, "score", None),
79 text=getattr(item, "text", None),
80 highlights=highlights,
81 highlight_scores=scores,
82 )
85def run(args: argparse.Namespace) -> dict[str, Any]:
86 if not args.query.strip():
87 raise ValueError("The query must not be empty.")
88 if not 1 <= args.num_results <= 100:
89 raise ValueError("--num-results must be between 1 and 100.")
90 category = "publication" if args.category == "research paper" else args.category
91 if category in {"company", "people"} and (
92 args.exclude_domains or args.start_published_date or args.end_published_date
93 ):
94 raise ValueError("company/people do not support exclude-domains or publication-date filters.")
95 api_key = os.environ.get("EXA_API_KEY")
96 if not api_key:
97 print("EXA_API_KEY environment variable is not set.", file=sys.stderr)
98 sys.exit(2)
100 client = Exa(api_key=api_key)
101 # Attribute API usage to this skill for integration tracking.
102 client.headers["x-exa-integration"] = EXA_INTEGRATION_HEADER
104 contents = _build_contents(args.text, args.highlights)
106 kwargs: dict[str, Any] = {
107 "query": args.query,
108 "num_results": args.num_results,
109 "type": args.type,
110 }
111 if category:
112 kwargs["category"] = category
113 if include := _split_csv(args.include_domains):
114 kwargs["include_domains"] = include
115 if exclude := _split_csv(args.exclude_domains):
116 kwargs["exclude_domains"] = exclude
117 if args.start_published_date:
118 kwargs["start_published_date"] = args.start_published_date
119 if args.end_published_date:
120 kwargs["end_published_date"] = args.end_published_date
121 if args.user_location:
122 kwargs["user_location"] = args.user_location
124 # SDK 2.x retrieves text by default; preserve this CLI's explicit content flags.
125 response = client.search(**kwargs, contents=contents if contents is not None else False)
127 typed = [_result_to_typed(item) for item in getattr(response, "results", []) or []]
128 return {
129 "query": args.query,
130 "type": args.type,
131 "num_results": len(typed),
132 "autoprompt_string": getattr(response, "autoprompt_string", None),
133 "results": [asdict(result) for result in typed],
134 }
137def build_parser() -> argparse.ArgumentParser:
138 parser = argparse.ArgumentParser(description="Search the web with Exa.")
139 parser.add_argument("query", help="Natural-language search query.")
140 parser.add_argument(
141 "--type",
142 default="auto",
143 choices=["auto", "fast", "instant", "deep-lite", "deep", "deep-reasoning"],
144 help="Search mode: auto balances speed/quality; instant minimizes latency; deep variants perform multi-step research.",
145 )
146 parser.add_argument("--num-results", type=int, default=10, help="Number of results (1-100).")
147 parser.add_argument(
148 "--category",
149 default=None,
150 help="Category (publication, company, news, personal site, financial report, people) or custom hint. Legacy 'research paper' maps to publication.",
151 )
152 parser.add_argument("--include-domains", default=None, help="Comma-separated allowlist.")
153 parser.add_argument("--exclude-domains", default=None, help="Comma-separated blocklist.")
154 parser.add_argument("--start-published-date", default=None, help="ISO 8601 timestamp, e.g. 2026-01-01T00:00:00Z.")
155 parser.add_argument("--end-published-date", default=None, help="ISO 8601 timestamp, e.g. 2026-10-01T00:00:00Z.")
156 parser.add_argument("--user-location", default=None, help="Two-letter ISO country code.")
157 parser.add_argument("--text", action="store_true", help="Return full-text content per result.")
158 parser.add_argument("--highlights", action="store_true", help="Return extracted highlight snippets.")
159 parser.add_argument("-o", "--output", default=None, help="Write JSON to this file (default: stdout).")
160 return parser
163def main(argv: list[str] | None = None) -> int:
164 parser = build_parser()
165 args = parser.parse_args(argv)
166 try:
167 payload = run(args)
168 except ValueError as exc:
169 parser.error(str(exc))
170 text = json.dumps(payload, indent=2, ensure_ascii=False)
171 if args.output:
172 with open(args.output, "w", encoding="utf-8") as fh:
173 fh.write(text)
174 print(f"Wrote {len(payload['results'])} results to {args.output}")
175 else:
176 print(text)
177 return 0
180if __name__ == "__main__":
181 sys.exit(main())