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"
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 = int(arguments.get("limit", 50))
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
)
truncated = unique_interactions[:limit]
return {
"status": "success",
"data": {
"query_accession": accession,
"interactions": truncated,
},
"metadata": {
"source": "EBI Proteins API / IntAct (ebi.ac.uk/proteins)",
"total_interactions": len(unique_interactions),
"returned": len(truncated),
},
}
[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').",
}
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),
}
)
# 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))
# 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
top_interactions = sorted(
seen.values(), key=lambda x: x["experiments"], reverse=True
)[:50]
return {
"status": "success",
"data": {
"query_accession": accession,
"protein_name": protein_name,
"protein_existence": protein_existence,
"organism": organism,
"total_interaction_entries": len(data),
"top_interactions": top_interactions,
"diseases": sorted(diseases),
"subcellular_locations": sorted(locations),
},
"metadata": {
"source": "EBI Proteins API / IntAct (ebi.ac.uk/proteins)",
"total_interactions": len(seen),
},
}