mcp/hpo.py

#!/usr/bin/env python3
"""MedGenesis – Human Phenotype Ontology (HPO) async helper.

Adds reliability & caching over the minimal original version.

Capabilities
~~~~~~~~~~~~
* `get_hpo(term_id)`   – fetch full term JSON (`HP:XXXXXXX`), cached 24 h.
* `search_hpo(term)`   – simple keyword search, returns top hit (id + name).

API docs: https://hpo.jax.org/api 
"""
from __future__ import annotations

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

_BASE = "https://hpo.jax.org/api/hpo"
_TIMEOUT = 15
_HEADERS = {"User-Agent": "MedGenesis/1.0 (https://huggingface.co/spaces)"}

# ---------------------------------------------------------------------
# Internal helper with retry
# ---------------------------------------------------------------------
async def _get(url: str, *, retries: int = 3) -> Dict:
    delay = 2
    last: Optional[httpx.Response] = None
    for _ in range(retries):
        async with httpx.AsyncClient(timeout=_TIMEOUT, headers=_HEADERS) as cli:
            last = await cli.get(url)
            if last.status_code == 200:
                return last.json()
        await asyncio.sleep(delay)
        delay *= 2
    last.raise_for_status()  # type: ignore

# ---------------------------------------------------------------------
# Public helpers (cached)
# ---------------------------------------------------------------------
@lru_cache(maxsize=512)
async def get_hpo(term_id: str) -> Dict:
    """Return JSON for `HP:NNNNNNN` (raises if not found)."""
    return await _get(f"{_BASE}/term/{term_id}")


@lru_cache(maxsize=256)
async def search_hpo(term: str) -> Optional[Dict]:
    """Return top search hit (`id`, `name`) for free‑text query or None."""
    data = await _get(f"{_BASE}/search/{term}")
    hits: List[Dict] = data.get("_embedded", {}).get("terms", [])
    return hits[0] if hits else None

# ---------------------------------------------------------------------
# CLI test
# ---------------------------------------------------------------------
if __name__ == "__main__":
    async def _demo():
        hit = await search_hpo("microcephaly")
        print("Top search hit:", hit)
        if hit:
            full = await get_hpo(hit["id"])
            print("Definition:", full.get("definition"))
    asyncio.run(_demo())