Source code for tooluniverse.lipidmaps_tool
# lipidmaps_tool.py
"""
LIPID MAPS Structure Database REST API tool for ToolUniverse.
LIPID MAPS (Lipid Metabolites And Pathways Strategy) is a comprehensive
classification system for lipids that includes lipid structures, genes,
proteins, and mass spectrometry data.
API Documentation: https://lipidmaps.org/resources/rest
"""
from typing import Dict, Any
from urllib.parse import quote
import requests
from .base_tool import BaseTool
from .tool_registry import register_tool
# Base URL for LIPID MAPS REST API
LIPIDMAPS_BASE_URL = "https://www.lipidmaps.org/rest"
# LIPID MAPS sits behind Cloudflare and 403s the default python-requests UA
# with a "Just a moment..." challenge page. A normal browser UA passes through
# the JS-less challenge for plain REST endpoints — same trick as
# file_download_tool.py uses for Wikipedia.
_REQUEST_HEADERS = {
"User-Agent": (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36 ToolUniverse/LipidMaps"
),
"Accept": "application/json,text/plain,*/*",
}
_COMPOUND_OUTPUT_ITEMS = {
"all",
"classification",
"regno",
"lm_id",
"name",
"sys_name",
"synonyms",
"abbrev",
"abbrev_chains",
"core",
"main_class",
"sub_class",
"class_level4",
"exactmass",
"formula",
"inchi",
"inchi_key",
"kegg_id",
"hmdb_id",
"chebi_id",
"lipidbank_id",
"pubchem_cid",
"smiles",
"molfile",
"structure",
"physchem",
"png",
}
[docs]
@register_tool("LipidMapsTool")
class LipidMapsTool(BaseTool):
"""
Tool for querying LIPID MAPS Structure Database REST API.
LIPID MAPS provides comprehensive lipidomics data including:
- Lipid structure information (LMSD - LIPID MAPS Structure Database)
- Lipid-related gene information (LMPD - LIPID MAPS Proteome Database)
- Lipid-related protein information
- Mass spectrometry m/z search
No authentication required. Free for academic/research use.
URL Pattern: /rest/{context}/{input_item}/{input_value}/{output_item}/{output_format}
"""
[docs]
def __init__(self, tool_config: Dict[str, Any]):
super().__init__(tool_config)
self.timeout = tool_config.get("timeout", 30)
# Get the context type from config (compound, gene, protein, moverz)
self.context = tool_config.get("fields", {}).get("context", "compound")
self.input_item = tool_config.get("fields", {}).get("input_item", "lm_id")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the LIPID MAPS API call based on the configured context."""
context = self.context
try:
if context == "compound":
return self._query_compound(arguments)
elif context == "gene":
return self._query_gene(arguments)
elif context == "protein":
return self._query_protein(arguments)
elif context == "moverz":
return self._search_moverz(arguments)
else:
return {"status": "error", "error": f"Unknown context: {context}"}
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"LIPID MAPS API request timed out after {self.timeout} seconds",
}
except requests.exceptions.ConnectionError:
return {
"status": "error",
"error": "Failed to connect to LIPID MAPS API. Check network connectivity.",
}
except requests.exceptions.HTTPError as e:
return {
"status": "error",
"error": f"LIPID MAPS API HTTP error: {e.response.status_code}",
}
except Exception as e:
return {
"status": "error",
"error": f"Unexpected error querying LIPID MAPS: {str(e)}",
}
[docs]
def _make_request(self, sub_path: str) -> Dict[str, Any]:
"""Central method to handle API requests and response parsing."""
url = f"{LIPIDMAPS_BASE_URL}/{sub_path}"
response = requests.get(url, timeout=self.timeout, headers=_REQUEST_HEADERS)
response.raise_for_status()
raw_text = response.text.strip()
# Handle empty or null responses
if (
not raw_text
or raw_text.lower() == "null"
or raw_text == '""'
or raw_text in ("{}", "[]")
):
return {
"status": "success",
"data": [],
"message": "No results found",
"metadata": {
"total_results": 0,
"result_set_complete": True,
},
}
try:
data = response.json()
# LIPID MAPS returns results with Row1, Row2, ... keys for multiple results
# or a flat dict for single results
if isinstance(data, dict):
# Check if it's a multi-row result (Row1, Row2, ...)
row_keys = [k for k in data.keys() if k.startswith("Row")]
if row_keys:
if len(row_keys) != len(data) or any(
not key[3:].isdigit() or not isinstance(data[key], dict)
for key in row_keys
):
return {
"status": "error",
"error": "LIPID MAPS returned malformed keyed row records",
}
# Convert Row1, Row2, ... to a list
row_keys.sort(key=lambda key: int(key[3:]))
results = [data[key] for key in row_keys]
return {
"status": "success",
"data": results,
"metadata": {
"total_results": len(results),
"result_set_complete": True,
},
}
# Single result or selected-field response.
return {
"status": "success",
"data": data,
"metadata": {"total_results": 1, "result_set_complete": True},
}
elif isinstance(data, list):
if any(not isinstance(record, dict) for record in data):
return {
"status": "error",
"error": "LIPID MAPS returned a malformed record list",
}
return {
"status": "success",
"data": data,
"metadata": {
"total_results": len(data),
"result_set_complete": True,
},
}
else:
return {
"status": "error",
"error": "LIPID MAPS returned an unexpected JSON response shape",
}
except ValueError:
return {
"status": "error",
"error": "LIPID MAPS JSON endpoint returned a non-JSON response "
f"(first 200 chars: {raw_text[:200]})",
}
# Cross-reference / structure input items accepted by the LMSD compound
# route. Lets a single tool resolve any external metabolite/lipid id to a
# LIPID MAPS record by passing xref_type (e.g. kegg_id, hmdb_id).
_XREF_INPUT_ITEMS = {
"kegg_id",
"hmdb_id",
"chebi_id",
"pubchem_cid",
"inchi_key",
"lipidbank_id",
"lm_id",
"abbrev",
"abbrev_chains",
"formula",
"smiles",
"inchi",
}
[docs]
def _query_compound(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Query compound/lipid information from LMSD."""
input_value = arguments.get("input_value")
output_item = arguments.get("output_item", "all")
if not isinstance(input_value, str) or not input_value.strip():
return {
"status": "error",
"error": "input_value must be a non-empty string",
}
input_value = input_value.strip()
if (
not isinstance(output_item, str)
or output_item not in _COMPOUND_OUTPUT_ITEMS
):
return {
"status": "error",
"error": "output_item is not a supported LIPID MAPS compound field",
}
# Reject unrecognized parameters. `xref_type` has a default (kegg_id), so
# a typo like `input_type` (instead of `xref_type`) was silently accepted
# and the lookup fell back to kegg_id -- a PubChem CID searched as a KEGG
# id then returned an empty "not found" the user wrongly trusted. Name
# the bad parameter instead of silently returning a false empty.
_recognized = {"input_value", "output_item", "xref_type", "input_item"}
unknown = sorted(k for k in arguments if k not in _recognized)
if unknown:
return {
"status": "error",
"error": (
f"Unrecognized parameter(s): {', '.join(unknown)}. Supported: "
"input_value, xref_type (e.g. 'pubchem_cid', 'hmdb_id', "
"'chebi_id', 'kegg_id', 'inchi_key', 'abbrev_chains'), "
"output_item. Did you mean 'xref_type'?"
),
}
# Allow the input item (the LMSD lookup key) to be overridden per call
# via xref_type / input_item, falling back to the config default. This
# powers the cross-reference lookup tool without a new tool class.
xref_type = arguments.get("xref_type")
explicit_input_item = arguments.get("input_item")
if (
xref_type is not None
and explicit_input_item is not None
and xref_type != explicit_input_item
):
return {
"status": "error",
"error": "Conflicting xref_type and input_item values were supplied",
}
input_item = xref_type or explicit_input_item or self.input_item
if not isinstance(input_item, str):
return {"status": "error", "error": "xref_type must be a string"}
input_item = input_item.strip()
if input_item not in self._XREF_INPUT_ITEMS:
return {
"status": "error",
"error": f"Unsupported xref_type '{input_item}'. Supported: "
+ ", ".join(sorted(self._XREF_INPUT_ITEMS)),
}
# LMSD's `abbrev` field only stores sum-composition shorthand (e.g.
# 'PC 34:1'), never the acyl-chain-specific form real lipidomics
# workflows use ('PC 16:0/18:1') -- that regiospecific form only
# matches the separate `abbrev_chains` field, and even there only as
# 'PC 16:0_18:1' (underscore, not slash -- confirmed live that a
# literal or percent-encoded '/' in this path segment 404s against
# LIPID MAPS' router, so the substitution below is required, not
# cosmetic). Retry automatically so a name copy-pasted straight out
# of a lipidomics results table (slash form) still resolves instead
# of silently returning zero hits.
resolved_via = None
effective_item = input_item
effective_value = input_value
if input_item in ("abbrev", "abbrev_chains") and "/" in input_value:
effective_item = "abbrev_chains"
effective_value = input_value.replace("/", "_")
resolved_via = f"abbrev_chains={effective_value}"
sub_path = "/".join(
quote(segment, safe="")
for segment in (
"compound",
effective_item,
effective_value,
output_item,
"json",
)
)
result = self._make_request(sub_path)
metadata = result.setdefault("metadata", {})
metadata.update(
{
"input_item": effective_item,
"input_value": effective_value,
"output_item": output_item,
"field_scope_complete": output_item == "all",
}
)
if resolved_via:
metadata["resolved_via"] = resolved_via
if (
result.get("status") == "success"
and output_item == "all"
and effective_item == "lm_id"
and result.get("data")
):
records = (
result["data"] if isinstance(result["data"], list) else [result["data"]]
)
mismatches = [
record.get("lm_id")
for record in records
if not isinstance(record, dict)
or (record.get("lm_id") or "").upper() != effective_value.upper()
]
if mismatches:
return {
"status": "error",
"error": "LIPID MAPS returned a different LMID than requested",
"returned_lm_ids": mismatches,
}
metadata["exact_identifier_verified"] = True
return result
[docs]
def _query_gene(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Query lipid-related gene information from LMPD."""
input_value = arguments.get("input_value", "")
output_item = arguments.get("output_item", "all")
if not input_value:
return {"status": "error", "error": "input_value parameter is required"}
return self._make_request(
f"gene/{self.input_item}/{input_value}/{output_item}/json"
)
[docs]
def _query_protein(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Query lipid-related protein information from LMPD."""
input_value = arguments.get("input_value", "")
output_item = arguments.get("output_item", "all")
if not input_value:
return {"status": "error", "error": "input_value parameter is required"}
return self._make_request(
f"protein/{self.input_item}/{input_value}/{output_item}/json"
)
[docs]
def _search_moverz(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search lipids by m/z value for mass spectrometry identification."""
mz_value = arguments.get("mz_value")
ion_type = arguments.get("ion_type", "M-H")
tolerance = arguments.get("tolerance", 0.01)
if mz_value is None:
return {"status": "error", "error": "mz_value parameter is required"}
# LIPID MAPS m/z endpoint returns TSV, not JSON
url = f"{LIPIDMAPS_BASE_URL}/moverz/LIPIDS/{mz_value}/{ion_type}/-/{tolerance}/txt"
response = requests.get(url, timeout=self.timeout, headers=_REQUEST_HEADERS)
response.raise_for_status()
raw_text = response.text.strip()
if not raw_text:
return {
"status": "success",
"data": [],
"message": "No lipid matches found",
}
# Parse TSV response
lines = raw_text.split("\n")
results = []
headers = None
for line in lines:
if not line.strip():
continue
parts = line.split("\t")
if headers is None:
headers = parts
else:
if len(parts) == len(headers):
results.append(dict(zip(headers, parts)))
else:
results.append({"raw": line})
return {
"status": "success",
"data": results,
"metadata": {
"total_results": len(results),
"query_mz": mz_value,
"ion_type": ion_type,
"tolerance": tolerance,
},
}