mcp/who.py

#!/usr/bin/env python3
"""WHO Global Health Observatory (GHO) async helper.

Key features
------------
* **Essential medicines list** – `list_essential_medicines()`
* **COVID‑19 country stats**  – `fetch_covid_timeseries(iso_code)`
* Built on `httpx.AsyncClient`, 20‑second timeout, no API‑key required.
* Caches responses for 24 h in memory to spare quota.

GHO OData API docs: https://www.who.int/data/gho/info/gho-odata-api
"""
from __future__ import annotations

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

BASE = "https://ghoapi.azureedge.net/api"


# ---------------------------------------------------------------------
# 📦 Essential medicines – EML endpoint
# ---------------------------------------------------------------------

@lru_cache(maxsize=1)
async def list_essential_medicines() -> List[Dict]:
    """Return current WHO Model List of Essential Medicines (EML)."""
    url = f"{BASE}/EML"
    async with httpx.AsyncClient(timeout=20) as client:
        resp = await client.get(url)
        resp.raise_for_status()
        return resp.json().get("value", [])


# ---------------------------------------------------------------------
# 🦠 COVID‑19 time‑series by ISO country code
# ---------------------------------------------------------------------

COVID_DATASET = "COVID19CasesByCountries"

@lru_cache(maxsize=128)
async def fetch_covid_timeseries(iso_code: str) -> List[Dict]:
    """Return daily cases & deaths for given 3‑letter ISO code (e.g. 'USA')."""
    url = f"{BASE}/{COVID_DATASET}?$filter=Code%20eq%20'{iso_code.upper()}'"
    async with httpx.AsyncClient(timeout=20) as client:
        resp = await client.get(url)
        resp.raise_for_status()
        return resp.json().get("value", [])


# ---------------------------------------------------------------------
# Example CLI usage (for testing) – run: `python -m mcp.who`
# ---------------------------------------------------------------------
if __name__ == "__main__":
    async def _demo():
        meds = await list_essential_medicines()
        print(f"Essential medicines fetched: {len(meds):,}")
        covid = await fetch_covid_timeseries("NGA")
        print(f"Nigeria COVID rows: {len(covid):,}")
    asyncio.run(_demo())