mcp/disgenet.py

"""
disgenet.py  ·  Disease-Gene associations helper
Docs: https://www.disgenet.com/downloads  (REST v1)   🛈

Change-log
──────────
• 2025-06-25  – .org → .COM redirect (301) broke calls.
               We now default to https://www.disgenet.com/api
               and still follow redirects if they add a CDN later.
• Graceful retry + 24 h LRU-cache.
• Empty list on any error so orchestrator never crashes.
"""

from __future__ import annotations
import os, asyncio, httpx
from functools import lru_cache
from typing import List, Dict

_TOKEN  = os.getenv("DISGENET_KEY")            # optional Bearer token
_BASE   = "https://www.disgenet.com/api"       # ← new canonical host
_HDRS   = {"Accept": "application/json"}
if _TOKEN:
    _HDRS["Authorization"] = f"Bearer {_TOKEN}"

_TIMEOUT = 12
_RETRIES = 2

# ────────────────────────────────────────────────────────────────────
@lru_cache(maxsize=512)
async def disease_to_genes(disease_name: str,
                           limit: int = 10) -> List[Dict]:
    """
    Return top-N gene associations for *disease_name*.
    Empty list on failure or if none found.
    """
    url = f"{_BASE}/gda/disease/{disease_name.lower()}"
    params = {"source": "ALL", "format": "json"}

    async def _one_call() -> List[Dict]:
        async with httpx.AsyncClient(timeout=_TIMEOUT,
                                     headers=_HDRS,
                                     follow_redirects=True) as cli:
            r = await cli.get(url, params=params)
            if r.status_code == 404:
                return []
            r.raise_for_status()
            return r.json()[:limit]

    delay = 0.0
    for _ in range(_RETRIES):
        try:
            return await _one_call()
        except (httpx.HTTPStatusError, httpx.ReadTimeout):
            await asyncio.sleep(delay or 0.7)
            delay = 0.0            # retry only once
        except Exception:
            break
    return []                      # graceful fallback