Source code for tooluniverse.pathwaycommons_tool
"""
PathwayCommons Tool - Unified Pathway and Interaction Database
Provides access to Pathway Commons REST API (PC2) for searching pathways,
retrieving pathway details, and querying gene interaction neighborhoods
across 22 integrated pathway/interaction databases including Reactome,
KEGG, WikiPathways, PID, BioGRID, IntAct, and more.
API base: https://www.pathwaycommons.org/pc2/
No authentication required.
Reference: Cerami et al., Nucleic Acids Res. 2011; Rodchenkov et al., Nucleic Acids Res. 2020
"""
import requests
import math
from typing import Dict, Any, Optional
from .base_tool import BaseTool
from .tool_registry import register_tool
PC2_BASE_URL = "https://www.pathwaycommons.org/pc2"
SIF_PATTERNS = {
"controls-state-change-of",
"controls-transport-of",
"controls-phosphorylation-of",
"controls-expression-of",
"catalysis-precedes",
"in-complex-with",
"interacts-with",
"neighbor-of",
"consumption-controlled-by",
"controls-production-of",
"controls-transport-of-chemical",
"chemical-affects",
"reacts-with",
"used-to-produce",
}
[docs]
@register_tool("PathwayCommonsTool")
class PathwayCommonsTool(BaseTool):
"""
Tool for querying Pathway Commons (PC2) unified pathway/interaction database.
Supported operations:
- search: Search for pathways, interactions, or molecular entities by gene/keyword
- get_pathway: Get pathway details by URI
- get_neighborhood: Get interaction neighborhood for a gene (SIF format)
"""
[docs]
def __init__(self, tool_config: Dict[str, Any]):
super().__init__(tool_config)
self.parameter = tool_config.get("parameter", {})
self.required = self.parameter.get("required", [])
self.session = requests.Session()
self.timeout = 120
[docs]
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
operation = (
arguments.get("operation")
or self.tool_config.get("fields", {}).get("operation")
or self.get_schema_const_operation()
)
if not operation:
return {"status": "error", "error": "Missing required parameter: operation"}
handlers = {
"search": self._search,
"get_pathway": self._get_pathway,
"get_neighborhood": self._get_neighborhood,
"paths_between": self._paths_between,
}
handler = handlers.get(operation)
if not handler:
return {
"status": "error",
"error": "Unknown operation: {}. Available: {}".format(
operation, list(handlers.keys())
),
}
try:
return handler(arguments)
except requests.exceptions.Timeout:
return {"status": "error", "error": "PathwayCommons API request timed out"}
except requests.exceptions.ConnectionError:
return {
"status": "error",
"error": "Failed to connect to PathwayCommons API",
}
except Exception as e:
return {
"status": "error",
"error": "PathwayCommons error: {}".format(str(e)),
}
[docs]
@staticmethod
def _validated_int(
arguments: Dict[str, Any],
name: str,
default: int,
minimum: int,
maximum: Optional[int] = None,
):
value = arguments.get(name, default)
if isinstance(value, bool) or not isinstance(value, int):
return None, "{} must be an integer".format(name)
if value < minimum:
return None, "{} must be at least {}".format(name, minimum)
if maximum is not None and value > maximum:
return None, "{} must be at most {}".format(name, maximum)
return value, None
[docs]
@staticmethod
def _graph_filters(arguments: Dict[str, Any], allow_direction: bool):
params = {}
for name in ("datasource", "organism"):
value = arguments.get(name)
if value is not None:
if not isinstance(value, str) or not value.strip():
return None, "{} must be a non-empty string".format(name)
params[name] = value.strip()
pattern = arguments.get("pattern")
if pattern is not None:
if (
not isinstance(pattern, list)
or not pattern
or any(
not isinstance(item, str) or not item.strip() for item in pattern
)
):
return None, "pattern must be a non-empty list of interaction types"
normalized = [item.strip().lower().replace("_", "-") for item in pattern]
unknown = sorted(set(normalized) - SIF_PATTERNS)
if unknown:
return None, "Unsupported pattern values: {}".format(", ".join(unknown))
params["pattern"] = normalized
subpw = arguments.get("subpw")
if subpw is not None:
if not isinstance(subpw, bool):
return None, "subpw must be a boolean"
params["subpw"] = subpw
direction = arguments.get("direction")
if direction is not None:
if not allow_direction:
return None, "direction is only supported for neighborhood queries"
if not isinstance(direction, str):
return None, "direction must be a string"
normalized_direction = direction.strip().lower().replace("_", "")
allowed = {"upstream", "downstream", "bothstream", "undirected"}
if normalized_direction not in allowed:
return None, "direction must be one of {}".format(
", ".join(sorted(allowed))
)
params["direction"] = normalized_direction
return params, None
[docs]
def _search(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Search PathwayCommons for pathways/interactions by gene or keyword."""
query = arguments.get("query") or arguments.get("q")
if not query:
return {"status": "error", "error": "Missing required parameter: query"}
params = {"q": query, "format": "json"}
entity_type = arguments.get("type")
if entity_type:
params["type"] = entity_type
datasource = arguments.get("datasource")
if datasource:
params["datasource"] = datasource
organism = arguments.get("organism")
if organism:
params["organism"] = organism
page, error = self._validated_int(arguments, "page", 0, 0)
if error:
return {
"status": "error",
"error": "page must be a non-negative integer (0-based)",
}
params["page"] = page
response = self.session.get(
"{}/search".format(PC2_BASE_URL), params=params, timeout=self.timeout
)
if response.status_code != 200:
return {
"status": "error",
"error": "PathwayCommons search returned HTTP {}".format(
response.status_code
),
}
data = response.json()
hits = data.get("searchHit", [])
results = []
for hit in hits:
results.append(
{
"uri": hit.get("uri"),
"name": hit.get("name"),
"biopax_class": hit.get("biopaxClass"),
"data_source": hit.get("dataSource", []),
"organism": hit.get("organism", []),
"pathway": hit.get("pathway", []),
"num_participants": hit.get("numParticipants"),
"num_processes": hit.get("numProcesses"),
}
)
total_hits = data.get("numHits")
page_size = data.get("maxHitsPerPage")
actual_page = data.get("pageNo")
if (
isinstance(total_hits, bool)
or not isinstance(total_hits, int)
or total_hits < 0
or isinstance(page_size, bool)
or not isinstance(page_size, int)
or page_size <= 0
or isinstance(actual_page, bool)
or not isinstance(actual_page, int)
or actual_page < 0
):
return {
"status": "error",
"error": "PathwayCommons search returned invalid pagination metadata",
}
if actual_page != page:
return {
"status": "error",
"error": "PathwayCommons returned page {} for requested page {}".format(
actual_page, page
),
}
total_pages = math.ceil(total_hits / page_size) if total_hits else 0
expected_more = actual_page + 1 < total_pages
pagination_stalled = expected_more and not results
has_more = expected_more and not pagination_stalled
return {
"status": "success",
"data": results,
"metadata": {
"total_hits": total_hits,
"total_hits_is_exact": True,
"returned": len(results),
"max_hits_per_page": page_size,
"page": actual_page,
"total_pages": total_pages,
"has_more": has_more,
"next_page": actual_page + 1 if has_more else None,
"pagination_stalled": pagination_stalled,
"complete": not has_more and not pagination_stalled,
"query": query,
},
}
[docs]
def _traverse(self, uri: str, path: str) -> Optional[list]:
"""Helper: call the PC2 traverse endpoint."""
params = {"uri": uri, "path": path}
resp = self.session.get(
"{}/traverse".format(PC2_BASE_URL), params=params, timeout=self.timeout
)
if resp.status_code != 200:
return None
data = resp.json()
entries = data.get("traverseEntry", [])
if entries:
return entries[0].get("value", [])
return []
[docs]
def _get_pathway(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get pathway details by URI from PathwayCommons using traverse API."""
uri = arguments.get("uri")
if not uri:
return {"status": "error", "error": "Missing required parameter: uri"}
# Get pathway metadata via traverse endpoint (fast, reliable)
name = self._traverse(uri, "Pathway/displayName")
comment = self._traverse(uri, "Pathway/comment")
organism = self._traverse(uri, "Pathway/organism/displayName")
data_source = self._traverse(uri, "Pathway/dataSource/displayName")
sub_pathways = self._traverse(
uri, "Pathway/pathwayComponent:Pathway/displayName"
)
participants = self._traverse(
uri, "Pathway/pathwayComponent/participant/displayName"
)
if name is None:
return {
"status": "error",
"error": "Failed to retrieve pathway data for URI: {}".format(uri),
}
# _traverse returns None on a failed request but [] on a genuinely
# empty-but-successful one -- track which secondary fields failed to
# fetch so the response doesn't silently conflate "fetch failed"
# with "this pathway genuinely has no such data."
fetch_failures = []
def _scalar(values, label):
if values is None:
fetch_failures.append(label)
return None
return values[0] if values else None
def _list(values, label):
if values is None:
fetch_failures.append(label)
return []
return values
result = {
"pathway": {
"uri": uri,
"name": name[0] if name else None,
"description": _scalar(comment, "comment"),
"organism": _scalar(organism, "organism"),
"data_source": _scalar(data_source, "data_source"),
},
"sub_pathways": _list(sub_pathways, "sub_pathways"),
"participants": _list(participants, "participants"),
}
metadata = {
"uri": uri,
"sub_pathway_count": len(result["sub_pathways"]),
"participant_count": len(result["participants"]),
}
if fetch_failures:
metadata["fetch_failures"] = fetch_failures
return {
"status": "success",
"data": result,
"metadata": metadata,
}
[docs]
def _get_neighborhood(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Get interaction neighborhood for a gene from PathwayCommons (SIF format)."""
gene = arguments.get("gene") or arguments.get("source")
if not gene:
return {"status": "error", "error": "Missing required parameter: gene"}
limit, error = self._validated_int(arguments, "limit", 1, 1, 100)
if error:
return {"status": "error", "error": error}
max_return, error = self._validated_int(arguments, "max_results", 100, 1, 500)
if error:
return {"status": "error", "error": error}
offset, error = self._validated_int(arguments, "offset", 0, 0)
if error:
return {"status": "error", "error": error}
filters, error = self._graph_filters(arguments, allow_direction=True)
if error:
return {"status": "error", "error": error}
params = {
"source": gene,
"kind": "neighborhood",
"format": "TXT",
"limit": limit,
}
params.update(filters)
response = self.session.get(
"{}/graph".format(PC2_BASE_URL), params=params, timeout=self.timeout
)
if response.status_code != 200:
return {
"status": "error",
"error": "PathwayCommons graph returned HTTP {}".format(
response.status_code
),
}
text = response.text.strip()
if not text:
return {
"status": "success",
"data": {"interactions": [], "gene": gene},
"metadata": {
"gene": gene,
"total_interactions": 0,
"returned": 0,
"offset": offset,
"has_more": False,
"next_offset": None,
"limit": limit,
"complete_graph_retrieved": True,
"source_resolution": self._source_resolution(gene),
},
}
lines = text.splitlines()
expected_header = [
"PARTICIPANT_A",
"INTERACTION_TYPE",
"PARTICIPANT_B",
]
header = lines[0].split("\t") if lines else []
if header[:3] != expected_header:
return {
"status": "error",
"error": "PathwayCommons returned an unexpected TXT graph format",
}
interactions = []
for line in lines[1:]:
if not line.strip():
break
fields = line.split("\t")
if len(fields) >= 3:
interaction = {
"participant_a": fields[0],
"interaction_type": fields[1],
"participant_b": fields[2],
}
if len(fields) > 3:
interaction["data_source"] = fields[3] if fields[3] else None
if len(fields) > 4:
interaction["pubmed_ids"] = fields[4] if fields[4] else None
if len(fields) > 5:
interaction["pathway_names"] = fields[5] if fields[5] else None
interactions.append(interaction)
# Collect unique interaction types and partners
interaction_types = {}
partners = set()
for ix in interactions:
itype = ix["interaction_type"]
interaction_types[itype] = interaction_types.get(itype, 0) + 1
if ix["participant_a"].upper() != gene.upper():
partners.add(ix["participant_a"])
if ix["participant_b"].upper() != gene.upper():
partners.add(ix["participant_b"])
page = interactions[offset : offset + max_return]
next_offset = offset + len(page)
has_more = next_offset < len(interactions)
return {
"status": "success",
"data": {
"gene": gene,
"interactions": page,
"interaction_type_counts": interaction_types,
"unique_partners": len(partners),
},
"metadata": {
"gene": gene,
"total_interactions": len(interactions),
"returned": len(page),
"offset": offset,
"has_more": has_more,
"next_offset": next_offset if has_more else None,
"limit": limit,
"complete_graph_retrieved": True,
"source_resolution": self._source_resolution(gene),
},
}
[docs]
@staticmethod
def _source_resolution(source: str) -> str:
normalized = source.lower()
if normalized.startswith(("http://", "https://", "bioregistry.io/")):
return "exact_uri"
return "exploratory_id_mapping"
[docs]
@staticmethod
def _parse_sif_edges(text: str) -> list:
"""Parse Pathway Commons SIF output into {source, relation, target} edges.
SIF format is tab-separated triples: "ENTITY_A <relation> ENTITY_B"
(e.g. "BRCA1 in-complex-with BARD1"). No header line. Lines with fewer
than 3 tab-separated fields are skipped.
"""
edges = []
for line in text.split("\n"):
line = line.strip()
if not line:
continue
fields = line.split("\t")
if len(fields) < 3:
continue
edges.append(
{
"source": fields[0],
"relation": fields[1],
"target": fields[2],
}
)
return edges
[docs]
def _paths_between(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""Find mechanistic interaction paths connecting a set of genes.
Uses the PC2 /graph endpoint with kind=PATHSBETWEEN, sending one
``source=`` parameter per gene and parsing the SIF response into edges.
Requires at least two distinct genes.
"""
genes = arguments.get("genes")
if genes is None:
# Tolerate a comma-separated string as a convenience.
single = arguments.get("source")
if isinstance(single, str):
genes = [g.strip() for g in single.split(",")]
if isinstance(genes, str):
genes = [g.strip() for g in genes.split(",")]
if not isinstance(genes, list):
return {
"status": "error",
"error": "Missing required parameter: genes (list of gene symbols/identifiers)",
}
# Clean: reject malformed members and de-duplicate while preserving order.
seen = set()
clean_genes = []
for g in genes:
if not isinstance(g, str):
return {
"status": "error",
"error": "genes must contain only strings",
}
g = g.strip()
if not g or g.upper() in seen:
continue
seen.add(g.upper())
clean_genes.append(g)
if len(clean_genes) < 2:
return {
"status": "error",
"error": "paths_between requires at least 2 distinct genes; got {}".format(
len(clean_genes)
),
}
# requests serializes a list value into repeated key=value pairs,
# i.e. one source= per gene, which is the correct PATHSBETWEEN form.
params = {
"source": clean_genes,
"kind": "PATHSBETWEEN",
"format": "SIF",
}
limit, error = self._validated_int(arguments, "limit", 1, 1, 3)
if error:
return {"status": "error", "error": error}
max_return, error = self._validated_int(arguments, "max_results", 200, 1, 2000)
if error:
return {"status": "error", "error": error}
offset, error = self._validated_int(arguments, "offset", 0, 0)
if error:
return {"status": "error", "error": error}
params["limit"] = limit
filters, error = self._graph_filters(arguments, allow_direction=False)
if error:
return {"status": "error", "error": error}
params.update(filters)
# PATHSBETWEEN graph queries are heavier than search; allow a longer
# per-request budget but cap at 30s to satisfy the contract.
response = self.session.get(
"{}/graph".format(PC2_BASE_URL), params=params, timeout=30
)
if response.status_code != 200:
return {
"status": "error",
"error": "PathwayCommons graph returned HTTP {}".format(
response.status_code
),
}
text = response.text.strip()
edges = self._parse_sif_edges(text)
relation_counts = {}
entities = set()
for edge in edges:
relation_counts[edge["relation"]] = (
relation_counts.get(edge["relation"], 0) + 1
)
entities.add(edge["source"])
entities.add(edge["target"])
note = None
if not edges:
note = (
"No mechanistic paths found connecting the given genes. "
"Check gene symbols (HGNC) and that at least two are present "
"in Pathway Commons."
)
page = edges[offset : offset + max_return]
next_offset = offset + len(page)
has_more = next_offset < len(edges)
return {
"status": "success",
"data": {
"genes": clean_genes,
"edges": page,
"relation_counts": relation_counts,
"entity_count": len(entities),
"note": note,
},
"metadata": {
"genes": clean_genes,
"total_edges": len(edges),
"returned": len(page),
"offset": offset,
"has_more": has_more,
"next_offset": next_offset if has_more else None,
"limit": limit,
"complete_graph_retrieved": True,
"source_resolution": [
self._source_resolution(gene) for gene in clean_genes
],
},
}