Skip to content

tools

zocrates.agents.online_search.tools

Tools for the Online Literature Discovery Subagent.

CANDIDATE_PAPERS module-attribute 

CANDIDATE_PAPERS = 'candidates'

INITIAL_ONLINE_TOOL_BUDGET module-attribute 

INITIAL_ONLINE_TOOL_BUDGET = 8

MAX_ONLINE_EXTENSIONS module-attribute 

MAX_ONLINE_EXTENSIONS = 2

ONLINE_BUDGET_EXTENSION_INCREMENT module-attribute 

ONLINE_BUDGET_EXTENSION_INCREMENT = 5

TEMP_EXTENSION_COUNT module-attribute 

TEMP_EXTENSION_COUNT = 'temp:extension_count'

TEMP_EXTENSION_REQUESTED module-attribute 

TEMP_EXTENSION_REQUESTED = 'temp:extension_requested'

TEMP_EXTENSION_REQUEST_PASS module-attribute 

TEMP_EXTENSION_REQUEST_PASS = 'temp:extension_request_pass'

TEMP_ONLINE_CANDIDATES module-attribute 

TEMP_ONLINE_CANDIDATES = 'temp_online_candidates'

TEMP_RETRIEVAL_TOOL_BUDGET module-attribute 

TEMP_RETRIEVAL_TOOL_BUDGET = 'temp:retrieval_tool_budget'

TEMP_RETRIEVAL_TOOL_COUNT module-attribute 

TEMP_RETRIEVAL_TOOL_COUNT = 'temp:retrieval_tool_count'

TEMP_SEARCH_REPORT module-attribute 

TEMP_SEARCH_REPORT = 'temp:search_report'

logger module-attribute 

logger = logging.getLogger(__name__)

CandidatePaper dataclass 

CandidatePaper(candidate_id, title, authors=list(), year=None, venue=None, doi=None, arxiv_id=None, openalex_id='', abstract=None, tldr=None, is_oa=False, oa_pdf_url=None, oa_pdf_urls=list(), citation_count=0, primary_topic=None, keywords=list(), in_library=False)

Represents a discovered scholarly paper candidate.

from_dict classmethod 

from_dict(data)

Deserialize CandidatePaper from a dictionary.

Source code in zocrates/discovery/models.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "CandidatePaper":
    """Deserialize CandidatePaper from a dictionary."""
    citation_count_val = data.get("citation_count", 0)
    try:
        citation_count = int(citation_count_val) if citation_count_val is not None else 0
    except (ValueError, TypeError):
        citation_count = 0

    year_val = data.get("year")
    year: int | None = None
    if year_val is not None:
        try:
            year = int(year_val)
        except (ValueError, TypeError):
            year = None

    raw_authors = data.get("authors")
    if isinstance(raw_authors, list):
        authors = [str(a).strip() for a in raw_authors if a and str(a).strip()]
    elif isinstance(raw_authors, str) and raw_authors.strip():
        authors = [raw_authors.strip()]
    else:
        authors = []

    raw_keywords = data.get("keywords")
    if isinstance(raw_keywords, list):
        keywords = [str(k).strip() for k in raw_keywords if k and str(k).strip()]
    elif isinstance(raw_keywords, str) and raw_keywords.strip():
        keywords = [raw_keywords.strip()]
    else:
        keywords = []

    def _clean_opt_str(val: Any) -> str | None:
        """Clean and trim an optional string or scalar value.

        Args:
            val: Input value to convert to a stripped string.

        Returns:
            Stripped string, or None if empty or invalid.
        """
        if val is not None and isinstance(val, (str, int, float)):
            s = str(val).strip()
            return s if s else None
        return None

    raw_urls = data.get("oa_pdf_urls")
    oa_pdf_urls: list[str] = []
    if isinstance(raw_urls, list):
        oa_pdf_urls = [str(u).strip() for u in raw_urls if u and str(u).strip()]
    primary_pdf = _clean_opt_str(data.get("oa_pdf_url"))
    if primary_pdf and primary_pdf not in oa_pdf_urls:
        oa_pdf_urls.insert(0, primary_pdf)

    return cls(
        candidate_id=str(data.get("candidate_id") or ""),
        title=str(data.get("title") or "").strip(),
        authors=authors,
        year=year,
        venue=_clean_opt_str(data.get("venue")),
        doi=_clean_opt_str(data.get("doi")),
        arxiv_id=_clean_opt_str(data.get("arxiv_id")),
        openalex_id=str(data.get("openalex_id") or ""),
        abstract=_clean_opt_str(data.get("abstract")),
        tldr=_clean_opt_str(data.get("tldr")),
        is_oa=bool(data.get("is_oa", False)),
        oa_pdf_url=primary_pdf,
        oa_pdf_urls=oa_pdf_urls,
        citation_count=citation_count,
        primary_topic=_clean_opt_str(data.get("primary_topic")),
        keywords=keywords,
        in_library=bool(data.get("in_library", False)),
    )

