Source code for tooluniverse.hpo_tool
# hpo_tool.py
"""
Human Phenotype Ontology (HPO) API tool for ToolUniverse.
HPO provides a standardized vocabulary of phenotypic abnormalities
encountered in human disease. Each term describes a phenotypic
feature (sign, symptom, or finding) and is organized in a directed
acyclic graph (DAG) hierarchy.
API: https://ontology.jax.org/api/
No authentication required. Free public access.
"""
import requests
from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool
HPO_BASE_URL = "https://ontology.jax.org/api/hp"
HPO_ANNOTATION_URL = "https://ontology.jax.org/api/network/annotation"
def _hpo_medical_actions(payload):
"""MAxO curated treatments, as the JAX annotation endpoint reports them.
Written once because two handlers read the same endpoint: the disease
annotation tool already returned these, and the phenotype association
tools read the same payload for an obsolescence probe and then discarded
them -- so a phenotype's medical actions were unreachable through any
tool while sitting in a response already paid for.
"""
out = []
for action in payload.get("medicalActions") or []:
if not isinstance(action, dict):
continue
out.append(
{
"id": action.get("id"),
"name": action.get("name"),
"relations": action.get("relations") or [],
"sources": action.get("sources") or [],
"targets": [
{"id": t.get("id"), "name": t.get("name")}
for t in (action.get("targets") or [])
if isinstance(t, dict)
],
}
)
return out
def _normalize_hpo_id(term_id: str) -> str:
"""Normalize an HPO term id to the canonical colon CURIE 'HP:0001250'.
Accepts the underscore form 'HP_0001250' (which OpenTargets and other tools
emit, e.g. phenotypeHPO.id) and a bare numeric id '0001250'. Without this,
'HP_0001250' failed the old `startswith("HP:")` check and was turned into
'HP:HP_0001250', causing an HPO API 404 -- a cross-tool chaining break in the
OpenTargets-phenotype -> HPO_get_term path a clinician follows."""
tid = str(term_id).strip()
# Underscore CURIE from OpenTargets et al. -> colon CURIE.
if tid.upper().startswith("HP_"):
tid = "HP:" + tid[3:]
elif not tid.upper().startswith("HP:"):
tid = f"HP:{tid}"
# Canonicalize the prefix case ('hp:' -> 'HP:').
return "HP:" + tid.split(":", 1)[1] if ":" in tid else tid
def _merge_disclosure(requested_id: str, resolved_id: Any) -> Dict[str, Any]:
"""Report an upstream merge of `requested_id` into `resolved_id`, else {}.
The JAX ontology API silently serves an obsolete/merged ID's replacement
record -- no obsolescence flag, no 404, no redirect marker -- so the id
mismatch is the only evidence, and every caller that resolves a term the
user named needs the same three keys back.
"""
if not resolved_id or resolved_id == requested_id:
return {}
return {
"requested_id": requested_id,
"resolved_id": resolved_id,
"note": (
f"The requested term ID ({requested_id}) differs from the "
f"returned term's ID ({resolved_id}). This usually means "
f"{requested_id} is obsolete or was merged into {resolved_id} "
"upstream. Use get_phenotype_by_HPO_ID for an explicit "
"'deprecated' flag, or HPO_search_terms to find the "
"current preferred term."
),
}
TRANSLATIONS_PER_TERM = 25
def _translations(payload) -> list:
"""The term's labels in other languages, with how official each is.
HPO publishes nine translations for HP:0002069 and marks each OFFICIAL
or CANDIDATE. The status is the point: a CANDIDATE label is a proposal,
not a name the ontology stands behind, and a clinician matching a
Japanese or Spanish record needs to know which it is. The term handler
returned `language` and `name` and dropped the status and the
translated definition; the hierarchy handler returned no translations at
all, although `translations` is one of the four keys its endpoint sends.
"""
rows = []
for item in (payload.get("translations") or [])[:TRANSLATIONS_PER_TERM]:
if not isinstance(item, dict) or not item.get("name"):
continue
rows.append(
{
"language": item.get("language"),
"name": item.get("name"),
"definition": item.get("definition") or None,
# OFFICIAL or CANDIDATE.
"status": item.get("status"),
}
)
return rows
[docs]
@register_tool("HPOTool")
class HPOTool(BaseTool):
"""
Tool for querying the Human Phenotype Ontology (HPO) at JAX.
HPO provides structured phenotype terms used in clinical genetics,
rare disease research, and differential diagnosis. Each term has
definitions, synonyms, cross-references, and hierarchical relationships.
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", "get_term")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the HPO API call."""
try:
return self._query(arguments)
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"HPO API request timed out after {self.timeout} seconds",
}
except requests.exceptions.ConnectionError:
return {
"status": "error",
"error": "Failed to connect to HPO API. Check network connectivity.",
}
except requests.exceptions.HTTPError as e:
return {
"status": "error",
"error": f"HPO API HTTP error: {e.response.status_code}",
}
except Exception as e:
return {
"status": "error",
"error": f"Unexpected error querying HPO: {str(e)}",
}
[docs]
def _query(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Route to appropriate HPO endpoint."""
if self.endpoint == "get_term":
return self._get_term(arguments)
elif self.endpoint == "search_terms":
return self._search_terms(arguments)
elif self.endpoint == "get_term_hierarchy":
return self._get_term_hierarchy(arguments)
elif self.endpoint == "get_associated_genes":
return self._get_associations(arguments, "genes")
elif self.endpoint == "get_associated_diseases":
return self._get_associations(arguments, "diseases")
elif self.endpoint == "get_disease_annotations":
return self._get_disease_annotations(arguments)
else:
return {"status": "error", "error": f"Unknown endpoint: {self.endpoint}"}
[docs]
def _get_disease_annotations(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get the full HPO annotation profile for a disease.
Uses the JAX network-annotation endpoint keyed by a disease ID
(OMIM/ORPHA/DECIPHER). Returns curated MAxO medical actions
(treatments / management with TREATS / PREVENTS relations and the
phenotypes they target), phenotypes grouped by body system, and
associated genes.
"""
disease_id = arguments.get("disease_id") or arguments.get("id", "")
disease_id = str(disease_id).strip()
if not disease_id:
return {
"status": "error",
"error": (
"disease_id parameter is required "
"(e.g., 'OMIM:154700', 'ORPHA:558', 'DECIPHER:...')"
),
}
url = f"{HPO_ANNOTATION_URL}/{disease_id}"
response = requests.get(url, timeout=self.timeout)
if response.status_code == 404:
return {
"status": "error",
"error": (
f"No HPO disease annotation found for '{disease_id}'. "
"Use an OMIM / ORPHA / DECIPHER disease ID "
"(e.g., 'OMIM:154700')."
),
}
response.raise_for_status()
data = response.json()
# --- Medical actions (MAxO curated treatments / management) ---
medical_actions = _hpo_medical_actions(data)
# --- Phenotypes grouped by body system ---
categories = []
raw_categories = data.get("categories") or {}
if isinstance(raw_categories, dict):
for system, phenotypes in raw_categories.items():
pheno_list = []
if isinstance(phenotypes, list):
for p in phenotypes:
if isinstance(p, dict):
meta = p.get("metadata") or {}
pheno_list.append(
{
"id": p.get("id"),
"name": p.get("name"),
"frequency": meta.get("frequency") or None,
"onset": meta.get("onset") or None,
}
)
categories.append(
{
"body_system": system,
"phenotype_count": len(pheno_list),
"phenotypes": pheno_list,
}
)
categories.sort(key=lambda c: c["phenotype_count"], reverse=True)
# --- Associated genes ---
genes = []
for g in data.get("genes") or []:
if isinstance(g, dict):
genes.append({"id": g.get("id"), "name": g.get("name")})
disease = data.get("disease") or {}
total_phenotypes = sum(c["phenotype_count"] for c in categories)
return {
"status": "success",
"data": {
"disease": {
"id": disease.get("id"),
"name": disease.get("name"),
"mondo_id": disease.get("mondoId"),
"description": disease.get("description"),
},
"medical_actions": medical_actions,
"categories": categories,
"genes": genes,
},
"metadata": {
"source": "HPO (JAX Ontology) network annotation",
"disease_id": disease.get("id") or disease_id,
"total_medical_actions": len(medical_actions),
"total_body_systems": len(categories),
"total_phenotypes": total_phenotypes,
"total_genes": len(genes),
},
}
[docs]
def _get_associations(self, arguments: Dict[str, Any], kind: str) -> Dict[str, Any]:
"""Get genes or diseases annotated to an HPO phenotype term.
Uses the JAX network-annotation endpoint, which returns the genes,
diseases, assays and medical actions linked to a phenotype in a
single response (the endpoint itself has no server-side paging).
"""
term_id = arguments.get("term_id", "")
if not term_id:
return {
"status": "error",
"error": "term_id parameter is required (e.g., 'HP:0001250')",
}
term_id = _normalize_hpo_id(term_id)
try:
limit = int(arguments.get("limit", 50))
except (TypeError, ValueError):
limit = 50
limit = max(1, min(limit, 500))
# Fix-19C-1/19C-2: a phenotype term can be linked to thousands of
# genes/diseases (confirmed live: HP:0001263 "Global developmental
# delay" has 2077 genes), but `limit` is capped at 500 and this
# endpoint has no server-side paging -- previously the tool always
# sliced from position 0, so well-known genes/diseases sorted past
# position 500 by the API (e.g. SHANK3, MECP2 for that term) were
# permanently unreachable with no way to page further. The full list
# is already in memory from the one API call below, so paging is a
# free client-side slice -- no extra round-trip needed.
try:
offset = int(arguments.get("offset", 0) or 0)
except (TypeError, ValueError):
offset = 0
offset = max(0, offset)
# The annotation endpoint lives under /api/network/, not /api/hp/.
url = f"{HPO_ANNOTATION_URL}/{term_id}"
response = requests.get(url, timeout=self.timeout)
response.raise_for_status()
data = response.json()
items = data.get(kind) or []
total = len(items)
trimmed = []
for it in items[offset : offset + limit]:
entry = {"id": it.get("id"), "name": it.get("name")}
if kind == "diseases":
entry["mondo_id"] = it.get("mondoId")
trimmed.append(entry)
has_more = offset + len(trimmed) < total
metadata = {
"source": "HPO (JAX Ontology) network annotation",
"term_id": term_id,
"total": total,
"offset": offset,
"limit": limit,
"returned": len(trimmed),
"has_more": has_more,
"next_offset": offset + len(trimmed) if has_more else None,
"result_set_complete": offset == 0 and not has_more,
}
# An obsolete/merged term comes back as HTTP 200 with empty arrays,
# indistinguishable from a live term nothing is annotated to. The one
# payload carries genes, diseases, assays and medicalActions together,
# so a non-empty sibling array already proves the term resolves and the
# probe is only worth paying for when all four are empty.
if not any(
data.get(k) for k in ("genes", "diseases", "assays", "medicalActions")
):
metadata.update(self._merge_metadata(term_id))
# One request answers genes, diseases, assays AND medical actions, and
# the counts are already in hand from the probe above. A caller asking
# for genes could not see that the same phenotype has 3,136 diseases
# and 14 curated medical actions behind it without asking again.
metadata["also_in_this_response"] = {
key: len(data.get(key) or [])
for key in ("genes", "diseases", "assays", "medicalActions")
if key != kind
}
return {
"status": "success",
"data": {
kind: trimmed,
# The only route to a PHENOTYPE's medical actions: the disease
# tool reports them per disease, and nothing else exposes them.
"medical_actions": _hpo_medical_actions(data),
},
"metadata": metadata,
}
[docs]
def _merge_metadata(self, term_id: str) -> Dict[str, Any]:
"""Report the replacement term when `term_id` was merged, else {}.
Capped below `self.timeout`: the rows this accompanies are already
final, so a slow JAX must not stall them.
"""
try:
response = requests.get(
f"{HPO_BASE_URL}/terms/{term_id}", timeout=min(self.timeout, 10)
)
response.raise_for_status()
resolved_id = response.json().get("id")
except Exception:
return {}
return _merge_disclosure(term_id, resolved_id)
[docs]
def _get_term(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get detailed information about an HPO term by its ID."""
term_id = arguments.get("term_id", "")
if not term_id:
return {
"status": "error",
"error": "term_id parameter is required (e.g., 'HP:0001250')",
}
# Normalize the ID format (accepts HP:xxx, HP_xxx, and bare digits)
term_id = _normalize_hpo_id(term_id)
url = f"{HPO_BASE_URL}/terms/{term_id}"
response = requests.get(url, timeout=self.timeout)
response.raise_for_status()
data = response.json()
result = {
"id": data.get("id"),
"name": data.get("name"),
"definition": data.get("definition"),
"comment": data.get("comment"),
"synonyms": data.get("synonyms", []),
"synonym_count": len(data.get("synonyms", []) or []),
"descendant_count": data.get("descendantCount"),
"xrefs": data.get("xrefs", []),
"xref_count": len(data.get("xrefs", []) or []),
}
result["translations"] = _translations(data)
result["translation_count"] = len(data.get("translations") or [])
result["translations_truncated"] = len(
data.get("translations") or []
) > len(result["translations"])
metadata = {
"source": "HPO (JAX Ontology)",
"term_id": term_id,
}
# The record is already in hand, so this costs no extra request
# (confirmed live: querying the obsolete HP:0006887 silently returns
# HP:0001249's full record).
metadata.update(_merge_disclosure(term_id, result["id"]))
return {
"status": "success",
"data": result,
"metadata": metadata,
}
[docs]
def _search_terms(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search for HPO terms by keyword."""
query = arguments.get("query", "")
if not query:
return {"status": "error", "error": "query parameter is required"}
# Coerce max_results robustly: the schema allows integer|null, so a
# caller may omit it, pass null, or pass an out-of-range value. Using
# the raw value directly would crash on None (None > 50) and let
# negatives/zero through.
try:
max_results = int(arguments.get("max_results") or 10)
except (TypeError, ValueError):
max_results = 10
max_results = max(1, min(max_results, 50))
try:
offset = max(0, int(arguments.get("offset", 0)))
except (TypeError, ValueError):
return {"status": "error", "error": "offset must be an integer"}
url = f"{HPO_BASE_URL}/search"
# The JAX ontology search endpoint sizes the page with `limit`. The
# previous `max` key was silently ignored, capping every result set at
# the API default of 10 regardless of the requested count.
page = offset // max_results
leading_skip = offset % max_results
params = {"q": query, "limit": max_results, "page": page}
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
data = response.json()
terms = data.get("terms", [])
if not isinstance(terms, list):
terms = []
total_available = data.get("totalCount", len(terms))
# An arbitrary offset can start part-way through an upstream page. Fetch
# the following page when needed so the requested page remains full.
if leading_skip and offset + max_results < total_available:
next_params = {"q": query, "limit": max_results, "page": page + 1}
next_response = requests.get(url, params=next_params, timeout=self.timeout)
next_response.raise_for_status()
next_terms = next_response.json().get("terms", [])
if isinstance(next_terms, list):
terms.extend(next_terms)
# Defensive truncation in case an upstream change ever returns more
# rows than requested.
terms = terms[leading_skip : leading_skip + max_results]
results = []
for term in terms:
results.append(
{
"id": term.get("id"),
"name": term.get("name"),
"definition": term.get("definition"),
"descendant_count": term.get("descendantCount"),
"synonyms": term.get("synonyms", []) or [],
"synonym_count": len(term.get("synonyms", []) or []),
# What the term's definition rests on, and what the
# curators said about it. A definition with no reference
# cannot be followed to the paper it came from.
"publication_references": term.get("publicationReferences") or [],
"comment": term.get("comment"),
"xrefs": term.get("xrefs") or [],
}
)
has_more = offset + len(results) < total_available
return {
"status": "success",
"data": results,
"metadata": {
"source": "HPO (JAX Ontology)",
"query": query,
"total_results": len(results),
# Total matches available across all pages (the API reports this
# as `totalCount`), so callers can see more terms exist beyond
# the returned page instead of assuming this page is exhaustive.
"total_available": total_available,
"offset": offset,
"limit": max_results,
"has_more": has_more,
"truncated": has_more,
"next_offset": offset + len(results) if has_more else None,
},
}
[docs]
def _get_term_hierarchy(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get parent and child terms in the HPO hierarchy."""
term_id = arguments.get("term_id", "")
if not term_id:
return {
"status": "error",
"error": "term_id parameter is required (e.g., 'HP:0001250')",
}
term_id = _normalize_hpo_id(term_id)
direction = arguments.get("direction", "children")
url = f"{HPO_BASE_URL}/terms/{term_id}/{direction}"
response = requests.get(url, timeout=self.timeout)
response.raise_for_status()
data = response.json()
results = []
if isinstance(data, list):
for term in data:
results.append(
{
"id": term.get("id"),
"name": term.get("name"),
# How many terms sit under this one. Walking a
# hierarchy without it means expanding every child to
# find out which branch is worth expanding.
"descendant_count": term.get("descendantCount"),
"definition": term.get("definition"),
# One of the four keys this endpoint answers, and
# the row carried none of it.
"translations": _translations(term),
"translation_count": len(term.get("translations") or []),
}
)
return {
"status": "success",
"data": results,
"metadata": {
"source": "HPO (JAX Ontology)",
"term_id": term_id,
"direction": direction,
"total_results": len(results),
},
}