import os
import requests
from typing import Any, Dict, Optional
class MGnifyClient:
def __init__(self, base_url: Optional[str] = None, api_key: Optional[str] = None, timeout: float = 30.0):
self.base_url = (base_url or os.getenv("MG_BASE_URL")
or "https://www.ebi.ac.uk/metagenomics/api/v1").rstrip("/")
self.api_key = api_key or os.getenv("MG_API_KEY")
self.timeout = timeout
self.session = requests.Session()
if self.api_key:
self.session.headers.update({"Authorization": f"Bearer {self.api_key}"})
def _get(self, path: str, params: Optional[Dict[str, Any]] = None):
url = f"{self.base_url}{path}"
r = self.session.get(url, params=params, timeout=self.timeout)
if r.status_code == 429:
# propagate rate limit details for the MCP layer to surface
raise RuntimeError(f"Rate limited: retry-after={r.headers.get('Retry-After')}")
r.raise_for_status()
return r.json()
# ---- Convenience wrappers ------------------------------------------------
def biomes(self):
return self._get("/biomes")
def search_studies(self, query: Optional[str], biome: Optional[str], page: int, size: int):
params = {"page": page, "page_size": size}
if query:
params["search"] = query
if biome:
params["biome"] = biome
return self._get("/studies", params)
def run(self, run_id: str):
return self._get(f"/runs/{run_id}")
def analyses_for_study(self, study_id: str, typ: Optional[str], page: int, size: int):
params = {"page": page, "page_size": size}
if typ:
params["type"] = typ
return self._get(f"/studies/{study_id}/analyses", params)