to_dict 

to_dict()

Serialize CandidatePaper to a plain dictionary.

Source code in zocrates/discovery/models.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def to_dict(self) -> dict[str, Any]:
    """Serialize CandidatePaper to a plain dictionary."""
    return {
        "candidate_id": self.candidate_id,
        "title": self.title,
        "authors": list(self.authors),
        "year": self.year,
        "venue": self.venue,
        "doi": self.doi,
        "arxiv_id": self.arxiv_id,
        "openalex_id": self.openalex_id,
        "abstract": self.abstract,
        "tldr": self.tldr,
        "is_oa": self.is_oa,
        "oa_pdf_url": self.oa_pdf_url,
        "oa_pdf_urls": list(self.oa_pdf_urls),
        "citation_count": self.citation_count,
        "primary_topic": self.primary_topic,
        "keywords": list(self.keywords),
        "in_library": self.in_library,
    }

CandidateRecommendation

Bases: BaseModel

Structured paper recommendation with rationale for import.

DiscoveryEngine 

DiscoveryEngine(provider=None, deduplicator=None, abstract_fallback=None, unpaywall=None, email=DEFAULT_POLITE_EMAIL)

Coordinates scholarly search across providers, enforces strict novelty against the local Zotero library, backfills abstracts, and resolves Open Access PDFs.

Parameters:

Name Type Description Default
provider SearchProvider | None

SearchProvider instance (defaults to OpenAlexProvider).

 None
deduplicator Deduplicator | None

Deduplicator instance with loaded library state.

 None
abstract_fallback AbstractFallback | None

AbstractFallback instance for missing abstracts.

 None
unpaywall UnpaywallResolver | None

UnpaywallResolver instance for OA PDF links.

 None
email str

Polite pool email for upstream academic APIs.

 DEFAULT_POLITE_EMAIL
Source code in zocrates/discovery/engine.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
def __init__(
    self,
    provider: SearchProvider | None = None,
    deduplicator: Deduplicator | None = None,
    abstract_fallback: AbstractFallback | None = None,
    unpaywall: UnpaywallResolver | None = None,
    email: str = DEFAULT_POLITE_EMAIL,
) -> None:
    """
    Args:
        provider: SearchProvider instance (defaults to OpenAlexProvider).
        deduplicator: Deduplicator instance with loaded library state.
        abstract_fallback: AbstractFallback instance for missing abstracts.
        unpaywall: UnpaywallResolver instance for OA PDF links.
        email: Polite pool email for upstream academic APIs.
    """
    self.email = email
    self.provider = provider or OpenAlexProvider(email=email)
    self.deduplicator = deduplicator or Deduplicator()
    self.abstract_fallback = abstract_fallback or AbstractFallback(email=email)
    self.unpaywall = unpaywall or UnpaywallResolver(email=email)

resolve_identifier async 

resolve_identifier(identifier)

Resolve a single paper by DOI, arXiv ID, or OpenAlex ID and check its library presence.

Parameters:

Name Type Description Default
identifier str

DOI, arXiv ID, or OpenAlex ID string.

 required

Returns:

Type Description
tuple[CandidatePaper | None, bool]

