Skills › Scientific Agent Skills › exa-search › exa_extract.py
exa_extract.py
136 lines · skills/exa-search/scripts/exa_extract.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"""Fetch and extract content from URLs using Exa's /contents endpoint.
8Example:
9 uv run exa_extract.py \\
10 https://arxiv.org/abs/2401.04088 \\
11 https://www.nature.com/articles/s41586-024-07566-y \\
12 --text \\
13 -o extracted.json
14"""
15from __future__ import annotations
17import argparse
18import json
19import os
20import sys
21from dataclasses import asdict, dataclass, field
22from typing import Any
24try:
25 from exa_py import Exa
26except ImportError:
27 print(
28 "exa_py not installed. Run: uv pip install exa-py (or invoke with: uv run --with exa-py)",
29 file=sys.stderr,
30 )
31 sys.exit(2)
34EXA_INTEGRATION_HEADER = "k-dense-ai--scientific-agent-skills"
37@dataclass
38class ExtractedDocument:
39 """Typed view of a single extracted document for JSON export."""
41 url: str
42 id: str | None
43 title: str | None
44 author: str | None
45 published_date: str | None
46 text: str | None = None
47 highlights: list[str] = field(default_factory=list)
50def _build_contents(text: bool, highlights: bool) -> dict[str, Any]:
51 contents: dict[str, Any] = {}
52 if text:
53 contents["text"] = True
54 if highlights:
55 contents["highlights"] = True
56 if not text:
57 # get_contents otherwise inserts default text even with highlights=True.
58 contents["text"] = False
59 if not contents:
60 # Default to full text when the caller doesn't pick anything.
61 contents["text"] = True
62 return contents
65def _to_typed(item: Any) -> ExtractedDocument:
66 return ExtractedDocument(
67 url=getattr(item, "url", ""),
68 id=getattr(item, "id", None),
69 title=getattr(item, "title", None),
70 author=getattr(item, "author", None),
71 published_date=getattr(item, "published_date", None),
72 text=getattr(item, "text", None),
73 highlights=list(getattr(item, "highlights", None) or []),
74 )
77def run(args: argparse.Namespace) -> dict[str, Any]:
78 if not 1 <= len(args.urls) <= 100:
79 raise ValueError("Supply 1-100 URLs; split larger lists into separate calls.")
80 if args.max_age_hours is not None and not -1 <= args.max_age_hours <= 720:
81 raise ValueError("--max-age-hours must be between -1 and 720.")
82 api_key = os.environ.get("EXA_API_KEY")
83 if not api_key:
84 print("EXA_API_KEY environment variable is not set.", file=sys.stderr)
85 sys.exit(2)
87 client = Exa(api_key=api_key)
88 client.headers["x-exa-integration"] = EXA_INTEGRATION_HEADER
90 contents = _build_contents(args.text, args.highlights)
91 if args.max_age_hours is not None:
92 contents["max_age_hours"] = args.max_age_hours
93 response = client.get_contents(urls=args.urls, **contents)
95 typed = [_to_typed(item) for item in getattr(response, "results", []) or []]
96 return {
97 "urls": list(args.urls),
98 "num_results": len(typed),
99 "results": [asdict(doc) for doc in typed],
100 "statuses": [
101 {key: getattr(status, key, None) for key in ("id", "status", "source")}
102 for status in getattr(response, "statuses", None) or []
103 ],
104 }
107def build_parser() -> argparse.ArgumentParser:
108 parser = argparse.ArgumentParser(description="Extract content from URLs with Exa.")
109 parser.add_argument("urls", nargs="+", help="One or more URLs to extract.")
110 parser.add_argument("--text", action="store_true", help="Return full-text content.")
111 parser.add_argument("--highlights", action="store_true", help="Return extracted highlight snippets.")
112 parser.add_argument("--max-age-hours", type=int, default=None,
113 help="Cache age limit (-1 to 720): 0 fetches fresh; -1 uses cache only; omitted uses API fallback policy.")
114 parser.add_argument("-o", "--output", default=None, help="Write JSON to this file (default: stdout).")
115 return parser
118def main(argv: list[str] | None = None) -> int:
119 parser = build_parser()
120 args = parser.parse_args(argv)
121 try:
122 payload = run(args)
123 except ValueError as exc:
124 parser.error(str(exc))
125 text = json.dumps(payload, indent=2, ensure_ascii=False)
126 if args.output:
127 with open(args.output, "w", encoding="utf-8") as fh:
128 fh.write(text)
129 print(f"Wrote {len(payload['results'])} documents to {args.output}")
130 else:
131 print(text)
132 return 0
135if __name__ == "__main__":
136 sys.exit(main())