# interpro_entry_tool.py
"""
InterPro Entry API tool for ToolUniverse.
Provides access to InterPro REST API for retrieving domain/family
entries associated with a protein, and searching InterPro entries
by keyword.
API: https://www.ebi.ac.uk/interpro/api/
No authentication required. Free public access.
"""
import requests
from typing import Dict, Any
from urllib.parse import parse_qs, urlparse
from .base_tool import BaseTool
INTERPRO_BASE_URL = "https://www.ebi.ac.uk/interpro/api"
def _json_or_none(response):
"""Parse a JSON response body, returning None when there is no body.
The InterPro API reports an empty result set as HTTP 204 No Content with a
zero-length body instead of a 200 carrying ``{"count": 0, "results": []}``.
``raise_for_status()`` treats 204 as success, so calling ``.json()`` on it
raises a JSONDecodeError that used to escape as an opaque
"Unexpected error querying InterPro API" message -- callers could not tell
"nothing matched" from "the tool is broken".
"""
if response.status_code == 204:
return None
body = response.content
if not body or not body.strip():
return None
return response.json()
def _match_locations(protein_rows: list) -> list:
"""Where on a protein an InterPro entry matches.
`entry_protein_locations` is what "this protein contains this domain"
actually means: the residue ranges, each possibly in several fragments,
with the signature model and score behind the call. The twin endpoint in
the sibling module builds the same shape and the two must agree -- a
caller joining a domain's proteins to a protein's domains reads both.
"""
locations = []
for row in protein_rows or []:
if not isinstance(row, dict):
continue
for location in row.get("entry_protein_locations") or []:
if not isinstance(location, dict):
continue
locations.append(
{
"fragments": [
{
"start": fragment.get("start"),
"end": fragment.get("end"),
"status": fragment.get("dc-status"),
}
for fragment in (location.get("fragments") or [])
if isinstance(fragment, dict)
],
"model": location.get("model"),
"score": location.get("score"),
"representative": location.get("representative"),
}
)
return locations
[docs]
class InterProEntryTool(BaseTool):
"""
Tool for InterPro entry API providing protein-to-domain mappings
and domain/family search.
Complements existing InterPro tools by offering reverse lookups
(protein -> all domains) and keyword-based entry search.
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", "entries_for_protein")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the InterPro entry API call."""
try:
return self._query(arguments)
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"InterPro API timed out after {self.timeout}s",
}
except requests.exceptions.ConnectionError:
return {"status": "error", "error": "Failed to connect to InterPro API"}
except requests.exceptions.HTTPError as e:
code = e.response.status_code if e.response is not None else "unknown"
if code == 404:
return {
"status": "error",
"error": f"Not found: {arguments.get('accession', arguments.get('query', ''))}",
}
return {"status": "error", "error": f"InterPro API HTTP error: {code}"}
except Exception as e:
return {
"status": "error",
"error": f"Unexpected error querying InterPro API: {str(e)}",
}
[docs]
def _query(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Route to appropriate endpoint."""
if self.endpoint == "entries_for_protein":
return self._get_entries_for_protein(arguments)
elif self.endpoint == "search_entries":
return self._search_entries(arguments)
else:
return {"status": "error", "error": f"Unknown endpoint: {self.endpoint}"}
[docs]
@staticmethod
def _next_cursor(data: Dict[str, Any]):
next_url = data.get("next")
if not next_url:
return None
values = parse_qs(urlparse(next_url).query).get("cursor", [])
return values[0] if values else None
[docs]
@staticmethod
def _add_cursor(params: Dict[str, Any], arguments: Dict[str, Any]) -> None:
cursor = arguments.get("cursor")
if cursor:
params["cursor"] = cursor
[docs]
@staticmethod
def _page_controls(arguments: Dict[str, Any], default: int, maximum: int):
page_size = arguments.get("page_size", default)
if page_size is None:
page_size = default
if (
isinstance(page_size, bool)
or not isinstance(page_size, int)
or not 1 <= page_size <= maximum
):
return {
"status": "error",
"error": f"page_size must be an integer from 1 to {maximum}",
}
cursor = arguments.get("cursor")
if cursor is not None and (not isinstance(cursor, str) or not cursor.strip()):
return {
"status": "error",
"error": "cursor must be a non-empty string when supplied",
}
return page_size, cursor
[docs]
def _get_entries_for_protein(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get all InterPro domain/family entries annotated on a protein."""
accession = arguments.get("accession", "")
if not accession:
return {
"status": "error",
"error": "accession parameter is required (UniProt accession, e.g., 'P04637')",
}
url = f"{INTERPRO_BASE_URL}/entry/interpro/protein/uniprot/{accession}"
controls = self._page_controls(arguments, 100, 200)
if isinstance(controls, dict):
return controls
page_size, cursor = controls
params = {"page_size": page_size}
self._add_cursor(params, arguments)
response = requests.get(
url,
params=params,
headers={"Accept": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
# HTTP 204 / empty body == "this protein has no InterPro entries".
data = _json_or_none(response) or {}
results = data.get("results", [])
entries = []
for r in results:
meta = r.get("metadata", {})
# The InterPro protein-entries endpoint returns metadata.name as a
# plain string; keep the dict guard in case the API shape changes.
name_obj = meta.get("name", "")
name = (
name_obj.get("name", "")
if isinstance(name_obj, dict)
else (str(name_obj) if name_obj else "")
)
# Where on the protein the entry matches, which is the point of
# asking what entries a protein has: the CDD cd08367 match on P53
# runs 109-288, and the row said only that the entry exists. The
# response carries the fragments, the model and the score, plus
# the entry's GO terms, its member databases and whether the
# protein is in AlphaFold.
protein_rows = [
pr for pr in (r.get("proteins") or []) if isinstance(pr, dict)
]
locations = _match_locations(protein_rows)
first_protein = protein_rows[0] if protein_rows else {}
entries.append(
{
"accession": meta.get("accession"),
"name": name,
"type": meta.get("type"),
"source_database": meta.get("source_database"),
"integrated_interpro": meta.get("integrated"),
"member_databases": meta.get("member_databases") or {},
"go_terms": [
{
"identifier": g.get("identifier"),
"name": g.get("name"),
"category": (g.get("category") or {}).get("name"),
}
for g in (meta.get("go_terms") or [])
if isinstance(g, dict)
],
"match_locations": locations,
"match_count": len(locations),
"protein_length": first_protein.get("protein_length"),
"protein_source_database": first_protein.get("source_database"),
"in_alphafold": first_protein.get("in_alphafold"),
}
)
next_cursor = self._next_cursor(data)
total = data.get("count", len(entries))
return {
"status": "success",
"data": {
"protein_accession": accession,
"total_entries": total,
"returned": len(entries),
"page_size": page_size,
"has_more": next_cursor is not None,
"next_cursor": next_cursor,
"truncated": cursor is not None or next_cursor is not None,
"result_set_complete": bool(
cursor is None and next_cursor is None and len(entries) == total
),
"pagination_stalled": bool(next_cursor and not entries),
"entries": entries,
},
"metadata": {
"source": "InterPro API - Entries for Protein",
"accession": accession,
},
}
[docs]
def _search_entries(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search InterPro entries by keyword."""
query = arguments.get("query", "")
if not query:
return {
"status": "error",
"error": "query parameter is required (e.g., 'zinc finger', 'kinase', 'p53')",
}
entry_type = arguments.get("entry_type", None)
valid_entry_types = {
"domain",
"family",
"homologous_superfamily",
"repeat",
"conserved_site",
"active_site",
"binding_site",
"ptm",
}
if entry_type is not None and entry_type not in valid_entry_types:
return {
"status": "error",
"error": "entry_type must be one of: "
+ ", ".join(sorted(valid_entry_types)),
}
controls = self._page_controls(arguments, 20, 50)
if isinstance(controls, dict):
return controls
page_size, cursor = controls
url = f"{INTERPRO_BASE_URL}/entry/interpro"
params = {
"search": query,
"page_size": page_size,
# Without this the listing omits the counts entirely and
# protein_count / structure_count / taxa_count below are always
# null. The API spells the field "counts" (not "counters", which
# it rejects with a 404 listing the valid names) and delivers it
# per result under "extra_fields", not inside "metadata".
"extra_fields": "counts",
}
self._add_cursor(params, arguments)
if entry_type:
params["type"] = entry_type
response = requests.get(
url,
params=params,
headers={"Accept": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
# HTTP 204 / empty body == "no entry matched"; report it as an empty
# result set rather than letting .json() raise.
data = _json_or_none(response) or {}
results = data.get("results", [])
entries = []
for r in results:
meta = r.get("metadata", {})
# metadata.name is a plain string here; keep the dict guard for safety.
name_obj = meta.get("name", "")
name = (
name_obj.get("name", "")
if isinstance(name_obj, dict)
else (str(name_obj) if name_obj else "")
)
counters = (
(r.get("extra_fields") or {}).get("counts")
or meta.get("counters")
or {}
)
entries.append(
{
"accession": meta.get("accession"),
"name": name,
"type": meta.get("type"),
"protein_count": counters.get("proteins"),
"structure_count": counters.get("structures"),
"taxa_count": counters.get("taxa"),
}
)
next_cursor = self._next_cursor(data)
total = data.get("count", len(entries))
return {
"status": "success",
"data": {
"query": query,
"total_results": total,
"returned": len(entries),
"page_size": page_size,
"has_more": next_cursor is not None,
"next_cursor": next_cursor,
"truncated": cursor is not None or next_cursor is not None,
"result_set_complete": bool(
cursor is None and next_cursor is None and len(entries) == total
),
"pagination_stalled": bool(next_cursor and not entries),
"entries": entries,
},
"metadata": {
"source": "InterPro API - Entry Search",
"query": query,
},
}