Tuple of (CandidatePaper or None, in_library boolean).
Source code in zocrates/discovery/engine.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
async def resolve_identifier(self, identifier: str) -> tuple[CandidatePaper | None, bool]:
    """
    Resolve a single paper by DOI, arXiv ID, or OpenAlex ID and check its library presence.

    Args:
        identifier: DOI, arXiv ID, or OpenAlex ID string.

    Returns:
        Tuple of (CandidatePaper or None, in_library boolean).
    """
    ident = identifier.strip() if identifier else ""
    if not ident:
        return None, False

    try:
        candidate = await self.provider.resolve_identifier(ident)
    except Exception as e:
        logger.warning("Failed resolving identifier %r: %s", ident, e)
        return None, False

    if candidate is None:
        return None, False

    # Check library presence
    in_library = self.deduplicator.is_in_library(candidate)
    candidate.in_library = in_library

    # Only perform external network lookups if paper is not already in the library
    if not in_library:
        # Backfill abstract if missing
        if not candidate.abstract:
            try:
                candidate = await self.abstract_fallback.backfill_abstract(candidate)
            except Exception as e:
                logger.debug("Abstract backfill failed for %r: %s", ident, e)

        # Resolve OA PDF if missing
        if not candidate.oa_pdf_url and candidate.doi:
            try:
                pdf_url = await self.unpaywall.resolve_pdf(candidate.doi)
                if pdf_url:
                    candidate.oa_pdf_url = pdf_url
                    candidate.is_oa = True
            except Exception as e:
                logger.debug("Unpaywall PDF resolution failed for %r: %s", ident, e)

    return candidate, in_library

search async 

search(query, limit=10, year_from=None, year_to=None, is_oa_only=False)

Search for scholarly literature, filtering out papers already in the library and enriching top results with abstracts and OA PDF URLs.

Parameters:

Name Type Description Default
query str

Free-text search terms or conceptual query.

 required
limit int

Target number of novel candidate papers to return.

 10
year_from int | None

Optional starting publication year.

 None
year_to int | None

Optional ending publication year.

 None
is_oa_only bool

Restrict results to open-access papers.

 False

Returns:

Type Description
list[CandidatePaper]

List of novel CandidatePaper instances (at most limit).
Source code in zocrates/discovery/engine.py 
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
async def search(
    self,
    query: str,
    limit: int = 10,
    year_from: int | None = None,
    year_to: int | None = None,
    is_oa_only: bool = False,
) -> list[CandidatePaper]:
    """
    Search for scholarly literature, filtering out papers already in the library
    and enriching top results with abstracts and OA PDF URLs.

    Args:
        query: Free-text search terms or conceptual query.
        limit: Target number of novel candidate papers to return.
        year_from: Optional starting publication year.
        year_to: Optional ending publication year.
        is_oa_only: Restrict results to open-access papers.

    Returns:
        List of novel CandidatePaper instances (at most `limit`).
    """
    q = query.strip() if query else ""
    if not q:
        return []

    # Over-fetch from provider to account for library deduplication
    fetch_limit = min(max(limit * 2, 20), 100)
    try:
        raw_candidates = await self.provider.search(
            query=q,
            limit=fetch_limit,
            year_from=year_from,
            year_to=year_to,
            is_oa_only=is_oa_only,
        )
    except Exception as e:
        logger.warning("Provider search failed for query %r: %s", q, e)
        return []

    if not raw_candidates:
        return []

    # Filter strictly novel papers against local library and batch duplicates
    novel_candidates = self.deduplicator.filter_novel(raw_candidates)

    # Slice to requested limit
    selected = novel_candidates[:limit]
    if not selected:
        return []

    # Opportunistic abstract backfill for selected candidates lacking abstracts
    try:
        selected = await self.abstract_fallback.backfill_batch(selected)
    except Exception as e:
        logger.debug("Abstract backfill encountered error: %s", e)

    # Resolve OA PDF URLs via Unpaywall for OA candidates missing direct PDF URLs
    async def _resolve_oa_pdf(c: CandidatePaper) -> None:
        """Resolve Open Access PDF URL via Unpaywall for a candidate paper.

        Args:
            c: CandidatePaper to inspect and update with resolved PDF URL.
        """
        if not c.oa_pdf_url and c.doi:
            try:
                pdf_url = await self.unpaywall.resolve_pdf(c.doi)
                if pdf_url:
                    c.oa_pdf_url = pdf_url
                    c.is_oa = True
            except Exception as ex:
                logger.debug("Unpaywall resolution failed for DOI %s: %s", c.doi, ex)

    pdf_tasks = [_resolve_oa_pdf(c) for c in selected if not c.oa_pdf_url and c.doi]
    if pdf_tasks:
        await asyncio.gather(*pdf_tasks, return_exceptions=True)

    return selected

