Source code for tooluniverse.geo_tool

"""
GEO Database REST API Tool

This tool provides access to gene expression data from the GEO database.
GEO is a public repository that archives and freely distributes microarray,
next-generation sequencing, and other forms of high-throughput functional
genomics data.
"""

import requests
from typing import Dict, Any, List
from .base_tool import BaseTool
from .tool_registry import register_tool

GEO_BASE_URL = "https://eutils.ncbi.nlm.nih.gov/entrez/eutils"


[docs] @register_tool("GEORESTTool") class GEORESTTool(BaseTool): """ GEO Database REST API tool. Generic wrapper for GEO API endpoints defined in expression_tools.json. """
[docs] def __init__(self, tool_config): super().__init__(tool_config) fields = tool_config.get("fields", {}) parameter = tool_config.get("parameter", {}) self.endpoint_template: str = fields.get("endpoint", "/esearch.fcgi") self.required: List[str] = parameter.get("required", []) self.output_format: str = fields.get("return_format", "JSON")
def _build_url(self, arguments: Dict[str, Any]) -> str | Dict[str, Any]: """Build URL for GEO API request.""" url_path = self.endpoint_template return GEO_BASE_URL + url_path def _build_params(self, arguments: Dict[str, Any]) -> Dict[str, Any]: """Build parameters for GEO API request.""" params = {"db": "gds", "retmode": "json", "retmax": 50} # Build search query query_parts = [] if "query" in arguments: query_parts.append(arguments["query"]) if "organism" in arguments: organism = arguments["organism"] if organism.lower() == "homo sapiens": query_parts.append("Homo sapiens[organism]") elif organism.lower() == "mus musculus": query_parts.append("Mus musculus[organism]") else: query_parts.append(f'"{organism}"[organism]') if "study_type" in arguments: study_type = arguments["study_type"] query_parts.append(f'"{study_type}"[study_type]') if "platform" in arguments: platform = arguments["platform"] query_parts.append(f'"{platform}"[platform]') if "date_range" in arguments: date_range = arguments["date_range"] if ":" in date_range: start_year, end_year = date_range.split(":") query_parts.append(f'"{start_year}"[PDAT] : "{end_year}"[PDAT]') if query_parts: params["term"] = " AND ".join(query_parts) if "limit" in arguments: params["retmax"] = min(arguments["limit"], 500) if "sort" in arguments: sort = arguments["sort"] if sort == "date": params["sort"] = "relevance" elif sort == "title": params["sort"] = "title" else: params["sort"] = "relevance" return params def _make_request(self, url: str, params: Dict[str, Any]) -> Dict[str, Any]: """Perform a GET request and handle common errors.""" try: response = requests.get(url, params=params, timeout=30) response.raise_for_status() if self.output_format == "JSON": return response.json() else: return {"data": response.text} except requests.exceptions.RequestException as e: return {"error": f"Request failed: {str(e)}"} except Exception as e: return {"error": f"Unexpected error: {str(e)}"}
[docs] def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]: """Execute the tool with given arguments.""" # Validate required parameters for param in self.required: if param not in arguments: return {"error": f"Missing required parameter: {param}"} url = self._build_url(arguments) if isinstance(url, dict) and "error" in url: return url params = self._build_params(arguments) return self._make_request(url, params)