Source code for tooluniverse.pfam_tool
# pfam_tool.py
"""
Pfam protein families tool for ToolUniverse.
Provides access to Pfam data via the InterPro API (Pfam is now hosted at InterPro):
- Search Pfam families by keyword
- Get detailed Pfam family information (description, counters, clan membership)
- Get proteins containing a Pfam domain (with optional species filter)
- Get Pfam annotations for a specific protein
- List Pfam clans (superfamilies) with search
- Get proteome distribution for a Pfam family
API: https://www.ebi.ac.uk/interpro/api/
No authentication required. Free public access.
"""
import re
import requests
from typing import Dict, Any, Optional
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 Pfam API" message -- callers could not tell
"no such family" from "the tool is broken".
Callers translate a ``None`` return into their own empty-result shape (for
search/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()
[docs]
class PfamTool(BaseTool):
"""
Tool for Pfam protein family queries 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", 90)
fields = tool_config.get("fields", {})
self.endpoint = fields.get("endpoint", "search_families")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the Pfam API call."""
try:
return self._query(arguments)
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"InterPro/Pfam API timed out after {self.timeout}s",
}
except requests.exceptions.ConnectionError:
return {
"status": "error",
"error": "Failed to connect to InterPro/Pfam 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("query", "")),
)
return {
"status": "error",
"error": f"Not found in Pfam/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/Pfam API HTTP error: {code}"}
except Exception as e:
return {
"status": "error",
"error": f"Unexpected error querying Pfam API: {str(e)}",
}
[docs]
def _query(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Route to appropriate endpoint."""
if self.endpoint == "search_families":
return self._search_families(arguments)
elif self.endpoint == "get_family_detail":
return self._get_family_detail(arguments)
elif self.endpoint == "get_family_proteins":
return self._get_family_proteins(arguments)
elif self.endpoint == "get_protein_pfam":
return self._get_protein_pfam(arguments)
elif self.endpoint == "search_clans":
return self._search_clans(arguments)
elif self.endpoint == "get_family_proteomes":
return self._get_family_proteomes(arguments)
else:
return {"status": "error", "error": f"Unknown endpoint: {self.endpoint}"}
[docs]
def _strip_html(self, text: str) -> str:
"""Remove HTML tags from text."""
if not text:
return ""
return re.sub(r"<[^>]+>", "", text).strip()
[docs]
@staticmethod
def _next_cursor(data: Dict[str, Any]) -> Optional[str]:
"""Extract InterPro's opaque cursor from its next-page URL."""
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:
"""Forward a caller-provided InterPro cursor without interpreting it."""
cursor = arguments.get("cursor")
if cursor:
params["cursor"] = cursor
[docs]
def _search_families(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search Pfam families by keyword."""
query = arguments.get("query", "")
if not query:
return {
"status": "error",
"error": "query parameter is required (e.g., 'kinase', 'zinc finger')",
}
max_results = max(1, min(int(arguments.get("max_results") or 20), 100))
url = f"{INTERPRO_BASE_URL}/entry/pfam/"
params = {"search": query, "page_size": max_results}
self._add_cursor(params, arguments)
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
# HTTP 204 / empty body == "no family matched"; report it as an empty
# result set rather than letting .json() raise.
data = _json_or_none(response) or {}
total = data.get("count", 0)
results = data.get("results", [])
next_cursor = self._next_cursor(data)
families = []
for r in results:
meta = r.get("metadata", {})
families.append(
{
"accession": meta.get("accession", ""),
"name": meta.get("name", ""),
"type": meta.get("type", ""),
"integrated_interpro": meta.get("integrated"),
# Which member database the entry comes from. A Pfam
# search answers Pfam entries, and the field says so for
# every row -- which matters because the same endpoint
# serves InterPro's other member databases.
"source_database": meta.get("source_database"),
"member_databases": meta.get("member_databases"),
"go_terms": meta.get("go_terms") or [],
"description": meta.get("description"),
}
)
return {
"status": "success",
"data": {
"query": query,
"total_results": total,
"returned": len(families),
"has_more": next_cursor is not None,
"next_cursor": next_cursor,
"families": families,
},
"metadata": {
"source": "InterPro API (Pfam family search)",
"query": query,
},
}
[docs]
def _get_family_detail(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get detailed information about a specific Pfam family."""
pfam_acc = arguments.get("pfam_accession", "")
if not pfam_acc:
return {
"status": "error",
"error": "pfam_accession parameter is required (e.g., 'PF00001')",
}
url = f"{INTERPRO_BASE_URL}/entry/pfam/{pfam_acc}"
response = requests.get(url, timeout=self.timeout)
response.raise_for_status()
# A single-record endpoint answers an unknown accession with HTTP 204,
# not 404. Report it as not-found rather than a success full of blanks.
data = _json_or_none(response)
if data is None:
return {
"status": "error",
"error": f"Not found in Pfam/InterPro: {pfam_acc}",
}
meta = data.get("metadata", {})
name_info = meta.get("name", {})
if isinstance(name_info, dict):
full_name = name_info.get("name", "")
short_name = name_info.get("short", "")
else:
full_name = str(name_info)
short_name = ""
# Extract description text
desc_list = meta.get("description", [])
description = ""
if desc_list and isinstance(desc_list, list):
description = self._strip_html(desc_list[0].get("text", ""))
# Extract counters
counters = meta.get("counters", {})
# Extract set/clan info
set_info = meta.get("set_info", {})
clan_accession = set_info.get("accession") if set_info else None
clan_name = set_info.get("name") if set_info else None
# Extract representative structure
rep_struct = meta.get("representative_structure", {})
# Extract Wikipedia
wiki = meta.get("wikipedia", [])
wikipedia_title = wiki[0].get("title", "") if wiki else None
# Extract literature count
lit = meta.get("literature", {})
literature_count = len(lit) if lit else 0
# Extract GO terms
go_terms = meta.get("go_terms", []) or []
return {
"status": "success",
"data": {
"accession": meta.get("accession", ""),
"name": full_name,
"short_name": short_name,
"type": meta.get("type", ""),
"source_database": meta.get("source_database", ""),
"integrated_interpro": meta.get("integrated"),
"description": description[:2000] if description else None,
"clan_accession": clan_accession,
"clan_name": clan_name,
"counters": {
"proteins": counters.get("proteins", 0),
"structures": counters.get("structures", 0),
"taxa": counters.get("taxa", 0),
"proteomes": counters.get("proteomes", 0),
"domain_architectures": counters.get("domain_architectures", 0),
"matches": counters.get("matches", 0),
},
"representative_structure": {
"pdb_id": rep_struct.get("accession"),
"name": rep_struct.get("name"),
}
if rep_struct
else None,
"wikipedia_title": wikipedia_title,
"literature_count": literature_count,
"go_terms": go_terms[:20] if go_terms else [],
},
"metadata": {
"source": "InterPro API (Pfam family detail)",
"pfam_accession": pfam_acc,
},
}
[docs]
def _get_family_proteins(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get proteins containing a specific Pfam domain, optionally filtered by species."""
pfam_acc = arguments.get("pfam_accession", "")
if not pfam_acc:
return {
"status": "error",
"error": "pfam_accession parameter is required (e.g., 'PF00001')",
}
max_results = max(1, min(int(arguments.get("max_results") or 20), 100))
reviewed_only = arguments.get("reviewed_only", True)
tax_id = arguments.get("tax_id", None)
db = "reviewed" if reviewed_only else "uniprot"
if tax_id:
url = f"{INTERPRO_BASE_URL}/protein/{db}/entry/pfam/{pfam_acc}/taxonomy/uniprot/{tax_id}/"
else:
url = f"{INTERPRO_BASE_URL}/protein/{db}/entry/pfam/{pfam_acc}/"
params = {"page_size": max_results}
self._add_cursor(params, arguments)
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
# HTTP 204 / empty body == "no protein matched this family (+ filter)".
data = _json_or_none(response) or {}
total = data.get("count", 0)
results = data.get("results", [])
next_cursor = self._next_cursor(data)
proteins = []
for r in results:
meta = r.get("metadata", {})
organism = meta.get("source_organism", {})
entries = r.get("entries", [])
# Get domain positions
domain_positions = []
for entry in entries:
if entry.get("accession", "").upper() == pfam_acc.upper():
for loc in entry.get("entry_protein_locations", []):
for frag in loc.get("fragments", []):
domain_positions.append(
{
"start": frag.get("start"),
"end": frag.get("end"),
# The match score and the model it was
# matched against. A domain position
# without them is a claim with no
# strength attached.
"score": loc.get("score"),
"model_accession": loc.get("model")
or loc.get("model_acc"),
"dc_status": frag.get("dc-status"),
}
)
proteins.append(
{
"accession": meta.get("accession", ""),
"name": meta.get("name", ""),
"gene": meta.get("gene"),
"length": meta.get("length"),
"organism": organism.get("scientificName") if organism else None,
"tax_id": organism.get("taxId") if organism else None,
"domain_positions": domain_positions,
"in_alphafold": meta.get("in_alphafold", False),
# Reviewed (Swiss-Prot) or not, which is the quality
# distinction the `reviewed_only` filter is about and
# which the rows never stated; plus the organism's full
# name and the entry types matched.
"source_database": meta.get("source_database"),
"organism_full_name": organism.get("fullName")
if organism
else None,
"entry_types": sorted(
{
entry.get("entry_type")
for entry in entries
if isinstance(entry, dict) and entry.get("entry_type")
}
),
"entry_source_databases": sorted(
{
entry.get("source_database")
for entry in entries
if isinstance(entry, dict)
and entry.get("source_database")
}
),
"entries_integrated": sorted(
{
entry.get("entry_integrated")
for entry in entries
if isinstance(entry, dict)
and entry.get("entry_integrated")
}
),
# The taxonomy rows the same response carries for the
# organism, where the protein row gave a tax id.
"taxa": [
{
"accession": taxon.get("accession"),
"source_database": taxon.get("source_database"),
}
for taxon in (r.get("taxa") or [])
if isinstance(taxon, dict)
],
}
)
return {
"status": "success",
"data": {
"pfam_accession": pfam_acc,
"total_proteins": total,
"returned": len(proteins),
"has_more": next_cursor is not None,
"next_cursor": next_cursor,
"reviewed_only": reviewed_only,
"tax_id_filter": tax_id,
"proteins": proteins,
},
"metadata": {
"source": "InterPro API (Pfam family proteins)",
"pfam_accession": pfam_acc,
},
}
[docs]
def _get_protein_pfam(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get all Pfam domain annotations for a specific 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/pfam/protein/uniprot/{accession}"
params = {"page_size": 50}
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
# HTTP 204 / empty body == "this protein has no Pfam annotations".
data = _json_or_none(response) or {}
results = data.get("results", [])
domains = []
protein_length = None
for r in results:
meta = r.get("metadata", {})
name_info = meta.get("name", {})
if isinstance(name_info, dict):
name = name_info.get("name", "")
short_name = name_info.get("short", "")
else:
name = str(name_info) if name_info else ""
short_name = ""
proteins = r.get("proteins", [])
for p in proteins:
if protein_length is None:
protein_length = p.get("protein_length")
for loc in p.get("entry_protein_locations", []):
for frag in loc.get("fragments", []):
domains.append(
{
"pfam_accession": meta.get("accession", ""),
"name": name,
"short_name": short_name,
"type": meta.get("type", ""),
"integrated_interpro": meta.get("integrated"),
"start": frag.get("start"),
"end": frag.get("end"),
"score": loc.get("score"),
# Which protein the domain is on, and from
# which database. The row named the Pfam
# family and the coordinates and left the
# protein unnamed, so a multi-protein answer
# could not be split.
"protein_accession": p.get("accession"),
"protein_source_database": p.get("source_database"),
"protein_organism": p.get("organism"),
"protein_length": p.get("protein_length"),
"protein_name": p.get("name"),
"model_accession": loc.get("model")
or loc.get("model_acc"),
"fragment_dc_status": frag.get("dc-status"),
# Which member database the entry is from and
# what kind of entry it is: the same endpoint
# serves InterPro's other member databases,
# and "domain" against "family" changes what
# the match means.
"source_database": meta.get("source_database"),
"entry_type": meta.get("type"),
"go_terms": meta.get("go_terms") or [],
}
)
# Sort domains by start position
domains.sort(key=lambda d: d.get("start", 0))
return {
"status": "success",
"data": {
"accession": accession,
"protein_length": protein_length,
"domain_count": len(domains),
"domains": domains,
},
"metadata": {
"source": "InterPro API (Pfam annotations for protein)",
"accession": accession,
},
}
[docs]
def _search_clans(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search Pfam clans (superfamilies)."""
query = arguments.get("query", "")
max_results = max(1, min(int(arguments.get("max_results") or 20), 100))
url = f"{INTERPRO_BASE_URL}/set/pfam/"
params = {"page_size": max_results}
self._add_cursor(params, arguments)
if query:
params["search"] = query
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
# HTTP 204 / empty body == "no clan matched".
data = _json_or_none(response) or {}
total = data.get("count", 0)
results = data.get("results", [])
next_cursor = self._next_cursor(data)
clans = []
for r in results:
meta = r.get("metadata", {})
clans.append(
{
"accession": meta.get("accession", ""),
"name": meta.get("name", ""),
"source_database": meta.get("source_database", ""),
}
)
return {
"status": "success",
"data": {
"query": query if query else "(all clans)",
"total_results": total,
"returned": len(clans),
"has_more": next_cursor is not None,
"next_cursor": next_cursor,
"clans": clans,
},
"metadata": {
"source": "InterPro API (Pfam clan search)",
"query": query,
},
}
[docs]
def _get_family_proteomes(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get proteome distribution for a Pfam family."""
pfam_acc = arguments.get("pfam_accession", "")
if not pfam_acc:
return {
"status": "error",
"error": "pfam_accession parameter is required (e.g., 'PF00001')",
}
max_results = max(1, min(int(arguments.get("max_results") or 20), 100))
url = f"{INTERPRO_BASE_URL}/proteome/uniprot/entry/pfam/{pfam_acc}/"
params = {"page_size": max_results}
self._add_cursor(params, arguments)
response = requests.get(url, params=params, timeout=self.timeout)
response.raise_for_status()
# HTTP 204 / empty body == "no proteome carries this family".
data = _json_or_none(response) or {}
total = data.get("count", 0)
results = data.get("results", [])
next_cursor = self._next_cursor(data)
proteomes = []
for r in results:
meta = r.get("metadata", {})
proteomes.append(
{
"proteome_accession": meta.get("accession", ""),
"organism_name": meta.get("name", ""),
"taxonomy_id": meta.get("taxonomy"),
"is_reference": meta.get("is_reference", False),
# Where the proteome sits in the tree, and how the family
# matches in it: a proteome row that cannot say which
# proteins matched, how long they are or how well they
# scored is a name and a tax id.
"lineage": meta.get("lineage"),
"source_database": meta.get("source_database"),
"strain": meta.get("strain"),
"assembly": meta.get("assembly"),
"proteins": [
{
"accession": entry.get("accession"),
"protein_length": entry.get("protein_length"),
"source_database": entry.get("source_database"),
"entry_type": entry.get("entry_type"),
# The InterPro entry this Pfam family is
# integrated into, which is how a Pfam match is
# carried into InterPro's own classification.
"entry_integrated": entry.get("entry_integrated"),
"locations": [
{
"start": fragment.get("start"),
"end": fragment.get("end"),
"score": location.get("score"),
}
for location in (
entry.get("entry_protein_locations") or []
)
for fragment in (location.get("fragments") or [])
],
}
for entry in (r.get("entries") or [])
if isinstance(entry, dict)
],
}
)
return {
"status": "success",
"data": {
"pfam_accession": pfam_acc,
"total_proteomes": total,
"returned": len(proteomes),
"has_more": next_cursor is not None,
"next_cursor": next_cursor,
"proteomes": proteomes,
},
"metadata": {
"source": "InterPro API (Pfam family proteomes)",
"pfam_accession": pfam_acc,
},
}