Source code for tooluniverse.uniref_tool
# uniref_tool.py
"""
UniProt UniRef Clusters API tool for ToolUniverse.
UniRef provides clustered sets of protein sequences at different levels
of sequence identity: UniRef100, UniRef90, and UniRef50. These clusters
group related sequences to reduce redundancy and improve search speed.
API: https://rest.uniprot.org/uniref/
No authentication required. Free public access.
"""
import requests
from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool
from .uniprot_pagination import pagination_metadata
UNIREF_BASE_URL = "https://rest.uniprot.org/uniref"
MEMBERS_PER_CLUSTER = 200
def _go_terms(cluster: dict) -> list:
"""The GO annotations UniRef assigns to the cluster.
Both endpoints answer `goTerms` and neither handler read it: a cluster's
GO terms are the functional summary of the group, and no UniRef tool
returned them anywhere. UniRef90_A5H1J3 carries GO:0003677 (GO Molecular
Function); clusters with no annotation answer null, which is a real
answer and not an omission.
"""
rows = []
for term in cluster.get("goTerms") or []:
if not isinstance(term, dict):
continue
rows.append({"go_id": term.get("goId"), "aspect": term.get("aspect")})
return rows
def _member(member: dict) -> dict:
"""One member of a UniRef cluster, with the ids that reach it.
`accessions` is what joins a member to UniProtKB; `uniref50Id`,
`uniref100Id` and `uniparcId` are the same sequence at the other
thresholds. UniRef sends all of them per member and the cluster is its
membership.
"""
return {
"member_id": member.get("memberId"),
"member_id_type": member.get("memberIdType"),
"accessions": member.get("accessions") or [],
"organism_name": member.get("organismName"),
"organism_tax_id": member.get("organismTaxId"),
"protein_name": member.get("proteinName"),
"sequence_length": member.get("sequenceLength"),
"uniref50_id": member.get("uniref50Id"),
"uniref100_id": member.get("uniref100Id"),
"uniparc_id": member.get("uniparcId"),
}
[docs]
@register_tool("UniRefTool")
class UniRefTool(BaseTool):
"""
Tool for querying UniProt UniRef protein sequence clusters.
Supports:
- Get cluster details by ID (UniRef90_XXXXX, UniRef50_XXXXX, UniRef100_XXXXX)
- Search clusters by protein name, gene, or organism
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_cluster")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the UniRef API call."""
try:
return self._query(arguments)
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"UniRef API timed out after {self.timeout}s",
}
except requests.exceptions.ConnectionError:
return {"status": "error", "error": "Failed to connect to UniRef API"}
except requests.exceptions.HTTPError as e:
return {
"status": "error",
"error": f"UniRef API HTTP error: {e.response.status_code}",
}
except Exception as e:
return {"status": "error", "error": f"Unexpected error: {str(e)}"}
[docs]
def _query(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Route to appropriate endpoint."""
if self.endpoint == "get_cluster":
return self._get_cluster(arguments)
elif self.endpoint == "search_clusters":
return self._search_clusters(arguments)
else:
return {"status": "error", "error": f"Unknown endpoint: {self.endpoint}"}
[docs]
def _get_cluster(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get UniRef cluster details by cluster ID."""
cluster_id = arguments.get("cluster_id", "")
if not cluster_id:
return {
"status": "error",
"error": "cluster_id parameter is required (e.g., 'UniRef90_P04637')",
}
cluster_id = cluster_id.strip()
url = f"{UNIREF_BASE_URL}/{cluster_id}"
response = requests.get(
url, headers={"Accept": "application/json"}, timeout=self.timeout
)
response.raise_for_status()
data = response.json()
# Parse representative member
rep = data.get("representativeMember") or {}
representative = {
"member_id": rep.get("memberId"),
"member_id_type": rep.get("memberIdType"),
"organism_name": rep.get("organismName"),
"organism_tax_id": rep.get("organismTaxId"),
"sequence_length": rep.get("sequenceLength"),
"protein_name": rep.get("proteinName"),
"accessions": rep.get("accessions", []),
}
# Sequence info
seq_obj = rep.get("sequence") or {}
sequence = seq_obj.get("value")
sequence_length = seq_obj.get("length")
# Common taxon
common_taxon = data.get("commonTaxon") or {}
raw_members = data.get("members") or []
members = [
_member(m)
for m in raw_members[:MEMBERS_PER_CLUSTER]
if isinstance(m, dict)
]
return {
"status": "success",
"data": {
"cluster_id": data.get("id"),
"name": data.get("name"),
"entry_type": data.get("entryType"),
"member_count": data.get("memberCount"),
"updated": data.get("updated"),
"seed_id": data.get("seedId"),
"common_taxon": {
"scientific_name": common_taxon.get("scientificName"),
"taxon_id": common_taxon.get("taxonId"),
},
"go_terms": _go_terms(data),
"representative_member": representative,
# The cluster's other members. member_count said 254 for
# UniRef90_P05067 and only the representative was returned,
# so the 253 sequences the cluster groups -- the thing a
# cluster is -- had no path to the caller.
"members": members,
"members_returned": len(members),
"members_in_response": len(raw_members),
"members_truncated": len(raw_members) > len(members),
"sequence": sequence,
"sequence_length": sequence_length
or representative.get("sequence_length"),
},
"metadata": {
"source": "UniProt UniRef",
"cluster_id": cluster_id,
},
}
[docs]
def _search_clusters(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search UniRef clusters by protein name, gene, or organism."""
query = arguments.get("query", "")
if not query:
return {
"status": "error",
"error": "query parameter is required (e.g., 'p53', 'insulin', 'kinase Homo sapiens')",
}
cluster_type = arguments.get("cluster_type") or "UniRef90"
identity_map = {
"UniRef100": "1.0",
"UniRef90": "0.9",
"UniRef50": "0.5",
}
if cluster_type not in identity_map:
return {
"status": "error",
"error": (
"cluster_type must be one of UniRef100, UniRef90, or UniRef50; "
"invalid values are not sent as an unfiltered search"
),
}
raw_size = arguments.get("size")
raw_size = 10 if raw_size is None else raw_size
try:
if isinstance(raw_size, bool):
raise ValueError
size = int(raw_size)
if isinstance(raw_size, float) and not raw_size.is_integer():
raise ValueError
if not 1 <= size <= 500:
raise ValueError
except (TypeError, ValueError):
return {
"status": "error",
"error": "size must be an integer from 1 to 500",
}
# Build the search query with identity filter
# UniRef API uses decimal identity: 1.0 (UniRef100), 0.9 (UniRef90), 0.5 (UniRef50)
identity_val = identity_map[cluster_type]
search_query = f"{query} AND identity:{identity_val}"
url = f"{UNIREF_BASE_URL}/search"
params = {
"query": search_query,
"size": size,
}
if arguments.get("cursor"):
params["cursor"] = arguments["cursor"]
response = requests.get(
url,
params=params,
headers={"Accept": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
data = response.json()
results_raw = data.get("results", [])
results = []
for cluster in results_raw:
rep = cluster.get("representativeMember") or {}
common_taxon = cluster.get("commonTaxon") or {}
results.append(
{
"cluster_id": cluster.get("id"),
"name": cluster.get("name"),
"entry_type": cluster.get("entryType"),
"member_count": cluster.get("memberCount"),
"updated": cluster.get("updated"),
"representative_member_id": rep.get("memberId"),
"representative_organism": rep.get("organismName"),
"representative_protein": rep.get("proteinName"),
"representative_sequence_length": rep.get("sequenceLength"),
"common_taxon": common_taxon.get("scientificName"),
# The name alone cannot be joined; the reply carries the
# taxon id beside it (Amniota is 32524).
"common_taxon_id": common_taxon.get("taxonId"),
# Available on this endpoint and on the lookup, read by
# neither until now.
"go_terms": _go_terms(cluster),
# member_count is the cluster's size; the members
# themselves come from UniRef_get_cluster, which returns
# them, so they are not duplicated into every search row.
"members_from": "UniRef_get_cluster",
}
)
return {
"status": "success",
"data": results,
"metadata": {
"source": "UniProt UniRef",
"query": query,
"cluster_type": cluster_type,
**pagination_metadata(response, len(results)),
},
}