# interpro_domain_arch_tool.py
"""
InterPro Domain Architecture tool for ToolUniverse.
Provides access to InterPro API endpoints for domain architecture analysis:
- Protein domain architecture with exact residue positions
- PDB structures containing a specific Pfam domain
- Pfam clan (superfamily) member families
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".
Callers translate a ``None`` return into their own empty-result shape (for
list endpoints) or a not-found error (for single-record endpoints).
"""
if response.status_code == 204:
return None
body = response.content
if not body or not body.strip():
return None
return response.json()
def _domain_locations(entry: dict, key: str) -> list:
"""Where a domain sits, in one of the two coordinate frames InterPro gives.
`entry_protein_locations` is in UniProt numbering and
`entry_structure_locations` in the structure's, with `auth_start` and
`auth_end` carrying the PDB author numbering beside it. For PF00870 in
1gzh those are 100-288 and 6-194 respectively, and the author numbers are
the ones a paper or a mutation name uses -- the same distinction round
thirty's PDBe-KB fix turned on.
`score` is the match e-value, 6.8e-60 here, which is how significant the
domain call is.
"""
out = []
for location in entry.get(key) or []:
if not isinstance(location, dict):
continue
out.append(
{
"fragments": [
{
"start": fragment.get("start"),
"end": fragment.get("end"),
"auth_start": fragment.get("auth_start"),
"auth_end": fragment.get("auth_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 out
[docs]
class InterProDomainArchTool(BaseTool):
"""
Tool for InterPro domain architecture analysis via the InterPro API.
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", "protein_domain_architecture")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the InterPro 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:
param = arguments.get(
"accession",
arguments.get(
"pfam_accession", arguments.get("clan_accession", "")
),
)
return {"status": "error", "error": f"Not found in InterPro: {param}"}
# NB: 204 never reaches here -- raise_for_status() only raises for
# 4xx/5xx. It is handled as an empty result set by _json_or_none().
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 == "protein_domain_architecture":
return self._get_protein_domain_architecture(arguments)
elif self.endpoint == "structures_for_domain":
return self._get_structures_for_domain(arguments)
elif self.endpoint == "clan_members":
return self._get_clan_members(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]
def _get_protein_domain_architecture(
self, arguments: Dict[str, Any]
) -> Dict[str, Any]:
"""Get Pfam domain architecture for a protein with exact positions."""
accession = arguments.get("accession", "")
if not accession:
return {
"status": "error",
"error": "accession parameter is required (UniProt accession, e.g., 'P04637')",
}
max_results = max(1, min(int(arguments.get("max_results") or 50), 200))
url = f"{INTERPRO_BASE_URL}/entry/pfam/protein/uniprot/{accession}"
params = {"page_size": max_results, "format": "json"}
self._add_cursor(params, arguments)
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
# HTTP 204 / empty body == "this protein has no Pfam domains".
data = _json_or_none(response) or {}
results = data.get("results", [])
next_cursor = self._next_cursor(data)
domains = []
protein_length = None
protein_source_database = None
protein_organism = None
for r in results:
metadata = r.get("metadata", {})
name_info = metadata.get("name", {})
name = (
name_info.get("name", "")
if isinstance(name_info, dict)
else str(name_info)
)
proteins = r.get("proteins", [])
for p in proteins:
if protein_length is None:
protein_length = p.get("protein_length")
# Reviewed (Swiss-Prot) against unreviewed (TrEMBL), and the
# organism, for the protein the architecture is OF. Both
# sibling endpoints of this family already report the first.
# `proteins[].accession` and `metadata.source_database` are
# NOT added: they are the queried accession echoed back and
# the constant "pfam" from the request path.
if protein_source_database is None:
protein_source_database = p.get("source_database")
if protein_organism is None:
protein_organism = p.get("organism")
for loc in p.get("entry_protein_locations", []):
for frag in loc.get("fragments", []):
domains.append(
{
"pfam_accession": metadata.get("accession", ""),
"name": name,
"type": metadata.get("type", ""),
"integrated_interpro": metadata.get("integrated"),
"start": frag.get("start"),
"end": frag.get("end"),
"score": loc.get("score"),
}
)
# Sort domains by start position
domains.sort(key=lambda d: d.get("start", 0))
return {
"status": "success",
"data": {
"accession": accession,
"protein_length": protein_length,
"protein_source_database": protein_source_database,
"protein_organism": protein_organism,
"total_entries": data.get("count", len(results)),
"returned_entries": len(results),
"has_more": next_cursor is not None,
"next_cursor": next_cursor,
"domain_count": len(domains),
"domains": domains,
},
"metadata": {
"source": "InterPro API (Pfam domain architecture)",
"accession": accession,
},
}
[docs]
def _get_structures_for_domain(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Find PDB structures containing a specific Pfam domain."""
pfam_acc = arguments.get("pfam_accession", "")
if not pfam_acc:
return {
"status": "error",
"error": "pfam_accession parameter is required (e.g., 'PF00870')",
}
max_results = max(1, min(int(arguments.get("max_results") or 20), 200))
url = f"{INTERPRO_BASE_URL}/structure/pdb/entry/pfam/{pfam_acc}"
params = {"page_size": max_results, "format": "json"}
self._add_cursor(params, arguments)
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
# HTTP 204 / empty body == "no PDB structure contains this domain".
data = _json_or_none(response) or {}
total = data.get("count", 0)
results = data.get("results", [])
next_cursor = self._next_cursor(data)
structures = []
for r in results:
m = r.get("metadata", {})
# Each hit's `entries` block says WHERE in the structure the
# domain is -- which chain, which residues in both the UniProt
# and the PDB author frames -- and how significant the call is.
# The row answered which structures contain the domain and
# nothing about the containment, the same gap this sweep closed
# for the protein and the domain endpoints of this family.
entry_rows = [e for e in (r.get("entries") or []) if isinstance(e, dict)]
matched = next(
(
e
for e in entry_rows
if str(e.get("accession") or "").upper() == pfam_acc.upper()
),
entry_rows[0] if entry_rows else {},
)
structures.append(
{
"pdb_id": m.get("accession", ""),
"name": m.get("name"),
"experiment_type": m.get("experiment_type"),
"resolution": m.get("resolution"),
"chains": sorted(
{e.get("chain") for e in entry_rows if e.get("chain")}
),
"entry_type": matched.get("entry_type"),
"entry_source_database": matched.get("source_database"),
"entry_integrated_interpro": matched.get("entry_integrated"),
"protein_length": matched.get("protein_length"),
"protein_locations": _domain_locations(
matched, "entry_protein_locations"
),
"structure_locations": _domain_locations(
matched, "entry_structure_locations"
),
"entries_matched": sorted(
{e.get("accession") for e in entry_rows if e.get("accession")}
),
}
)
return {
"status": "success",
"data": {
"pfam_accession": pfam_acc,
"total_structures": total,
"returned": len(structures),
"has_more": next_cursor is not None,
"next_cursor": next_cursor,
"structures": structures,
},
"metadata": {
"source": "InterPro API (structures for domain)",
"pfam_accession": pfam_acc,
},
}
[docs]
def _get_clan_members(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get member families in a Pfam clan (superfamily)."""
clan_acc = arguments.get("clan_accession", "")
if not clan_acc:
return {
"status": "error",
"error": "clan_accession parameter is required (e.g., 'CL0016')",
}
max_results = max(1, min(int(arguments.get("max_results") or 50), 200))
member_offset = max(int(arguments.get("member_offset") or 0), 0)
# First get clan metadata
url = f"{INTERPRO_BASE_URL}/set/pfam/{clan_acc}"
params = {"format": "json"}
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
# A single-record endpoint answers an unknown clan with HTTP 204, not
# 404. Report it as not-found rather than a success full of blanks.
clan_data = _json_or_none(response)
if clan_data is None:
return {"status": "error", "error": f"Not found in InterPro: {clan_acc}"}
metadata = clan_data.get("metadata", {})
name_info = metadata.get("name", {})
clan_name = (
name_info.get("name", "") if isinstance(name_info, dict) else str(name_info)
)
# Get member families from relationships
relationships = metadata.get("relationships", {})
nodes = relationships.get("nodes", [])
members = []
for node in nodes:
members.append(
{
"accession": node.get("accession", ""),
"short_name": node.get("short_name"),
"name": node.get("name"),
"type": node.get("type"),
"score": node.get("score"),
}
)
# Sort by accession
members.sort(key=lambda m: m.get("accession", ""))
member_total = len(members)
member_page = members[member_offset : member_offset + max_results]
next_member_offset = member_offset + len(member_page)
if next_member_offset >= member_total:
next_member_offset = None
return {
"status": "success",
"data": {
"clan_accession": clan_acc,
"clan_name": clan_name,
"description": metadata.get("description"),
"member_count": member_total,
"returned": len(member_page),
"members_truncated": next_member_offset is not None,
"next_member_offset": next_member_offset,
"members": member_page,
},
"metadata": {
"source": "InterPro API (Pfam clan members)",
"clan_accession": clan_acc,
},
}
