Source code for tooluniverse.ebi_proteins_interactions_tool
# ebi_proteins_interactions_tool.py
"""
EBI Proteins Interactions tool for ToolUniverse.
Provides protein-protein interaction data from the EBI Proteins API,
sourced from IntAct. Returns experimentally validated binary interactions
with partner details and experiment counts.
API: https://www.ebi.ac.uk/proteins/api/proteins/interaction/
No authentication required.
"""
import requests
from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool
EBI_PROTEINS_BASE_URL = "https://www.ebi.ac.uk/proteins/api"
# TP53's entry carries 9 disease annotations and 6 subcellular locations, each
# with its own evidence list. The evidence is capped per record.
EVIDENCE_PER_ANNOTATION = 10
def _evidences(node: Any) -> list:
"""Who says so: the ECO code and the publication behind an annotation."""
rows = node.get("evidences") if isinstance(node, dict) else None
if isinstance(rows, dict):
rows = [rows]
if not isinstance(rows, list):
return []
out = []
for row in rows[:EVIDENCE_PER_ANNOTATION]:
if not isinstance(row, dict):
continue
source = row.get("source") or {}
out.append(
{
"code": row.get("code"),
"source": source.get("name"),
"id": source.get("id"),
"url": source.get("url"),
}
)
return out
def _disease_rows(entry: Dict[str, Any]) -> list:
"""The disease annotations, not just the names of the named ones.
`diseases` was built by taking `diseaseId or acronym` and discarding
everything else, which dropped three things. The OMIM id sits in
`dbReference`; the definition sits in `description`; and one of TP53's
nine records has no diseaseId at all -- it is the free-text disease
involvement paragraph, which was therefore dropped entirely.
"""
rows = entry.get("diseases")
if isinstance(rows, dict):
rows = [rows]
if not isinstance(rows, list):
return []
out = []
for row in rows:
if not isinstance(row, dict):
continue
reference = row.get("dbReference") or {}
description = row.get("description") or {}
texts = row.get("text")
if isinstance(texts, dict):
texts = [texts]
out.append(
{
"name": row.get("diseaseId"),
"acronym": row.get("acronym"),
"description": description.get("value")
if isinstance(description, dict)
else description,
"reference_database": reference.get("type"),
"reference_id": reference.get("id"),
"notes": [
item.get("value")
for item in (texts or [])
if isinstance(item, dict) and item.get("value")
],
"evidences": _evidences(description)
or _evidences(row)
or [
evidence
for item in (texts or [])
if isinstance(item, dict)
for evidence in _evidences(item)
],
}
)
return out
def _location_rows(entry: Dict[str, Any]) -> list:
"""Each subcellular location with its topology and its evidence."""
groups = entry.get("subcellularLocations")
if isinstance(groups, dict):
groups = [groups]
if not isinstance(groups, list):
return []
out = []
for group in groups:
if not isinstance(group, dict):
continue
for item in group.get("locations") or [group]:
if not isinstance(item, dict):
continue
location = item.get("location") or {}
topology = item.get("topology") or {}
orientation = item.get("orientation") or {}
out.append(
{
"location": location.get("value")
if isinstance(location, dict)
else location,
"topology": topology.get("value")
if isinstance(topology, dict)
else topology,
"orientation": orientation.get("value")
if isinstance(orientation, dict)
else orientation,
"evidences": _evidences(location)
if isinstance(location, dict)
else [],
}
)
return out
def _entries_for_accession(data: Any, accession: str) -> list:
"""Keep only the payload entries describing `accession` itself.
/proteins/interaction/{acc} answers with the queried protein followed by
its neighbours, so an entry's own `accession` is the only thing that ties
an interaction list to the protein it belongs to.
"""
if not isinstance(data, list):
return []
wanted = accession.upper()
return [
entry
for entry in data
if isinstance(entry, dict) and str(entry.get("accession", "")).upper() == wanted
]
[docs]
@register_tool("EBIProteinsInteractionsTool")
class EBIProteinsInteractionsTool(BaseTool):
"""
Tool for querying EBI Proteins protein-protein interaction data.
Supports:
- Get interaction partners for a protein (from IntAct)
- Get detailed protein info with interactions, diseases, locations
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", "interactions")
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Execute the EBI Proteins Interactions API call."""
try:
return self._query(arguments)
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"EBI Proteins API timed out after {self.timeout}s",
}
except requests.exceptions.ConnectionError:
return {"status": "error", "error": "Failed to connect to EBI Proteins API"}
except requests.exceptions.HTTPError as e:
status = e.response.status_code if e.response is not None else "unknown"
if status == 400:
return {
"status": "error",
"error": f"Invalid accession. Use a UniProt accession (e.g., P04637).",
}
return {"status": "error", "error": f"EBI Proteins API HTTP {status}"}
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 == "interactions":
return self._get_interactions(arguments)
elif self.endpoint == "interaction_details":
return self._get_interaction_details(arguments)
else:
return {"status": "error", "error": f"Unknown endpoint: {self.endpoint}"}
[docs]
def _get_interactions(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get protein-protein interaction partners."""
accession = arguments.get("accession", "")
if not accession:
return {
"status": "error",
"error": "accession is required (e.g., 'P04637').",
}
limit = min(max(int(arguments.get("limit") or 50), 1), 500)
offset = max(int(arguments.get("offset") or 0), 0)
url = f"{EBI_PROTEINS_BASE_URL}/proteins/interaction/{accession}"
response = requests.get(
url,
headers={"Accept": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
data = response.json()
# The endpoint returns the queried protein *plus* ~100 neighbour
# proteins, each carrying its own full interaction list. Pooling them
# attributed other proteins' partners to the query: 44 of the 50 rows
# returned for TP53 were pairs like AXIN1-GSK3B and BRCA2-RAD51, and
# the total read 3258 instead of TP53's own few hundred.
all_interactions = []
for entry in _entries_for_accession(data, accession):
for interaction in entry.get("interactions", []):
partner_acc = interaction.get(
"accession2", interaction.get("accession1")
)
# Skip self-interactions
if partner_acc == accession:
partner_acc = interaction.get("accession1")
all_interactions.append(
{
"partner_accession": partner_acc,
"gene_name": interaction.get("gene"),
"experiments": interaction.get("experiments", 0),
"organism_differ": interaction.get("organismDiffer", False),
"intact_id_a": interaction.get("interactor1"),
"intact_id_b": interaction.get("interactor2"),
}
)
# Deduplicate by partner accession, keep highest experiment count
seen = {}
for interaction in all_interactions:
partner = interaction["partner_accession"]
if (
partner not in seen
or interaction["experiments"] > seen[partner]["experiments"]
):
seen[partner] = interaction
unique_interactions = sorted(
seen.values(), key=lambda x: x["experiments"], reverse=True
)
interaction_page = unique_interactions[offset : offset + limit]
interactions_truncated = offset + len(interaction_page) < len(
unique_interactions
)
return {
"status": "success",
"data": {
"query_accession": accession,
"interactions": interaction_page,
"total_interactions": len(unique_interactions),
"offset": offset,
"interactions_returned": len(interaction_page),
"interactions_truncated": interactions_truncated,
"next_offset": (
offset + len(interaction_page) if interactions_truncated else None
),
},
"metadata": {
"source": "EBI Proteins API / IntAct (ebi.ac.uk/proteins)",
"total_interactions": len(unique_interactions),
"returned": len(interaction_page),
"offset": offset,
},
}
[docs]
def _get_interaction_details(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get protein info with interactions, diseases, and locations."""
accession = arguments.get("accession", "")
if not accession:
return {
"status": "error",
"error": "accession is required (e.g., 'P04637').",
}
limit = min(max(int(arguments.get("limit") or 50), 1), 500)
offset = max(int(arguments.get("offset") or 0), 0)
url = f"{EBI_PROTEINS_BASE_URL}/proteins/interaction/{accession}"
response = requests.get(
url,
headers={"Accept": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
data = response.json()
if not isinstance(data, list) or not data:
return {
"status": "error",
"error": f"No interaction data found for {accession}",
}
# Metadata must come from the queried protein's own entry, not merely
# the first one in the payload.
own_entries = _entries_for_accession(data, accession)
first_entry = own_entries[0] if own_entries else data[0]
# Feature-66B-008a: "accession" returns the accession again; use "name" for protein name
protein_name = first_entry.get("name", accession)
protein_existence = first_entry.get("proteinExistence")
organism = None
taxonomy = first_entry.get("taxonomy")
if taxonomy:
organism = taxonomy if isinstance(taxonomy, str) else str(taxonomy)
# Collect all interactions across entries
all_interactions = []
# Feature-66B-008c: diseases/locations are properties of the query protein only (data[0]),
# not of all interaction partners — iterating all entries mixes partner data in.
diseases = set()
locations = set()
# Same scoping as the interactions tool: neighbour entries describe
# their own partners, not this protein's.
for entry in own_entries:
for interaction in entry.get("interactions", []):
partner_acc = interaction.get(
"accession2", interaction.get("accession1")
)
if partner_acc == accession:
partner_acc = interaction.get("accession1")
all_interactions.append(
{
"partner_accession": partner_acc,
"gene_name": interaction.get("gene"),
"experiments": interaction.get("experiments", 0),
"organism_differ": interaction.get("organismDiffer", False),
# The sibling tool over the same endpoint returns the
# IntAct interactor ids and this one did not, so a
# row here could not be followed into IntAct.
"intact_id_a": interaction.get("interactor1"),
"intact_id_b": interaction.get("interactor2"),
}
)
# Extract diseases and locations from the query protein entry only
for disease in first_entry.get("diseases", []):
# Feature-67B-003A: disease.get("type") returns the UniProt type discriminator
# string "DISEASE" (not a disease name) — omit it from the fallback chain.
disease_name = disease.get("diseaseId") or disease.get("acronym")
if disease_name:
diseases.add(str(disease_name))
for loc in first_entry.get("subcellularLocations", []):
for subloc in loc.get("locations", [loc]):
# Feature-66B-008b: subloc["value"] is wrong; the value is nested under "location"
if isinstance(subloc, dict):
loc_name = subloc.get("location", {}).get("value")
else:
loc_name = str(subloc)
if loc_name:
locations.add(str(loc_name))
# `diseases` and `subcellular_locations` stay the sorted name lists
# they have always been; these carry what the records also held.
disease_records = _disease_rows(first_entry)
location_records = _location_rows(first_entry)
# Deduplicate and sort
seen = {}
for interaction in all_interactions:
partner = interaction["partner_accession"]
if (
partner not in seen
or interaction["experiments"] > seen[partner]["experiments"]
):
seen[partner] = interaction
sorted_interactions = sorted(
seen.values(), key=lambda x: x["experiments"], reverse=True
)
interaction_page = sorted_interactions[offset : offset + limit]
interactions_truncated = offset + len(interaction_page) < len(
sorted_interactions
)
return {
"status": "success",
"data": {
"query_accession": accession,
"protein_name": protein_name,
"protein_existence": protein_existence,
"organism": organism,
"total_interaction_entries": len(data),
"top_interactions": interaction_page,
"total_interactions": len(sorted_interactions),
"offset": offset,
"interactions_returned": len(interaction_page),
"interactions_truncated": interactions_truncated,
"next_offset": (
offset + len(interaction_page) if interactions_truncated else None
),
"diseases": sorted(diseases),
"disease_records": disease_records,
"subcellular_locations": sorted(locations),
"subcellular_location_records": location_records,
},
"metadata": {
"source": "EBI Proteins API / IntAct (ebi.ac.uk/proteins)",
"total_interactions": len(seen),
"returned": len(interaction_page),
"offset": offset,
},
}