Source code for tooluniverse.pdbe_ligands_tool
# pdbe_ligands_tool.py
"""
PDBe Ligands and Residues tool for ToolUniverse.
The PDBe REST API provides information about ligands bound to PDB structures
and detailed per-residue information for protein chains. These complement the
existing PDBe compound tools (which look up compounds by ID) by providing
structure-centric queries.
API: https://www.ebi.ac.uk/pdbe/api/pdb/entry/
No authentication required.
"""
import requests
from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool
PDBE_API_BASE_URL = "https://www.ebi.ac.uk/pdbe/api/pdb/entry"
[docs]
@register_tool("PDBeLigandsTool")
class PDBeLigandsTool(BaseTool):
"""
Tool for querying PDBe structure-bound ligands and residue details.
Supports:
- Get all ligands bound in a PDB structure (drug-like, cofactors, ions)
- Get per-residue listing with observed ratio for a PDB chain
No authentication required.
"""
[docs]
def __init__(self, tool_config: Dict[str, Any]):
super().__init__(tool_config)
self.timeout = tool_config.get("timeout", 30)
fields = tool_config.get("fields", {})
self.endpoint = fields.get("endpoint", "ligand_monomers")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the PDBe API call."""
try:
return self._query(arguments)
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"PDBe API timed out after {self.timeout}s",
}
except requests.exceptions.ConnectionError:
return {"status": "error", "error": "Failed to connect to PDBe API"}
except requests.exceptions.HTTPError as e:
status = e.response.status_code if e.response is not None else "unknown"
if status == 404:
pdb_id = arguments.get("pdb_id", "unknown")
return {
"status": "error",
"error": f"PDB entry '{pdb_id}' not found. Provide a valid 4-character PDB ID (e.g., '4hhb', '3ert').",
}
return {"status": "error", "error": f"PDBe API HTTP {status}"}
except Exception as e:
return {"status": "error", "error": f"Unexpected error: {str(e)}"}
[docs]
def _entry_exists(self, pdb_id: str) -> bool:
"""Check whether a PDB entry exists via the (near-universally populated) summary endpoint."""
try:
resp = requests.get(
f"{PDBE_API_BASE_URL}/summary/{pdb_id}", timeout=self.timeout
)
return resp.status_code == 200
except requests.exceptions.RequestException:
return False
[docs]
def _query(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Route to appropriate endpoint."""
if self.endpoint == "ligand_monomers":
return self._get_ligand_monomers(arguments)
elif self.endpoint == "residue_listing":
return self._get_residue_listing(arguments)
else:
return {"status": "error", "error": f"Unknown endpoint: {self.endpoint}"}
[docs]
def _get_ligand_monomers(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get all ligand monomers bound in a PDB structure."""
pdb_id = arguments.get("pdb_id", "")
if not pdb_id:
return {
"status": "error",
"error": "pdb_id is required (4-character PDB ID, e.g., '4hhb', '3ert', '1m17').",
}
pdb_id = pdb_id.lower().strip()
url = f"{PDBE_API_BASE_URL}/ligand_monomers/{pdb_id}"
response = requests.get(url, timeout=self.timeout)
if response.status_code == 404:
# Fix-R2A-003: this endpoint 404s whenever an entry has no
# non-polymer ligand data — including well-known entries whose
# only "ligand" (e.g. a peptidomimetic inhibitor) is modeled as a
# polymer chain, not just when the entry itself doesn't exist.
# Disambiguate via the summary endpoint instead of reporting a
# real, well-known entry as "not found".
if self._entry_exists(pdb_id):
return {
"status": "success",
"data": {"pdb_id": pdb_id, "ligands": [], "total_ligands": 0},
"metadata": {"source": "PDBe REST API (ebi.ac.uk/pdbe)"},
"note": (
f"PDB entry '{pdb_id}' exists but has no non-polymer "
"ligand data. This commonly means the entry's "
"inhibitor/ligand is modeled as a polymer or peptide "
"chain rather than a discrete non-polymer ligand; "
"check the entry's polymer entities instead."
),
}
return {
"status": "error",
"error": f"PDB entry '{pdb_id}' not found. Provide a valid 4-character PDB ID (e.g., '4hhb', '3ert').",
}
response.raise_for_status()
data = response.json()
if pdb_id not in data:
return {"status": "error", "error": f"No ligand data for PDB '{pdb_id}'."}
ligands_raw = data[pdb_id]
ligands = []
ligand_offset = max(int(arguments.get("ligand_offset") or 0), 0)
ligand_limit = min(max(int(arguments.get("ligand_limit") or 50), 1), 500)
annotation_offset = max(int(arguments.get("annotation_offset") or 0), 0)
annotation_limit = min(max(int(arguments.get("annotation_limit") or 5), 1), 500)
ligand_page = ligands_raw[ligand_offset : ligand_offset + ligand_limit]
for lig in ligand_page:
annotations = lig.get("annotations", [])
annotation_info = []
annotation_page = annotations[
annotation_offset : annotation_offset + annotation_limit
]
for ann in annotation_page:
interacting = ann.get("interacting_entity", {})
annotation_info.append(
{
"type": ann.get("type"),
"interacting_entity_id": interacting.get("entity_id"),
"interacting_chain": interacting.get("auth_asym_id"),
"interacting_uniprot": interacting.get("best_unp_accession"),
}
)
ligands.append(
{
"chem_comp_id": lig.get("chem_comp_id"),
"chem_comp_name": lig.get("chem_comp_name"),
"weight": lig.get("weight"),
"chain_id": lig.get("chain_id"),
"entity_id": lig.get("entity_id"),
"author_residue_number": lig.get("author_residue_number"),
"carbohydrate_polymer": lig.get("carbohydrate_polymer", False),
"annotations": annotation_info,
"annotations_total": len(annotations),
"annotations_returned": len(annotation_page),
"annotations_truncated": (
annotation_offset + len(annotation_page) < len(annotations)
),
"next_annotation_offset": (
annotation_offset + len(annotation_page)
if annotation_offset + len(annotation_page) < len(annotations)
else None
),
}
)
ligands_truncated = ligand_offset + len(ligand_page) < len(ligands_raw)
return {
"status": "success",
"data": {
"pdb_id": pdb_id,
"ligands": ligands,
"total_ligands": len(ligands_raw),
"ligand_offset": ligand_offset,
"ligands_returned": len(ligand_page),
"ligands_truncated": ligands_truncated,
"next_ligand_offset": (
ligand_offset + len(ligand_page) if ligands_truncated else None
),
},
"metadata": {
"source": "PDBe REST API (ebi.ac.uk/pdbe)",
},
}
[docs]
def _get_residue_listing(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get per-residue listing for a PDB structure chain."""
pdb_id = arguments.get("pdb_id", "")
if not pdb_id:
return {
"status": "error",
"error": "pdb_id is required (4-character PDB ID, e.g., '4hhb', '3ert').",
}
chain_id = arguments.get("chain_id", None)
pdb_id = pdb_id.lower().strip()
if chain_id:
url = f"{PDBE_API_BASE_URL}/residue_listing/{pdb_id}/chain/{chain_id}"
else:
url = f"{PDBE_API_BASE_URL}/residue_listing/{pdb_id}"
response = requests.get(url, timeout=self.timeout)
if response.status_code == 404:
# Fix-R2A-003: see _get_ligand_monomers — a 404 here can mean "no
# data for this specific chain/entry at this endpoint" rather than
# "entry does not exist". Disambiguate via the summary endpoint.
if self._entry_exists(pdb_id):
return {
"status": "success",
"data": {"pdb_id": pdb_id, "molecules": []},
"metadata": {"source": "PDBe REST API (ebi.ac.uk/pdbe)"},
"note": (
f"PDB entry '{pdb_id}' exists but has no residue data "
f"at this endpoint{f' for chain {chain_id}' if chain_id else ''}."
),
}
return {
"status": "error",
"error": f"PDB entry '{pdb_id}' not found. Provide a valid 4-character PDB ID (e.g., '4hhb', '3ert').",
}
response.raise_for_status()
data = response.json()
if pdb_id not in data:
return {"status": "error", "error": f"No residue data for PDB '{pdb_id}'."}
entry_data = data[pdb_id]
molecules = entry_data.get("molecules", [])
result_molecules = []
molecule_offset = max(int(arguments.get("molecule_offset") or 0), 0)
molecule_limit = min(max(int(arguments.get("molecule_limit") or 10), 1), 500)
chain_offset = max(int(arguments.get("chain_offset") or 0), 0)
chain_limit = min(max(int(arguments.get("chain_limit") or 5), 1), 500)
residue_offset = max(int(arguments.get("residue_offset") or 0), 0)
residue_limit = min(max(int(arguments.get("residue_limit") or 30), 1), 5000)
molecule_page = molecules[molecule_offset : molecule_offset + molecule_limit]
for mol in molecule_page:
chains = mol.get("chains", [])
chain_results = []
chain_page = chains[chain_offset : chain_offset + chain_limit]
for chain in chain_page:
residues = chain.get("residues", [])
residue_summary = []
residue_page = residues[residue_offset : residue_offset + residue_limit]
for res in residue_page:
residue_summary.append(
{
"residue_number": res.get("residue_number"),
"residue_name": res.get("residue_name"),
"author_residue_number": res.get("author_residue_number"),
"observed_ratio": res.get("observed_ratio"),
}
)
chain_results.append(
{
"chain_id": chain.get("chain_id"),
"struct_asym_id": chain.get("struct_asym_id"),
"total_residues": len(residues),
"residue_offset": residue_offset,
"residues_returned": len(residue_page),
"residues_truncated": (
residue_offset + len(residue_page) < len(residues)
),
"next_residue_offset": (
residue_offset + len(residue_page)
if residue_offset + len(residue_page) < len(residues)
else None
),
"residues": residue_summary,
}
)
result_molecules.append(
{
"entity_id": mol.get("entity_id"),
"chains": chain_results,
"total_chains": len(chains),
"chain_offset": chain_offset,
"chains_returned": len(chain_page),
"chains_truncated": (chain_offset + len(chain_page) < len(chains)),
"next_chain_offset": (
chain_offset + len(chain_page)
if chain_offset + len(chain_page) < len(chains)
else None
),
}
)
molecules_truncated = molecule_offset + len(molecule_page) < len(molecules)
return {
"status": "success",
"data": {
"pdb_id": pdb_id,
"molecules": result_molecules,
"total_molecules": len(molecules),
"molecule_offset": molecule_offset,
"molecules_returned": len(molecule_page),
"molecules_truncated": molecules_truncated,
"next_molecule_offset": (
molecule_offset + len(molecule_page)
if molecules_truncated
else None
),
},
"metadata": {
"source": "PDBe REST API (ebi.ac.uk/pdbe)",
},
}