_check_and_increment_online_budget 

_check_and_increment_online_budget(tool_context)

Check and update tool call budget for online retrieval.

Source code in zocrates/agents/online_search/tools.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
def _check_and_increment_online_budget(
    tool_context: ToolContext | None,
) -> tuple[bool, str | None, str | None]:
    """Check and update tool call budget for online retrieval."""
    if tool_context is None:
        return True, None, None

    state = tool_context.state
    raw_budget = (
        state.get(TEMP_RETRIEVAL_TOOL_BUDGET)
        if state.get(TEMP_RETRIEVAL_TOOL_BUDGET) is not None
        else state.get("temp_retrieval_tool_budget")
    )
    budget = int(raw_budget) if raw_budget is not None else INITIAL_ONLINE_TOOL_BUDGET

    raw_count = (
        state.get(TEMP_RETRIEVAL_TOOL_COUNT)
        if state.get(TEMP_RETRIEVAL_TOOL_COUNT) is not None
        else state.get("temp_retrieval_tool_count")
    )
    count = int(raw_count) if raw_count is not None else 0

    extension_count = int(
        state.get(TEMP_EXTENSION_COUNT)
        if state.get(TEMP_EXTENSION_COUNT) is not None
        else state.get("temp_extension_count", 0)
    )
    can_extend = extension_count < MAX_ONLINE_EXTENSIONS

    if count >= budget:
        action_msg = (
            "Call `request_search_extension` to request an extension, or `conclude_online_search` to finish."
            if can_extend
            else "Call `conclude_online_search` immediately with your recommendations to finish."
        )
        return (
            False,
            f"Retrieval tool budget reached ({budget}/{budget}). You cannot call more search tools. {action_msg}",
            None,
        )

    new_count = count + 1
    state[TEMP_RETRIEVAL_TOOL_COUNT] = new_count
    state["temp_retrieval_tool_count"] = new_count

    notice: str | None = None
    if new_count == budget:
        action_msg = (
            "Call `request_search_extension` to request an extension, or `conclude_online_search` to finish."
            if can_extend
            else "You must call `conclude_online_search` now to conclude retrieval."
        )
        notice = f"[Notice: Retrieval tool budget exhausted ({budget}/{budget}). {action_msg}]"
    elif new_count == (budget - 1):
        notice = (
            f"[Notice: {new_count}/{budget} retrieval tool calls used (1 call remaining). "
            f"Plan your final query, request an extension via `request_search_extension`, or prepare to call `conclude_online_search`.]"
        )
    elif new_count >= int(budget / 2):
        notice = (
            f"[Notice: {new_count}/{budget} retrieval tool calls used. "
            f"Continue your search, being mindful of your remaining retrieval budget.]"
        )

    return True, None, notice

_resolve_candidate_from_pool 

_resolve_candidate_from_pool(cid_raw, pool)

Resolve a candidate identifier to a CandidatePaper from the pool.

Supports multi-tier resolution: 1. Direct exact key match (cid in pool). 2. Short ID match: cid matches the suffix of a pool key (e.g., 'W3128491823' matches 'https://openalex.org/W3128491823'). 3. Ordinal / Index match: if cid is 'c1', '1', 1, etc., extract integer N and map to N-1 index of pool values. 4. Direct identifier match: cid matches cand.doi or cand.arxiv_id.

