Build internal research tools

import os, requests

SEARCH_TOOL = {
    "name": "search_papers",
    "description": (
        "Search 220M+ peer-reviewed research papers. Use for any question whose answer "
        "should rest on published evidence. Returns papers with title, authors, journal, "
        "year, citation count, DOI, and URL. Cite only papers this tool returned."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "A natural-language research question or topic."},
            "year_min": {"type": "integer", "description": "Exclude papers published before this year."},
            "study_types": {"type": "array", "items": {"type": "string"},
                            "description": "Restrict by design, e.g. rct, meta-analysis, systematic review."},
            "human": {"type": "boolean", "description": "Only include human studies."},
            "sjr_max": {"type": "integer", "description": "Worst journal quartile to include. 1 keeps top-quartile journals only."},
        },
        "required": ["query"],
    },
}

FIELDS = ("title", "authors", "journal_name", "publish_year", "citation_count", "doi", "url", "takeaway")

def search_papers(**args):
    resp = requests.get(
        "https://api.consensus.app/v1/search",
        headers={"x-api-key": os.environ["CONSENSUS_API_KEY"]},
        params=args,
        timeout=30,
    )
    resp.raise_for_status()
    # Trim each result so twenty papers fit comfortably in the model's context.
    return [{k: p.get(k) for k in FIELDS} for p in resp.json()["results"]]

import os, time, requests
from collections import Counter

API = "https://api.consensus.app/v1/search"
HEADERS = {"x-api-key": os.environ["CONSENSUS_API_KEY"]}

WATCHLISTS = {
    "alice": [
        {"query": "vision-language-action models for robot manipulation", "domain": "cs,eng"},
        {"query": "sim-to-real transfer for legged locomotion", "domain": "cs,eng"},
    ],
    "bob": [
        {"query": "event cameras for autonomous driving perception", "domain": "cs"},
    ],
}

def digest(person, since, seen_dois):
    fresh = []
    for watch in WATCHLISTS[person]:
        params = {**watch, "year_min": since.year, "month_min": since.month, "page_size": 100}
        resp = requests.get(API, headers=HEADERS, params=params, timeout=30)
        resp.raise_for_status()
        for paper in resp.json()["results"]:
            doi = paper.get("doi")
            if doi and doi not in seen_dois:
                seen_dois.add(doi)
                fresh.append(paper)
        time.sleep(1)                     # one search per second outside Enterprise
    return fresh

fresh = digest("alice", since=last_run, seen_dois=load_seen("alice"))
top_authors = Counter(a for p in fresh for a in p.get("authors") or [])
top_labs = Counter(i for p in fresh for i in p.get("institutions") or [])

def sweep(query, **filters):
    page, out = 0, []
    while page <= 49:                              # page is 0-indexed; 49 is the ceiling
        resp = requests.get(API, headers=HEADERS, timeout=30, params={
            "query": query, "page": page,
            "page_size": 1000,                     # clamped to your plan's maximum
            **filters,
        })
        resp.raise_for_status()
        body = resp.json()
        out.extend(body["results"])
        if body["is_end"] or body.get("next_page") is None:
            break
        page = body["next_page"]
        time.sleep(1)
    return out

# Everything in one journal since 2020: boost with journal_name, then filter exactly.
papers = sweep("perovskite solar cell stability", journal_name="Nature Energy", year_min=2020)
in_journal = [p for p in papers if p.get("journal_name") == "Nature Energy"]

ROWS = ["PCSK9", "ANGPTL3", "LPA"]

COLUMNS = {
    "human_genetics": dict(
        template="{row} loss-of-function variants cardiovascular outcomes Mendelian randomization",
        human=True,
    ),
    "clinical": dict(
        template="{row} inhibitor randomized controlled trial outcomes",
        study_types=["rct", "meta-analysis"],
    ),
    "safety": dict(
        template="{row} inhibition adverse events safety",
        human=True,
    ),
}

def fill_cell(row, column):
    spec = dict(COLUMNS[column])
    query = spec.pop("template").format(row=row)
    resp = requests.get(API, headers=HEADERS, timeout=30, params={
        "query": query, "include_full_text_chunks": True, **spec,
    })
    resp.raise_for_status()
    papers = resp.json()["results"]
    # Your LLM call: summarize from each paper's full_text_chunks, cite by DOI.
    return summarize_with_citations(row, column, papers)