Source code for tooluniverse.mobidb_tool
# mobidb_tool.py
"""
MobiDB tool for ToolUniverse.
MobiDB is a comprehensive database of protein disorder and mobility,
integrating curated data from DisProt, predicted disorder from MobiDB-lite,
and structural data from PDB to provide a consensus view of intrinsically
disordered regions, linear interacting peptides (LIPs), and binding modes.
API: https://mobidb.org/api/download
No authentication required.
"""
import requests
from typing import Dict, Any, Optional
from .base_tool import BaseTool
from .tool_registry import register_tool
MOBIDB_BASE_URL = "https://mobidb.org/api/download"
[docs]
@register_tool("MobiDBTool")
class MobiDBTool(BaseTool):
"""
Tool for querying MobiDB protein disorder database.
Supports:
- Get protein disorder predictions, curated regions, PTMs, binding modes
- Get consensus disorder summary for a protein
No authentication required.
"""
[docs]
def __init__(self, tool_config: Dict[str, Any]):
super().__init__(tool_config)
self.timeout = tool_config.get("timeout", 60)
fields = tool_config.get("fields", {})
self.endpoint = fields.get("endpoint", "get_protein")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the MobiDB API call."""
try:
return self._query(arguments)
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"MobiDB API timed out after {self.timeout}s.",
}
except requests.exceptions.ConnectionError:
return {
"status": "error",
"error": "Failed to connect to MobiDB API (mobidb.org).",
}
except requests.exceptions.HTTPError as e:
status = e.response.status_code if e.response is not None else "unknown"
if status == 404:
return {
"status": "error",
"error": "Protein not found in MobiDB. Check the UniProt accession.",
}
return {"status": "error", "error": f"MobiDB API HTTP {status}"}
except Exception as e:
return {"status": "error", "error": f"Unexpected error: {str(e)}"}
[docs]
def _fetch_protein_json(self, accession: str) -> Optional[Dict[str, Any]]:
"""GET the MobiDB download endpoint for an accession and parse JSON.
Returns None if MobiDB has no data for this accession.
Fix-R16C-1: for an unrecognized identifier MobiDB returns HTTP 200
with a completely empty body (confirmed live, e.g. a UniProt entry
name like "SNCA_HUMAN" instead of an accession like "P37840") rather
than a 404 -- so raise_for_status() never fires, and the previous
code called response.json() directly, letting a raw
json.JSONDecodeError ("Expecting value: line 1 column 1") leak
through as an opaque "Unexpected error" instead of an actionable
not-found message.
"""
response = requests.get(
MOBIDB_BASE_URL,
params={"acc": accession, "format": "json"},
allow_redirects=True,
timeout=self.timeout,
)
response.raise_for_status()
if not response.text.strip():
return None
return response.json()
[docs]
@staticmethod
def _not_found_error(accession: str) -> Dict[str, Any]:
"""Standard error for an accession MobiDB has no data for."""
return {
"status": "error",
"error": (
f"No MobiDB data found for accession '{accession}'. "
"MobiDB expects a UniProt accession (e.g. 'P37840'), "
"not a UniProt entry name (e.g. 'SNCA_HUMAN')."
),
}
[docs]
def _query(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Route to appropriate endpoint."""
if self.endpoint == "get_protein":
return self._get_protein(arguments)
elif self.endpoint == "get_consensus":
return self._get_consensus(arguments)
else:
return {"status": "error", "error": f"Unknown endpoint: {self.endpoint}"}
[docs]
def _get_protein(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get comprehensive disorder data for a protein from MobiDB."""
accession = arguments.get("accession", "")
if not accession:
return {
"status": "error",
"error": "accession is required (UniProt ID, e.g., 'P04637' for TP53).",
}
data = self._fetch_protein_json(accession)
if data is None:
return self._not_found_error(accession)
# Extract key disorder data
prediction = data.get("prediction-disorder-mobidb_lite", {})
curated_disorder = data.get("curated-disorder-disprot", {})
curated_lip = data.get("curated-lip-disprot", {})
# Extract PTM info
ptms = data.get("ptms", {})
ptm_summary = {}
if isinstance(ptms, dict):
ptm_names = ptms.get("names", [])
ptm_positions = ptms.get("positions", [])
ptm_summary = {
"count": len(ptm_positions) if isinstance(ptm_positions, list) else 0,
"types": ptm_names[:10] if isinstance(ptm_names, list) else [],
}
return {
"status": "success",
"data": {
"accession": data.get("acc"),
"gene": data.get("gene"),
"name": data.get("name"),
"organism": data.get("organism"),
"length": data.get("length"),
"sequence": data.get("sequence"),
"prediction_disorder": {
"regions": prediction.get("regions", []),
"content_fraction": prediction.get("content_fraction"),
"content_count": prediction.get("content_count"),
},
"curated_disorder": {
"regions": curated_disorder.get("regions", []),
"content_count": curated_disorder.get("content_count"),
},
"curated_lip": {
"regions": curated_lip.get("regions", []),
"content_count": curated_lip.get("content_count"),
},
"ptm_summary": ptm_summary,
"localization": data.get("localization", []),
},
"metadata": {
"source": "MobiDB (mobidb.org)",
"reviewed": data.get("reviewed"),
},
}
[docs]
def _get_consensus(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get consensus disorder summary for a protein."""
accession = arguments.get("accession", "")
if not accession:
return {
"status": "error",
"error": "accession is required (UniProt ID, e.g., 'P04637' for TP53).",
}
data = self._fetch_protein_json(accession)
if data is None:
return self._not_found_error(accession)
# Extract consensus disorder data
consensus_disorder = data.get("prediction-disorder-mobidb_lite", {})
curated_merge = data.get("curated-disorder-merge", {})
phase_sep = data.get("curated-phase_separation-priority", {})
# Collect all binding mode info
binding_modes = {}
for key in data:
if key.startswith("curated-binding_mode"):
bm_data = data[key]
if isinstance(bm_data, dict) and bm_data.get("regions"):
binding_modes[key] = {
"regions": bm_data.get("regions", []),
"content_count": bm_data.get("content_count"),
}
return {
"status": "success",
"data": {
"accession": data.get("acc"),
"gene": data.get("gene"),
"name": data.get("name"),
"organism": data.get("organism"),
"length": data.get("length"),
"predicted_disorder": {
"regions": consensus_disorder.get("regions", []),
"content_fraction": consensus_disorder.get("content_fraction"),
},
"curated_disorder_merged": {
"regions": curated_merge.get("regions", []),
},
"phase_separation": {
"regions": phase_sep.get("regions", []),
},
"binding_modes": binding_modes,
},
"metadata": {
"source": "MobiDB (mobidb.org)",
},
}