src/tools/fda.py

import httpx
import time

BASE = "https://api.fda.gov/drug/label.json"


async def query_fda_labels(drug_name: str = "", gene: str = "") -> list[dict]:
    """Search FDA drug labels for drug name or gene mentions."""
    search_term = drug_name if drug_name else gene
    params = {
        "search": f'indications_and_usage:"{search_term}"',
        "limit": 3,
    }

    async with httpx.AsyncClient(timeout=15) as client:
        r = await client.get(BASE, params=params)
        if r.status_code != 200:
            return []
        results_raw = r.json().get("results", [])

    return _parse_labels(results_raw)


async def query_fda_by_drugs(drug_names: list[str]) -> list[dict]:
    """Search FDA labels for specific drug names.

    Useful when the question mentions specific drug options.
    """
    all_results = []
    async with httpx.AsyncClient(timeout=15) as client:
        for name in drug_names[:5]:
            params = {
                "search": f'openfda.generic_name:"{name}" OR openfda.brand_name:"{name}"',
                "limit": 1,
            }
            r = await client.get(BASE, params=params)
            if r.status_code == 200:
                results_raw = r.json().get("results", [])
                all_results.extend(_parse_labels(results_raw))

    return all_results


def _parse_labels(results_raw: list[dict]) -> list[dict]:
    """Parse FDA label API results into evidence dicts."""
    results = []
    for item in results_raw:
        brand = item.get("openfda", {}).get("brand_name", [""])[0]
        generic = item.get("openfda", {}).get("generic_name", [""])[0]
        indication = item.get("indications_and_usage", [""])[0][:400]
        app_no = item.get("openfda", {}).get("application_number", [""])[0]
        url = (
            f"https://www.accessdata.fda.gov/scripts/cder/daf/index.cfm"
            f"?event=overview.process&ApplNo={app_no}"
            if app_no
            else "https://www.accessdata.fda.gov"
        )
        name_str = f"{brand} ({generic})" if generic else brand
        results.append(
            {
                "url": url,
                "snippet": f"{name_str}: {indication}",
                "date_retrieved": int(time.time()),
                "source_name": "FDA",
            }
        )
    return results