Source code in zocrates/agents/online_search/tools.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
def _resolve_candidate_from_pool(
    cid_raw: Any,
    pool: dict[str, Any],
) -> CandidatePaper | None:
    """Resolve a candidate identifier to a CandidatePaper from the pool.

    Supports multi-tier resolution:
    1. Direct exact key match (`cid in pool`).
    2. Short ID match: `cid` matches the suffix of a pool key (e.g., 'W3128491823' matches 'https://openalex.org/W3128491823').
    3. Ordinal / Index match: if `cid` is 'c1', '1', 1, etc., extract integer N and map to N-1 index of pool values.
    4. Direct identifier match: `cid` matches `cand.doi` or `cand.arxiv_id`.
    """
    if cid_raw is None or not pool:
        return None

    def _to_candidate(val: Any) -> CandidatePaper | None:
        """Coerce a candidate pool entry into a CandidatePaper instance.

        Args:
            val: CandidatePaper instance or dictionary representation.

        Returns:
            CandidatePaper instance, or None if conversion fails.
        """
        if isinstance(val, CandidatePaper):
            return val
        if isinstance(val, dict):
            try:
                return CandidatePaper.from_dict(val)
            except Exception:
                return None
        return None

    # Tier 1: Direct exact key match
    if cid_raw in pool:
        cand = _to_candidate(pool[cid_raw])
        if cand is not None:
            return cand

    cid_str = str(cid_raw).strip()
    if not cid_str:
        return None

    if cid_str in pool:
        cand = _to_candidate(pool[cid_str])
        if cand is not None:
            return cand

    # Tier 2: Short ID match (suffix of a pool key or openalex_id)
    cid_lower = cid_str.lower()
    for key, val in pool.items():
        key_str = str(key).strip()
        key_suffix = key_str.rstrip("/").split("/")[-1].lower()
        if key_suffix == cid_lower or key_str.lower().endswith("/" + cid_lower):
            cand = _to_candidate(val)
            if cand is not None:
                return cand
        cand = _to_candidate(val)
        if cand is not None and cand.openalex_id:
            oa_suffix = cand.openalex_id.rstrip("/").split("/")[-1].lower()
            if oa_suffix == cid_lower or cand.openalex_id.lower() == cid_lower:
                return cand

    # Tier 3: Ordinal / Index match (e.g. 1, "1", "c1", "C1", "candidate 1")
    ordinal_num: int | None = None
    if isinstance(cid_raw, int):
        ordinal_num = cid_raw
    elif cid_str.isdigit():
        ordinal_num = int(cid_str)
    else:
        m = re.match(r"^(?:c|candidate)[\s_-]*(\d+)$", cid_str, re.IGNORECASE)
        if m:
            ordinal_num = int(m.group(1))

    if ordinal_num is not None:
        pool_values = list(pool.values())
        if 1 <= ordinal_num <= len(pool_values):
            cand = _to_candidate(pool_values[ordinal_num - 1])
            if cand is not None:
                return cand

    # Tier 4: Direct identifier match (DOI or arXiv ID)
    norm_cid = cid_lower
    for prefix in (
        "https://doi.org/",
        "http://doi.org/",
        "doi:",
        "https://arxiv.org/abs/",
        "https://arxiv.org/pdf/",
        "arxiv:",
    ):
        if norm_cid.startswith(prefix):
            norm_cid = norm_cid[len(prefix) :]
            break

    for val in pool.values():
        cand = _to_candidate(val)
        if cand is None:
            continue
        if cand.doi:
            cand_doi = cand.doi.lower().strip()
            for prefix in ("https://doi.org/", "http://doi.org/", "doi:"):
                if cand_doi.startswith(prefix):
                    cand_doi = cand_doi[len(prefix) :]
                    break
            if cand_doi == norm_cid or cand_doi == cid_lower:
                return cand
        if cand.arxiv_id:
            cand_arxiv = cand.arxiv_id.lower().strip()
            for prefix in (
                "https://arxiv.org/abs/",
                "https://arxiv.org/pdf/",
                "arxiv:",
            ):
                if cand_arxiv.startswith(prefix):
                    cand_arxiv = cand_arxiv[len(prefix) :]
                    break
            if cand_arxiv == norm_cid or cand_arxiv == cid_lower:
                return cand

    return None

conclude_online_search(recommendations, summary=None, tool_context=None)

Conclude the online literature retrieval phase with selected candidate papers.

Call this tool when novel, high-impact candidate papers addressing the user's research inquiry have been identified.

Parameters:

Name Type Description Default
recommendations list[CandidateRecommendation]

