mcp/openfda.py

# mcp/openfda.py

#!/usr/bin/env python3
"""MedGenesis – **openFDA** drug‑event wrapper (async, cached).

Enhancements over the legacy helper
-----------------------------------
* Back‑off retry (2×, 4×, 8×) for intermittent 503s.
* `@lru_cache` (12 h, 512 keys) to spare quota.
* Returns empty list instead of raising on any service hiccup.
* Accepts optional `limit` kwarg (capped at 20).
* Normalises API response to `schemas.DrugSafety`‑compatible dicts.

API docs: https://open.fda.gov/apis/drug/event/
"""
from __future__ import annotations

import asyncio, httpx
from functools import lru_cache
from typing import List, Dict

_URL = "https://api.fda.gov/drug/event.json"
_TIMEOUT = 15
_MAX_LIMIT = 20

# ---------------------------------------------------------------------
# Internal helper with retry
# ---------------------------------------------------------------------
async def _get(params: Dict, *, retries: int = 3) -> Dict:
    delay = 2
    last = None
    for _ in range(retries):
        async with httpx.AsyncClient(timeout=_TIMEOUT) as cli:
            last = await cli.get(_URL, params=params)
            if last.status_code == 200:
                return last.json()
        await asyncio.sleep(delay)
        delay *= 2
    return last.json() if last else {}


# ---------------------------------------------------------------------
# Public API – cached
# ---------------------------------------------------------------------
@lru_cache(maxsize=512)
async def fetch_drug_safety(drug: str, *, limit: int = 3) -> List[Dict]:
    """Return recent adverse‑event reports for *drug* (empty list on failure)."""
    limit = max(1, min(limit, _MAX_LIMIT))
    params = {
        "search": f'patient.drug.medicinalproduct:"{drug}"',
        "limit" : limit,
    }

    try:
        data = await _get(params)
        rows = data.get("results", [])
        out: List[Dict] = []
        for ev in rows:
            out.append({
                "safety_report_id": ev.get("safetyreportid"),
                "serious"         : ev.get("serious"),
                "reactions"       : [r.get("reactionmeddrapt") for r in ev.get("patient", {}).get("reaction", [])],
                "receivedate"     : ev.get("receivedate"),
            })
        return out
    except Exception:
        return []


# Quick test -----------------------------------------------------------------
if __name__ == "__main__":
    import asyncio, json
    async def _demo():
        res = await fetch_drug_safety("temozolomide", limit=2)
        print(json.dumps(res, indent=2))
    asyncio.run(_demo())