mcp/mygene.py

"""
MyGene.info async helper
Docs: https://docs.mygene.info/en/latest/api_service.html
• Key is OPTIONAL – set MYGENE_KEY only for >500 QPS quota.
• One retry on 5xx/timeout then returns {} so UI never crashes.
"""

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

_URL   = "https://mygene.info/v3/query"
_KEY   = os.getenv("MYGENE_KEY")          # optional
_FIELDS = "symbol,name,summary,alias,entrezgene,location"
_TIMEOUT = 8
_RETRIES = 2            # initial + one retry

@lru_cache(maxsize=512)
async def fetch_gene_info(symbol: str) -> Dict:
    params = {"q": symbol, "fields": _FIELDS, "size": 1}
    if _KEY:
        params["api_key"] = _KEY

    delay = 0.0
    for _ in range(_RETRIES):
        try:
            async with httpx.AsyncClient(timeout=_TIMEOUT) as cli:
                r = await cli.get(_URL, params=params)
                r.raise_for_status()
                return r.json().get("hits", [{}])[0]
        except (httpx.HTTPStatusError, httpx.ReadTimeout):
            await asyncio.sleep(delay or 0.6)   # back-off once
            delay = 0.0                         # only retry once
    return {}                                   # graceful fallback