Structured list of recommendations pairing candidate_id and rationale.

 required
summary str | None

Brief summary of the literature discovered and why these papers were selected.

 None
tool_context ToolContext | None

ADK ToolContext.

 None

Returns:

Type Description
str

Confirmation message concluding the retrieval phase.
Source code in zocrates/agents/online_search/tools.py 
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
def conclude_online_search(
    recommendations: list[CandidateRecommendation],
    summary: str | None = None,
    tool_context: ToolContext | None = None,
) -> str:
    """Conclude the online literature retrieval phase with selected candidate papers.

    Call this tool when novel, high-impact candidate papers addressing the user's
    research inquiry have been identified.

    Args:
        recommendations: Structured list of recommendations pairing candidate_id and rationale.
        summary: Brief summary of the literature discovered and why these papers were selected.
        tool_context: ADK ToolContext.

    Returns:
        Confirmation message concluding the retrieval phase.
    """
    selected_candidates: list[CandidatePaper] = []
    if tool_context is not None and tool_context.state is not None:
        state = tool_context.state
        pool = (
            state.get(TEMP_ONLINE_CANDIDATES)
            if state.get(TEMP_ONLINE_CANDIDATES) is not None
            else state.get("temp_online_candidates")
        ) or {}
        if not isinstance(pool, dict):
            pool = {}

        seen_ids: set[str] = set()
        for rec in recommendations or []:
            if isinstance(rec, dict):
                cid_raw = rec.get("candidate_id")
                rat = str(rec.get("rationale", "") or "").strip()
            elif hasattr(rec, "candidate_id"):
                cid_raw = getattr(rec, "candidate_id", None)
                rat = str(getattr(rec, "rationale", "") or "").strip()
            else:
                continue

            if cid_raw is None or cid_raw == "":
                continue

            cand_obj = _resolve_candidate_from_pool(cid_raw, pool)
            if cand_obj is not None:
                if cand_obj.candidate_id in seen_ids:
                    continue
                seen_ids.add(cand_obj.candidate_id)
                cand = (
                    CandidatePaper.from_dict(cand_obj.to_dict())
                    if hasattr(cand_obj, "to_dict")
                    else CandidatePaper.from_dict(cand_obj)
                )
                if rat:
                    cand.tldr = rat
                selected_candidates.append(cand)
            else:
                logger.warning("Candidate ID %r not found in online candidate pool.", cid_raw)

        state["selected_candidates"] = selected_candidates
        state[CANDIDATE_PAPERS] = [c.to_dict() for c in selected_candidates]

        report = {
            "summary": summary or f"Selected {len(selected_candidates)} candidate papers.",
            "gaps": [],
            "notes": summary or "",
            "source": "conclude_online_search",
        }
        state[TEMP_SEARCH_REPORT] = report
        state["temp_online_search_report"] = report

        if tool_context.actions:
            tool_context.actions.skip_summarization = True

    count = (
        len(selected_candidates)
        if (tool_context and tool_context.state is not None)
        else len(recommendations or [])
    )
    return (
        f"Online literature search concluded with {count} candidate paper(s) selected. "
        "The Literature Synthesizer will now prepare the recommendations."
    )

get_default_discovery_engine 

get_default_discovery_engine(zotero_data_dir=None, polite_email=None, openalex_api_key=None)

Instantiate a configured DiscoveryEngine with local Zotero deduplication.

