Source code for tooluniverse.efo_tool
from __future__ import annotations
from typing import Any, Dict, Optional
import urllib.parse
import requests
from .base_tool import BaseTool
from .http_utils import request_with_retry
from .tool_registry import register_tool
[docs]
@register_tool("EFOTool")
class EFOTool(BaseTool):
"""
Tool to lookup Experimental Factor Ontology (EFO) IDs for diseases via the
EMBL-EBI OLS API.
"""
[docs]
def __init__(self, tool_config, base_url="https://www.ebi.ac.uk/ols4/api/search"):
super().__init__(tool_config)
self.base_url = base_url
[docs]
def run(self, arguments):
disease = arguments.get("disease")
rows = arguments.get("rows", 1)
if not disease:
return {"status": "error", "error": "`disease` parameter is required."}
return self._search(disease, rows)
[docs]
def _search(self, disease, rows):
# Fix-20B-2: OLS4's default relevance ranking scores hits on the full
# indexed text (including free-text descriptions), not just the
# term's own name. For an abbreviation like "PCOS" this put a term
# whose *description* merely mentions "PCOS" in passing (EFO_0005187
# "C-peptide measurement") ahead of the actual term for the disease
# (MONDO_0008487 "polycystic ovary syndrome", which lists "PCOS" as
# an exact_synonym) -- confirmed live against the OLS4 API. Silently
# returning the wrong disease's EFO ID is worse than a fetch failure,
# since callers treat it as authoritative and chain further lookups
# on it. Always fetch enough candidates to see past a mis-ranked top
# hit, then prefer any candidate whose own label/exact_synonym is a
# case-insensitive exact match to the query over the raw OLS order.
fetch_rows = max(rows, 10)
params = {"ontology": "efo", "q": disease, "rows": fetch_rows}
try:
response = requests.get(self.base_url, params=params, timeout=20)
response.raise_for_status()
except requests.RequestException as e:
return {
"status": "error",
"error": "OLS API request failed.",
"details": str(e),
}
data = response.json().get("response", {})
docs = data.get("docs", [])
if not docs:
return None
def _is_exact_match(doc):
target = disease.strip().lower()
label = (doc.get("label") or "").strip().lower()
if label == target:
return True
synonyms = doc.get("exact_synonyms") or []
return any(target == str(s).strip().lower() for s in synonyms)
exact = [d for d in docs if _is_exact_match(d)]
rest = [d for d in docs if d not in exact]
docs = (exact + rest)[:rows]
if rows == 1:
doc = docs[0]
return {"efo_id": doc.get("short_form"), "name": doc.get("label")}
return [
{"efo_id": doc.get("short_form"), "name": doc.get("label")} for doc in docs
]
[docs]
@register_tool("OLSRESTTool")
class OLSRESTTool(BaseTool):
"""
Generic tool for the EMBL-EBI OLS v4 REST API.
This is a JSON-config driven tool. Each tool config supplies:
- fields.kind: "search" | "term" | "children" | "ontology" | "ontologies"
- fields.base_url: optional override (defaults to OLS4 API root)
- fields.ontology_id: optional ontology scope (e.g., "efo")
"""
DEFAULT_BASE_URL = "https://www.ebi.ac.uk/ols4/api"
[docs]
@staticmethod
def _double_urlencode(value: str) -> str:
# OLS term endpoints use the term IRI as a path segment and require
# double URL-encoding to avoid routing issues with slashes.
once = urllib.parse.quote(value, safe="")
return urllib.parse.quote(once, safe="")
[docs]
@staticmethod
def _obo_id_to_efo_iri(obo_id: str) -> str:
# CURIEs like "EFO:0000400" or "MONDO:0004993" come from search results.
# Convert to the IRI used by the OLS term endpoints. EFO terms live under
# the EBI namespace; everything else EFO imports (MONDO, HP, GO, Orphanet,
# ...) lives under its native OBO/ORDO namespace — mapping only EFO: meant
# non-EFO disease terms (now predominantly MONDO) silently returned nothing.
if ":" not in obo_id:
return obo_id
prefix, num = obo_id.split(":", 1)
p = prefix.upper()
if p == "EFO":
return f"http://www.ebi.ac.uk/efo/EFO_{num}"
if p in ("ORPHA", "ORPHANET"):
return f"http://www.orpha.net/ORDO/Orphanet_{num}"
# MONDO, HP, GO, CHEBI, UBERON, CL, NCIT, DOID, ... use the OBO PURL base.
return f"http://purl.obolibrary.org/obo/{p}_{num}"
[docs]
def _resolve_term_iri(
self, *, iri: Optional[str] = None, obo_id: Optional[str] = None
) -> Optional[str]:
if iri:
return iri
if obo_id:
return self._obo_id_to_efo_iri(obo_id)
return None
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
fields = self.tool_config.get("fields") or {}
kind = fields.get("kind")
base_url = fields.get("base_url") or self.DEFAULT_BASE_URL
ontology_id = fields.get("ontology_id")
timeout_s = fields.get("timeout", 20)
try:
if kind == "search":
query = arguments.get("query")
rows = arguments.get("rows", 10)
if not query:
return {
"status": "error",
"error": "`query` parameter is required.",
}
params: Dict[str, Any] = {"q": query, "rows": rows}
if ontology_id:
params["ontology"] = ontology_id
url = f"{base_url}/search"
resp = request_with_retry(
requests, "GET", url, params=params, timeout=timeout_s
)
if not (200 <= resp.status_code < 300):
return {
"status": "error",
"url": resp.url,
"status_code": resp.status_code,
"error": "HTTP request failed",
"detail": resp.text[:2000],
}
j = resp.json()
docs = (j.get("response") or {}).get("docs") or []
terms = [
{
"iri": d.get("iri"),
"obo_id": d.get("obo_id"),
"short_form": d.get("short_form"),
"label": d.get("label"),
"description": d.get("description"),
"ontology_name": d.get("ontology_name"),
"ontology_prefix": d.get("ontology_prefix"),
"type": d.get("type"),
}
for d in docs
]
result = {
"status": "success",
"url": resp.url,
"count": len(terms),
"terms": terms,
}
return {"status": "success", "data": result}
if kind in {"term", "children", "descendants"}:
if not ontology_id:
return {
"status": "error",
"error": "Tool misconfigured: missing fields.ontology_id",
}
iri = self._resolve_term_iri(
iri=arguments.get("iri"), obo_id=arguments.get("obo_id")
)
if not iri:
return {
"status": "error",
"error": "Provide either `iri` or `obo_id`.",
}
encoded = self._double_urlencode(iri)
term_url = f"{base_url}/ontologies/{ontology_id}/terms/{encoded}"
if kind == "term":
resp = request_with_retry(
requests, "GET", term_url, timeout=timeout_s
)
if not (200 <= resp.status_code < 300):
return {
"status": "error",
"url": resp.url,
"status_code": resp.status_code,
"error": "HTTP request failed",
"detail": resp.text[:2000],
}
t = resp.json()
annotation = t.get("annotation") or {}
term = {
"iri": t.get("iri"),
"obo_id": t.get("obo_id"),
"short_form": t.get("short_form"),
"label": t.get("label"),
"description": t.get("description"),
"synonyms": t.get("synonyms"),
"has_children": t.get("has_children"),
"is_obsolete": t.get("is_obsolete"),
"ontology_name": t.get("ontology_name"),
"ontology_prefix": t.get("ontology_prefix"),
# An obsolete term is only half an answer without
# what replaced it. EFO:0000400 comes back as
# "obsolete_diabetes mellitus" with `is_obsolete`
# true, and the record names the replacement and the
# ids merged into it -- in `annotation`, which was
# not read.
"term_replaced_by": annotation.get("term_replaced_by")
or t.get("term_replaced_by"),
"alternative_ids": annotation.get("has_alternative_id")
or [],
"obsolescence_reason": annotation.get(
"has_obsolescence_reason"
),
# Each synonym with the scope OBO gives it: an exact
# synonym can be substituted for the label and a
# related one cannot, and `synonyms` is the flattened
# list of both.
"synonym_scopes": [
{
"name": item.get("name"),
"scope": item.get("scope"),
"type": item.get("type"),
"xrefs": item.get("xrefs") or [],
}
for item in (t.get("obo_synonym") or [])
if isinstance(item, dict)
],
# The term's cross-references to other vocabularies,
# which is how an EFO id is carried into a resource
# that does not speak EFO.
"cross_references": annotation.get(
"database_cross_reference"
)
or [],
"obo_xrefs": [
{
"database": item.get("database"),
"id": item.get("id"),
"description": item.get("description"),
"url": item.get("url"),
}
for item in (t.get("obo_xref") or [])
if isinstance(item, dict)
],
"obo_definition_citations": t.get(
"obo_definition_citation"
)
or [],
"in_subset": t.get("in_subset") or [],
"lang": t.get("lang"),
}
result = {"status": "success", "url": resp.url, "term": term}
return {"status": "success", "data": result}
# children (direct subclasses) or descendants (full subtree)
size = arguments.get("size", 20)
params = {"size": size}
endpoint = "descendants" if kind == "descendants" else "children"
resp = request_with_retry(
requests,
"GET",
f"{term_url}/{endpoint}",
params=params,
timeout=timeout_s,
)
if not (200 <= resp.status_code < 300):
return {
"status": "error",
"url": resp.url,
"status_code": resp.status_code,
"error": "HTTP request failed",
"detail": resp.text[:2000],
}
j = resp.json()
rows = ((j.get("_embedded") or {}).get("terms")) or []
out = [
{
"iri": c.get("iri"),
"obo_id": c.get("obo_id"),
"short_form": c.get("short_form"),
"label": c.get("label"),
# A descendant list of labels cannot be filtered or
# followed. The same rows carry the definition, the
# synonyms, whether the term is itself obsolete, and
# which ontology it came from -- which matters here
# because OLS returns descendants from importing
# ontologies too.
"description": c.get("description") or [],
"synonyms": c.get("synonyms") or [],
"is_obsolete": c.get("is_obsolete"),
"has_children": c.get("has_children"),
"ontology_name": c.get("ontology_name"),
"ontology_prefix": c.get("ontology_prefix"),
"alternative_ids": (c.get("annotation") or {}).get(
"has_alternative_id"
)
or [],
"synonym_scopes": [
{
"name": item.get("name"),
"scope": item.get("scope"),
"type": item.get("type"),
"xrefs": item.get("xrefs") or [],
}
for item in (c.get("obo_synonym") or [])
if isinstance(item, dict)
],
"obo_definition_citations": c.get("obo_definition_citation")
or [],
"ontology_iri": c.get("ontology_iri"),
"lang": c.get("lang"),
}
for c in rows
]
if kind == "descendants":
total = (j.get("page") or {}).get("totalElements")
result = {
"status": "success",
"url": resp.url,
"count": len(out),
"total": total,
"descendants": out,
}
# A zero subtree is ambiguous: the term may be a leaf, obsolete,
# or (for disease terms) routed through a non-EFO ontology where
# OLS does not expose descendants here. Flag it so callers don't
# read "no subtypes" into what is really "no descendants exposed".
if not total:
result["note"] = (
"No descendants returned. The term may be a leaf or "
"obsolete, or its subclasses live in another ontology "
"(e.g. EFO disease terms whose children are MONDO/Orphanet "
"nodes). Verify the term is current and try its native "
"ontology if this is a disease class."
)
return {"status": "success", "data": result}
result = {
"status": "success",
"url": resp.url,
"count": len(out),
"children": out,
}
return {"status": "success", "data": result}
if kind == "ontology":
if not ontology_id:
return {
"status": "error",
"error": "Tool misconfigured: missing fields.ontology_id",
}
url = f"{base_url}/ontologies/{ontology_id}"
resp = request_with_retry(requests, "GET", url, timeout=timeout_s)
if not (200 <= resp.status_code < 300):
return {
"status": "error",
"url": resp.url,
"status_code": resp.status_code,
"error": "HTTP request failed",
"detail": resp.text[:2000],
}
o = resp.json()
ontology = {
"ontologyId": o.get("ontologyId"),
"version": o.get("version"),
"status": o.get("status"),
"numberOfTerms": o.get("numberOfTerms"),
"updated": o.get("updated"),
"title": (o.get("config") or {}).get("title"),
"description": (o.get("config") or {}).get("description"),
"homepage": (o.get("config") or {}).get("homepage"),
"preferredPrefix": (o.get("config") or {}).get("preferredPrefix"),
"loaded": o.get("loaded"),
"numberOfProperties": o.get("numberOfProperties"),
"numberOfIndividuals": o.get("numberOfIndividuals"),
"languages": o.get("languages")
or (o.get("config") or {}).get("languages")
or [],
}
result = {"status": "success", "url": resp.url, "ontology": ontology}
return {"status": "success", "data": result}
if kind == "ontologies":
size = arguments.get("size", 20)
url = f"{base_url}/ontologies"
resp = request_with_retry(
requests, "GET", url, params={"size": size}, timeout=timeout_s
)
if not (200 <= resp.status_code < 300):
return {
"status": "error",
"url": resp.url,
"status_code": resp.status_code,
"error": "HTTP request failed",
"detail": resp.text[:2000],
}
j = resp.json()
onts = ((j.get("_embedded") or {}).get("ontologies")) or []
out = [
{
"ontologyId": o.get("ontologyId"),
"title": (o.get("config") or {}).get("title"),
"numberOfTerms": o.get("numberOfTerms"),
"status": o.get("status"),
# What the ontology IS, which version is loaded and
# when: a list of ids and term counts cannot be chosen
# between. `languages` matters for a term lookup
# against a multilingual ontology, where a label comes
# back in whichever language was asked for.
"description": (o.get("config") or {}).get("description"),
"preferredPrefix": (o.get("config") or {}).get(
"preferredPrefix"
),
"homepage": (o.get("config") or {}).get("homepage"),
"version": o.get("version"),
"updated": o.get("updated"),
"loaded": o.get("loaded"),
"numberOfProperties": o.get("numberOfProperties"),
"numberOfIndividuals": o.get("numberOfIndividuals"),
"languages": o.get("languages")
or (o.get("config") or {}).get("languages")
or [],
}
for o in onts
]
result = {
"status": "success",
"url": resp.url,
"count": len(out),
"ontologies": out,
}
return {"status": "success", "data": result}
return {
"status": "error",
"error": "Tool misconfigured: unknown fields.kind",
"detail": f"Unsupported kind={kind!r}",
}
except Exception as e:
return {"status": "error", "error": str(e)}