Create ncbi.py
Browse files- mcp/ncbi.py +35 -0
mcp/ncbi.py
ADDED
|
@@ -0,0 +1,35 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
# mcp/ncbi.py
|
| 2 |
+
"""
|
| 3 |
+
NCBI E-utilities helpers – Gene, Protein, ClinVar, MeSH.
|
| 4 |
+
"""
|
| 5 |
+
|
| 6 |
+
import os
|
| 7 |
+
import httpx
|
| 8 |
+
from typing import List, Dict
|
| 9 |
+
|
| 10 |
+
NCBI_KEY = os.getenv("BIO_KEY") # optional but increases rate limits
|
| 11 |
+
BASE = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils/"
|
| 12 |
+
|
| 13 |
+
async def _get(endpoint: str, params: Dict) -> Dict:
|
| 14 |
+
if NCBI_KEY:
|
| 15 |
+
params["api_key"] = NCBI_KEY
|
| 16 |
+
async with httpx.AsyncClient(timeout=20) as client:
|
| 17 |
+
r = await client.get(f"{BASE}{endpoint}", params=params)
|
| 18 |
+
r.raise_for_status()
|
| 19 |
+
return r.json() if r.headers["Content-Type"].startswith("application/json") else r.text
|
| 20 |
+
|
| 21 |
+
# ---------- Public helpers ----------
|
| 22 |
+
async def search_gene(term: str, retmax: int = 5) -> List[Dict]:
|
| 23 |
+
"""Return basic gene info (ID + name/symbol) by search term."""
|
| 24 |
+
data = await _get("esearch.fcgi", {"db": "gene", "term": term, "retmode": "json", "retmax": retmax})
|
| 25 |
+
ids = data["esearchresult"]["idlist"]
|
| 26 |
+
if not ids:
|
| 27 |
+
return []
|
| 28 |
+
summary = await _get("esummary.fcgi", {"db": "gene", "id": ",".join(ids), "retmode": "json"})
|
| 29 |
+
return list(summary["result"].values())[1:] # first key is 'uids'
|
| 30 |
+
|
| 31 |
+
async def get_mesh_definition(term: str) -> str:
|
| 32 |
+
"""Return MeSH term definition (first record)."""
|
| 33 |
+
text = await _get("esummary.fcgi", {"db": "mesh", "term": term, "retmode": "json", "retmax": 1})
|
| 34 |
+
recs = list(text["result"].values())[1:]
|
| 35 |
+
return recs[0].get("ds_meshterms", [""])[0] if recs else ""
|