Source code in zocrates/discovery/engine.py 
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def get_default_discovery_engine(
    zotero_data_dir: Path | str | None = None,
    polite_email: str | None = None,
    openalex_api_key: str | None = None,
) -> DiscoveryEngine:
    """Instantiate a configured DiscoveryEngine with local Zotero deduplication."""
    from zocrates.discovery.abstract_fallback import AbstractFallback
    from zocrates.discovery.dedup import Deduplicator
    from zocrates.discovery.providers.base import DEFAULT_POLITE_EMAIL
    from zocrates.discovery.providers.openalex import OpenAlexProvider
    from zocrates.discovery.providers.unpaywall import UnpaywallResolver
    from zocrates.config import get_openalex_api_key, get_scholar_polite_email, resolve_zotero_data_dir

    resolved_dir = zotero_data_dir or resolve_zotero_data_dir()
    path_dir = Path(resolved_dir) if resolved_dir else None
    email = polite_email
    if not email:
        try:
            email = get_scholar_polite_email(path_dir)
        except Exception:
            pass
    if not email:
        email = DEFAULT_POLITE_EMAIL

    api_key = openalex_api_key
    if not api_key:
        try:
            api_key = get_openalex_api_key(path_dir)
        except Exception:
            pass

    zotero_db = None
    if resolved_dir:
        sqlite_path = Path(resolved_dir) / "zotero.sqlite"
        storage_dir = Path(resolved_dir) / "storage"
        if sqlite_path.exists():
            try:
                from zocrates.rag.zotero_db import ZoteroDB

                zotero_db = ZoteroDB(sqlite_path=sqlite_path, storage_dir=storage_dir)
            except Exception as e:
                logger.debug("Could not initialize ZoteroDB for discovery deduplication: %s", e)

    deduplicator = Deduplicator(zotero_db=zotero_db)
    provider = OpenAlexProvider(email=email, api_key=api_key)
    abstract_fallback = AbstractFallback(email=email)
    unpaywall = UnpaywallResolver(email=email)

    return DiscoveryEngine(
        provider=provider,
        deduplicator=deduplicator,
        abstract_fallback=abstract_fallback,
        unpaywall=unpaywall,
        email=email,
    )

request_search_extension 

request_search_extension(rationale, covered_summary=None, missing_aspects=None, tool_context=None)

Request permission from the user to extend the online literature retrieval budget.

Call this tool when discovering adjacent subfields or preprints that warrant deeper literature exploration beyond the initial tool budget.

Parameters:

Name Type Description Default
rationale str

Clear explanation of why additional search queries or deeper prospecting are needed.

 required
covered_summary str | None

Optional summary of candidates found so far.

 None
missing_aspects list[str] | None

Optional list of missing topics or subfields to explore.

 None
tool_context ToolContext | None

ADK ToolContext.

 None

Returns:

Type Description
str

Confirmation message confirming extension request.
Source code in zocrates/agents/online_search/tools.py 
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
def request_search_extension(
    rationale: str,
    covered_summary: str | None = None,
    missing_aspects: list[str] | None = None,
    tool_context: ToolContext | None = None,
) -> str:
    """Request permission from the user to extend the online literature retrieval budget.

    Call this tool when discovering adjacent subfields or preprints that warrant deeper
    literature exploration beyond the initial tool budget.

    Args:
        rationale: Clear explanation of why additional search queries or deeper prospecting are needed.
        covered_summary: Optional summary of candidates found so far.
        missing_aspects: Optional list of missing topics or subfields to explore.
        tool_context: ADK ToolContext.

    Returns:
        Confirmation message confirming extension request.
    """
    if tool_context is not None and tool_context.state is not None:
        state = tool_context.state
        state[TEMP_EXTENSION_REQUESTED] = True
        state["temp_extension_requested"] = True
        state["extension_rationale"] = rationale

        ext_num = int(
            state.get(TEMP_EXTENSION_COUNT)
            if state.get(TEMP_EXTENSION_COUNT) is not None
            else state.get("temp_extension_count", 0)
        ) + 1
        state[TEMP_EXTENSION_REQUEST_PASS] = ext_num

        report = {
            "summary": covered_summary or "Preliminary literature discovery pass",
            "gaps": missing_aspects or [rationale],
            "notes": rationale,
            "source": "subagent_request",
        }
        state[TEMP_SEARCH_REPORT] = report
        state["temp_online_search_report"] = report

        if tool_context.actions:
            tool_context.actions.skip_summarization = True

    return "Online search extension requested. Concluding retrieval turn for user confirmation."

search_openalex async 

search_openalex(query, year_from=None, year_to=None, is_oa_only=False, tool_context=None)

Search OpenAlex for external scholarly papers and preprints.

Automatically excludes papers already in your local Zotero library and retrieves essential metadata, abstracts, and open-access PDF links.

Parameters:

Name Type Description Default
query str

Academic search query, keywords, paper title, or exact DOI/arXiv identifier.

 required
