Source code for tooluniverse.chebi_tool
# chebi_tool.py
"""
ChEBI 2.0 REST API tool for ToolUniverse.
ChEBI (Chemical Entities of Biological Interest) is a freely available
dictionary of molecular entities focused on 'small' chemical compounds,
maintained by EMBL-EBI. It provides an ontology-based classification
system, cross-references to other chemical databases, and detailed
structural information for 195,000+ compounds.
API: https://www.ebi.ac.uk/chebi/backend/api/
No authentication required. Free for all use.
"""
import re
import requests
from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool
CHEBI_BASE_URL = "https://www.ebi.ac.uk/chebi/backend/api/public"
# Aspirin carries 68 PubMed citations, 36 registry numbers and 24 roles. Each
# list is capped at this and the count and the cut are reported beside it.
ACCESSIONS_PER_TYPE = 50
ROLES_PER_COMPOUND = 50
# `database_accessions` is keyed by the kind of accession. CAS and
# REGISTRY_NUMBER are identifiers, CITATION is literature, MANUAL_X_REF is a
# link to another resource -- different things, and all four were dropped.
_ACCESSION_SECTIONS = {
"CAS": "cas_numbers",
"REGISTRY_NUMBER": "registry_numbers",
"CITATION": "citations",
"MANUAL_X_REF": "cross_references",
}
def _accession_rows(records):
"""One accession: its number, who assigned it and where it points."""
rows = []
for rec in records if isinstance(records, list) else []:
if not isinstance(rec, dict):
continue
rows.append(
{
"accession": rec.get("accession_number"),
"source": rec.get("source_name"),
"prefix": rec.get("prefix"),
"url": rec.get("url"),
}
)
return rows
def _accessions(raw):
"""The four accession sections, each capped and each counted."""
node = raw.get("database_accessions")
if not isinstance(node, dict):
node = {}
out = {}
for key, name in _ACCESSION_SECTIONS.items():
rows = _accession_rows(node.get(key))
out[name] = rows[:ACCESSIONS_PER_TYPE]
out[f"{name}_total"] = len(rows)
out[f"{name}_truncated"] = len(rows) > ACCESSIONS_PER_TYPE
other = {
key: len(_accession_rows(value))
for key, value in node.items()
if key not in _ACCESSION_SECTIONS
}
out["other_accession_counts"] = other
return out
def _roles(raw):
"""ChEBI's role classification, which is half of what ChEBI is for.
Each entry says whether the role is an application (what the compound is
used for), a chemical role or a biological role -- aspirin is classified
under 24 of them, including "non-steroidal anti-inflammatory drug".
"""
rows = []
for rec in raw.get("roles_classification") or []:
if not isinstance(rec, dict):
continue
rows.append(
{
"chebi_accession": rec.get("chebi_accession"),
"name": _strip_html(rec.get("name") or ""),
"definition": _strip_html(rec.get("definition") or ""),
"is_application": rec.get("application"),
"is_chemical_role": rec.get("chemical_role"),
"is_biological_role": rec.get("biological_role"),
}
)
return rows
# advanced_search is server-side paginated at 15 hits per page and exposes the
# page number as a 1-indexed query-string parameter (`?page=1` is the first
# page and is byte-identical to omitting the parameter; `?page=0` and any page
# beyond `number_pages` return a non-JSON error body). `limit` is clamped to
# 100, so at most ceil(100 / 15) == 7 pages are ever needed; the cap below is
# a hard stop that protects against a server that stops honouring `page`.
_SEARCH_PAGE_SIZE = 15
_SEARCH_MAX_PAGES = 8
_HTML_TAG_RE = re.compile(r"<[^>]+>")
def _strip_html(text):
"""Remove search-highlight markup (e.g. <em>...</em>) ChEBI embeds in
name/synonym fields. Non-string values pass through unchanged."""
if isinstance(text, str):
return _HTML_TAG_RE.sub("", text)
return text
def _normalize_chebi_id(chebi_id):
"""Accept the 'CHEBI:16113' CURIE that ChEBI_search / ChEBI_get_compound
emit (chebi_accession) and reduce it to the bare id the ontology endpoints
expect. Without this, chaining ChEBI_search -> ChEBI_get_ontology_parents
broke: the ontology tools required a bare integer and rejected the prefixed
accession the search hands back."""
if isinstance(chebi_id, str):
chebi_id = chebi_id.strip()
if chebi_id.upper().startswith("CHEBI:"):
chebi_id = chebi_id.split(":", 1)[1].strip()
return chebi_id
def _redirect_disclosure(requested_id, raw):
"""Disclose it when ChEBI answers about a *different* entity than the one
that was asked for.
ChEBI merges obsolete/duplicate accessions into a single primary entry and
then serves that primary entry for every accession that was merged into it.
`GET /compound/1/` returns id 18357, chebi_accession 'CHEBI:18357',
ascii_name '(R)-noradrenaline' -- CHEBI:1 is one of that entry's
`secondary_ids`. (EBI OLS4 agrees: CHEBI_1 is `is_obsolete: true` with
`term_replaced_by: CHEBI_18357`.) Serving the primary is correct upstream
behaviour; presenting it as though it *were* the requested compound is not.
A researcher pasting a legacy accession out of an old paper or dataset
would otherwise get a confident, success-labelled answer about the wrong
molecule.
Returns the keys to merge into the tool's `data` payload, or an empty dict
when the returned entity is the requested one -- so a direct hit stays
byte-identical to the pre-fix output.
"""
if not isinstance(raw, dict):
return {}
try:
requested_int = int(str(_normalize_chebi_id(requested_id)).strip())
returned_int = int(raw.get("id"))
except (TypeError, ValueError):
# Either side unparseable: we cannot prove a substitution happened, and
# inventing a warning would be worse than staying quiet.
return {}
if requested_int == returned_int:
return {}
requested_accession = f"CHEBI:{requested_int}"
returned_accession = raw.get("chebi_accession") or f"CHEBI:{returned_int}"
name = _strip_html(raw.get("ascii_name") or raw.get("name") or "")
described = f"{returned_accession} ({name})" if name else returned_accession
secondary_ids = raw.get("secondary_ids")
if not isinstance(secondary_ids, list):
secondary_ids = None
disclosure = {
"requested_chebi_id": requested_int,
"requested_chebi_accession": requested_accession,
}
if secondary_ids is None:
# The ontology endpoints redirect identically but do not return
# secondary_ids, so the precise relationship cannot be asserted from
# this response -- say only what this payload proves.
note = (
f"Requested {requested_accession}, but ChEBI returned "
f"{described}. Every field below describes {returned_accession}, "
f"not {requested_accession}. This endpoint does not report "
f"secondary_ids, so the exact relationship is not confirmable "
f"from this response; ChEBI normally serves the primary entry "
f"when an obsolete or secondary accession is requested. Call "
f"ChEBI_get_compound with {requested_accession} to confirm."
)
elif requested_accession in {str(s).strip().upper() for s in secondary_ids}:
note = (
f"Requested {requested_accession}, but ChEBI returned "
f"{described}. {requested_accession} is a secondary (merged or "
f"obsolete) accession of the primary entry {returned_accession}, "
f"as listed in that entry's secondary_ids. Every field below "
f"describes {returned_accession}, not {requested_accession}."
)
else:
note = (
f"Requested {requested_accession}, but ChEBI returned "
f"{described}, and {requested_accession} is NOT listed among that "
f"entry's secondary_ids. The reason for the substitution is "
f"therefore unknown and unverified. Every field below describes "
f"{returned_accession}; do not treat it as data about "
f"{requested_accession} without independent confirmation."
)
disclosure["redirect_note"] = note
if secondary_ids is not None:
# Provenance, and the evidence for the note itself. Emitted only on the
# redirect path, where it is the thing that makes the claim checkable --
# so the normal path carries no extra bytes.
disclosure["secondary_ids"] = [str(s) for s in secondary_ids]
return disclosure
[docs]
@register_tool("ChEBITool")
class ChEBITool(BaseTool):
"""
Tool for querying ChEBI (Chemical Entities of Biological Interest).
Provides compound lookup, text search, and ontology navigation
for small molecules of biological relevance.
No authentication required.
"""
[docs]
def __init__(self, tool_config: Dict[str, Any]):
super().__init__(tool_config)
self.timeout = tool_config.get("timeout", 30)
self.endpoint_type = tool_config.get("fields", {}).get(
"endpoint_type", "get_compound"
)
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the ChEBI API call."""
try:
return self._dispatch(arguments)
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"ChEBI API request timed out after {self.timeout} seconds",
}
except requests.exceptions.ConnectionError:
return {
"status": "error",
"error": "Failed to connect to ChEBI API. Check network connectivity.",
}
except requests.exceptions.HTTPError as e:
return {
"status": "error",
"error": f"ChEBI API HTTP error: {e.response.status_code}",
}
except Exception as e:
return {
"status": "error",
"error": f"Unexpected error querying ChEBI: {str(e)}",
}
[docs]
def _dispatch(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Route to appropriate endpoint based on config."""
if self.endpoint_type == "get_compound":
return self._get_compound(arguments)
elif self.endpoint_type == "search":
return self._search(arguments)
elif self.endpoint_type == "ontology_children":
return self._ontology_children(arguments)
elif self.endpoint_type == "ontology_parents":
return self._ontology_parents(arguments)
else:
return {
"status": "error",
"error": f"Unknown endpoint_type: {self.endpoint_type}",
}
[docs]
def _get_compound(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get detailed compound information by ChEBI ID."""
chebi_id = arguments.get("chebi_id", None)
if chebi_id is None:
return {
"status": "error",
"error": "chebi_id parameter is required (e.g., 15365 for aspirin)",
}
# Accept the "CHEBI:27732" CURIE form that ChEBI_search returns as
# chebi_accession, not just the bare integer — so the two tools chain
# without the caller having to strip the prefix.
chebi_id = _normalize_chebi_id(chebi_id)
url = f"{CHEBI_BASE_URL}/compound/{chebi_id}/"
response = requests.get(
url,
headers={"Accept": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
raw = response.json()
# Extract synonyms
synonyms = []
# A brand name, an IUPAC name and a KEGG synonym are not the same kind
# of name, and the kind was dropped: "8-hour bayer" (BRAND NAME, from
# ChEMBL) and "2-(acetyloxy)benzoic acid" (IUPAC NAME) arrived in one
# undifferentiated list. These rows run in parallel with `synonyms`.
synonym_details = []
names_dict = raw.get("names", {})
for name_type, name_list in names_dict.items():
if isinstance(name_list, list):
for entry in name_list:
if isinstance(entry, dict):
syn = _strip_html(entry.get("name", ""))
if syn and syn not in synonyms:
synonyms.append(syn)
synonym_details.append(
{
"name": syn,
"type": entry.get("type") or name_type,
"source": entry.get("source"),
"status": entry.get("status"),
"language_code": entry.get("language_code"),
}
)
synonym_offset = max(int(arguments.get("synonym_offset") or 0), 0)
synonym_limit = min(max(int(arguments.get("synonym_limit") or 20), 1), 500)
synonym_page = synonyms[synonym_offset : synonym_offset + synonym_limit]
synonym_detail_page = synonym_details[
synonym_offset : synonym_offset + synonym_limit
]
synonyms_truncated = synonym_offset + len(synonym_page) < len(synonyms)
# Chemical data is nested under 'chemical_data'
chem_data = raw.get("chemical_data", {})
if not isinstance(chem_data, dict):
chem_data = {}
# Structure data is under 'default_structure'
struct_data = raw.get("default_structure", {})
if not isinstance(struct_data, dict):
struct_data = {}
# Parse mass as float if string
mass_val = chem_data.get("mass", None)
if isinstance(mass_val, str):
try:
mass_val = float(mass_val)
except ValueError:
mass_val = None
mono_mass = chem_data.get("monoisotopic_mass", None)
if isinstance(mono_mass, str):
try:
mono_mass = float(mono_mass)
except ValueError:
mono_mass = None
result = {
"chebi_id": raw.get("id", chebi_id),
"chebi_accession": raw.get("chebi_accession", f"CHEBI:{chebi_id}"),
"name": _strip_html(raw.get("name", "")),
"definition": _strip_html(raw.get("definition", None)),
"stars": raw.get("stars", 0),
"formula": chem_data.get("formula", None),
"mass": mass_val,
"monoisotopic_mass": mono_mass,
"charge": chem_data.get("charge", None),
"smiles": struct_data.get("smiles", None),
"inchikey": struct_data.get("standard_inchi_key", None),
# The InChIKey is a hash; the InChI is the structure it hashes.
"inchi": struct_data.get("standard_inchi", None),
"wurcs": struct_data.get("wurcs", None),
"ascii_name": raw.get("ascii_name"),
# `secondary_ids` is deliberately NOT added here: the redirect
# disclosure emits it on a substitution, which is where it is the
# evidence for a claim, and test_chebi_obsolete_id_redirect_
# disclosed asserts a direct hit carries no disclosure key.
"roles": _roles(raw)[:ROLES_PER_COMPOUND],
"role_count": len(_roles(raw)),
"roles_truncated": len(_roles(raw)) > ROLES_PER_COMPOUND,
"compound_origins": raw.get("compound_origins") or [],
"modified_on": raw.get("modified_on"),
"is_released": raw.get("is_released"),
"created_by": raw.get("created_by") or [],
"relation_counts": {
"incoming": len(
(raw.get("ontology_relations") or {}).get("incoming_relations")
or []
),
"outgoing": len(
(raw.get("ontology_relations") or {}).get("outgoing_relations")
or []
),
},
**_accessions(raw),
"synonyms": synonym_page,
"synonym_details": synonym_detail_page,
"synonyms_total": len(synonyms),
"synonym_offset": synonym_offset,
"synonyms_returned": len(synonym_page),
"synonyms_truncated": synonyms_truncated,
"next_synonym_offset": (
synonym_offset + len(synonym_page) if synonyms_truncated else None
),
}
# ChEBI serves the primary entry when a merged/obsolete accession is
# requested. Lead with the disclosure so the substitution is the first
# thing a reader sees, not a footnote after the synonym list.
result = {**_redirect_disclosure(chebi_id, raw), **result}
return {
"status": "success",
"data": result,
"metadata": {
"source": "ChEBI",
"query": str(chebi_id),
"endpoint": "compound",
},
}
[docs]
def _search(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search ChEBI by name, formula, or keyword using advanced search."""
query = arguments.get("query", "")
limit = arguments.get("limit", 10)
if not query:
return {
"status": "error",
"error": "query parameter is required (e.g., 'glucose', 'caffeine')",
}
if limit is None:
limit = 10
try:
limit = int(limit)
except (TypeError, ValueError):
limit = 10
limit = max(0, min(limit, 100))
try:
offset = max(0, int(arguments.get("offset", 0)))
except (TypeError, ValueError):
return {"status": "error", "error": "offset must be an integer"}
# Use advanced_search endpoint for better relevance
url = f"{CHEBI_BASE_URL}/advanced_search/"
payload = {
"text_search_specification": {
"or_specification": [{"text": query, "category": "all"}]
},
"stars": [2, 3],
}
headers = {"Content-Type": "application/json", "Accept": "application/json"}
# The endpoint returns only one page (15 hits) per request, so a single
# POST can never satisfy limit > 15. Walk successive pages until we have
# `limit` compounds or the result set is exhausted.
compounds = []
total_matches = None
number_pages = None
page = offset // _SEARCH_PAGE_SIZE + 1
leading_skip = offset % _SEARCH_PAGE_SIZE
needed = leading_skip + limit
pages_fetched = 0
while pages_fetched < _SEARCH_MAX_PAGES:
try:
response = requests.post(
url,
json=payload,
params={"page": page},
headers=headers,
timeout=self.timeout,
)
response.raise_for_status()
raw = response.json()
except Exception:
# Never discard the pages already in hand: a failure while
# fetching page N > 1 degrades to "fewer results than asked
# for", not to an error. Only a failed first page is fatal.
if pages_fetched == 0:
raise
break
pages_fetched += 1
if not isinstance(raw, dict):
break
# `total` / `number_pages` are informational and may be absent on
# some responses; fall back to single-page behaviour rather than
# crashing or looping forever.
if total_matches is None and isinstance(raw.get("total"), int):
total_matches = raw["total"]
if number_pages is None and isinstance(raw.get("number_pages"), int):
number_pages = raw["number_pages"]
results = raw.get("results", [])
if not isinstance(results, list):
results = []
for hit in results:
source = hit.get("_source", {}) if isinstance(hit, dict) else {}
if not isinstance(source, dict):
source = {}
compounds.append(
{
"chebi_accession": source.get("chebi_accession", ""),
# Strip HTML tags from name (ChEBI stores
# stereochemistry as HTML).
"name": _strip_html(source.get("name", "")),
"formula": source.get("formula", None),
"mass": source.get("mass", None),
"stars": source.get("stars", None),
# The hit already carries the structure and the
# definition; a chemical search that answers a name and
# a mass and not a SMILES sends the caller back for a
# second call per hit.
"ascii_name": source.get("ascii_name", None),
"definition": _strip_html(source.get("definition", "") or ""),
"smiles": source.get("smiles", None),
"inchi": source.get("inchi", None),
"inchikey": source.get("inchikey", None),
"monoisotopic_mass": source.get("monoisotopicmass", None),
"charge": source.get("charge", None),
"relevance_score": hit.get("_score")
if isinstance(hit, dict)
else None,
}
)
if len(compounds) >= needed:
break
if not results:
break
if number_pages is not None:
if page >= number_pages:
break
elif len(results) < _SEARCH_PAGE_SIZE:
# No page metadata and a short page: this was the last one.
break
page += 1
compounds = compounds[leading_skip : leading_skip + limit]
has_more = (
offset + len(compounds) < total_matches
if total_matches is not None
else False
)
result = {
"query": query,
# `result_count` keeps its original meaning: how many compounds are
# in `compounds`. `total_matches` is how many the query matches in
# ChEBI overall, so a caller can distinguish "10 of 116" from
# "10 exist". It is None when the API omits the count.
"result_count": len(compounds),
"total_matches": total_matches,
"offset": offset,
"limit": limit,
"has_more": has_more,
"truncated": has_more,
"next_offset": offset + len(compounds) if has_more else None,
"compounds": compounds,
}
return {
"status": "success",
"data": result,
"metadata": {
"source": "ChEBI",
"query": query,
"endpoint": "advanced_search",
"pages_fetched": pages_fetched,
},
}
[docs]
def _ontology_children(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get ontology children of a ChEBI compound."""
chebi_id = arguments.get("chebi_id", None)
if chebi_id is None:
return {
"status": "error",
"error": "chebi_id parameter is required (e.g., 15365 for aspirin)",
}
chebi_id = _normalize_chebi_id(chebi_id)
url = f"{CHEBI_BASE_URL}/ontology/children/{chebi_id}/"
response = requests.get(
url,
headers={"Accept": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
raw = response.json()
# Extract relations
relations = []
ontology = raw.get("ontology_relations", {})
incoming = ontology.get("incoming_relations", [])
if isinstance(incoming, list):
for rel in incoming:
relations.append(
{
"child_id": rel.get("init_id", 0),
"child_name": rel.get("init_name", ""),
"relation_type": rel.get("relation_type", ""),
"parent_id": rel.get("final_id", 0),
"parent_name": rel.get("final_name", ""),
}
)
result = {
"chebi_id": raw.get("id", chebi_id),
"chebi_accession": raw.get("chebi_accession", f"CHEBI:{chebi_id}"),
# The response names the compound these relations belong to -- on
# the relation rows as `final_name` -- and the result did not.
"name": _strip_html(
raw.get("name")
or (incoming[0].get("final_name", "") if incoming else "")
),
"relation_count": len(relations),
"relations": relations,
}
# Same silent-substitution shape as _get_compound: /ontology/children/1/
# answers with id 18357. Disclosed through the same helper.
result = {**_redirect_disclosure(chebi_id, raw), **result}
return {
"status": "success",
"data": result,
"metadata": {
"source": "ChEBI",
"query": str(chebi_id),
"endpoint": "ontology/children",
},
}
[docs]
def _ontology_parents(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get ontology parents (is-a / has-role ancestors) of a ChEBI compound."""
chebi_id = arguments.get("chebi_id", None)
if chebi_id is None:
return {
"status": "error",
"error": "chebi_id parameter is required (e.g., 15377 for water)",
}
chebi_id = _normalize_chebi_id(chebi_id)
url = f"{CHEBI_BASE_URL}/ontology/parents/{chebi_id}/"
response = requests.get(
url,
headers={"Accept": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
raw = response.json()
# Parents are the targets of outgoing_relations (the symmetric
# counterpart of the children tool's incoming_relations).
relations = []
ontology = raw.get("ontology_relations", {})
outgoing = ontology.get("outgoing_relations", [])
if isinstance(outgoing, list):
for rel in outgoing:
relations.append(
{
"relation_type": rel.get("relation_type", ""),
"parent_id": rel.get("final_id", 0),
"parent_name": _strip_html(rel.get("final_name", "")),
}
)
result = {
"chebi_id": raw.get("id", chebi_id),
"chebi_accession": raw.get("chebi_accession", f"CHEBI:{chebi_id}"),
# `init_name` on every outgoing relation is this compound's own
# name, and the result never said whose parents these are.
"name": _strip_html(
raw.get("name")
or (outgoing[0].get("init_name", "") if outgoing else "")
),
"relation_count": len(relations),
"relations": relations,
}
# Same silent-substitution shape as _get_compound: /ontology/parents/1/
# answers with id 18357. Disclosed through the same helper.
result = {**_redirect_disclosure(chebi_id, raw), **result}
return {
"status": "success",
"data": result,
"metadata": {
"source": "ChEBI",
"query": str(chebi_id),
"endpoint": "ontology/parents",
},
}