Update mcp/hpo.py

mcp/hpo.py

@@ -1,16 +1,64 @@
-# mcp/hpo.py
-"""
-Human Phenotype Ontology quick lookup.
+#!/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
-from typing import Dict
+import httpx, asyncio
+from functools import lru_cache
+from typing import Dict, Optional, List
 
-BASE = "https://hpo.jax.org/api/hpo/term/"
+_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:
-    """Fetch HPO term details by ID (e.g., HP:0001250)."""
-    async with httpx.AsyncClient(timeout=15) as client:
-        r = await client.get(BASE + term_id)
-        r.raise_for_status()
-        return r.json()
+    """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())