year_from int | None

Optional starting publication year (e.g. 2023).

 None
year_to int | None

Optional ending publication year (e.g. 2025).

 None
is_oa_only bool

When True, restricts results strictly to verified open-access works.

 False
tool_context ToolContext | None

ADK ToolContext.

 None

Returns:

Type Description
str

Structured summary of discovered novel candidate papers.
Source code in zocrates/agents/online_search/tools.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
async def search_openalex(
    query: str,
    year_from: int | None = None,
    year_to: int | None = None,
    is_oa_only: bool = False,
    tool_context: ToolContext | None = None,
) -> str:
    """Search OpenAlex for external scholarly papers and preprints.

    Automatically excludes papers already in your local Zotero library and retrieves
    essential metadata, abstracts, and open-access PDF links.

    Args:
        query: Academic search query, keywords, paper title, or exact DOI/arXiv identifier.
        year_from: Optional starting publication year (e.g. 2023).
        year_to: Optional ending publication year (e.g. 2025).
        is_oa_only: When True, restricts results strictly to verified open-access works.
        tool_context: ADK ToolContext.

    Returns:
        Structured summary of discovered novel candidate papers.
    """
    allowed, err_msg, notice = _check_and_increment_online_budget(tool_context)
    if not allowed:
        return err_msg or "Search budget exhausted."

    engine: DiscoveryEngine | None = None
    if tool_context is not None and tool_context.state:
        raw_engine = tool_context.state.get("discovery_engine")
        if hasattr(raw_engine, "search") and callable(getattr(raw_engine, "search")):
            engine = raw_engine
        elif hasattr(tool_context.state, "pop"):
            tool_context.state.pop("discovery_engine", None)
        elif raw_engine is not None:
            tool_context.state["discovery_engine"] = None
    if engine is None:
        engine = get_default_discovery_engine()

    try:
        results: list[CandidatePaper] = await engine.search(
            query=query,
            limit=10,
            year_from=year_from,
            year_to=year_to,
            is_oa_only=is_oa_only,
        )
    except Exception as e:
        logger.warning("search_openalex failed for query %r: %s", query, e)
        return f"Error executing search for query '{query}': {e}"

    # Update candidate pool in state
    if tool_context is not None and tool_context.state is not None:
        state = tool_context.state
        pool = (
            state.get(TEMP_ONLINE_CANDIDATES)
            if state.get(TEMP_ONLINE_CANDIDATES) is not None
            else state.get("temp_online_candidates")
        )
        if not isinstance(pool, dict):
            pool = {}
        for c in results:
            pool[c.candidate_id] = c.to_dict() if hasattr(c, "to_dict") else c
        state[TEMP_ONLINE_CANDIDATES] = pool
        state["temp_online_candidates"] = pool

    if not results:
        msg = (
            f"No novel candidate papers found matching query: '{query}'. "
            "Try refining, broadening, or using alternative technical terminology."
        )
        if notice:
            msg += f"\n\n{notice}"
        return msg

    lines = [f"Found {len(results)} novel candidate paper(s) matching query '{query}':\n"]
    for i, c in enumerate(results, 1):
        authors_str = ", ".join(c.authors[:3]) + (" et al." if len(c.authors) > 3 else "")
        oa_str = "Yes" if c.is_oa else "No"
        venue_str = f" | Venue: {c.venue}" if c.venue else ""
        lines.append(
            f"{i}. [Candidate ID: {c.candidate_id}] {c.title} ({c.year or 'n.d.'})\n"
            f"   Authors: {authors_str or 'Unknown'}{venue_str} | Citations: {c.citation_count} | OA: {oa_str}\n"
            f"   Topic: {c.primary_topic or 'N/A'}"
        )
        abstract_txt = c.abstract or c.tldr
        if abstract_txt:
            snippet = abstract_txt[:300].strip() + ("..." if len(abstract_txt) > 300 else "")
            lines.append(f"   Abstract: {snippet}")
        elif c.keywords:
            lines.append(f"   Keywords: {', '.join(c.keywords[:5])}")
        lines.append("")

    if notice:
        lines.append(notice)

    return "\n".join(lines).strip()