Source code for tooluniverse.stitch_tool

# stitch_tool.py
"""
STITCH (Search Tool for Interacting Chemicals) API tool for ToolUniverse.

STITCH is a database of known and predicted interactions between
chemicals and proteins, combining data from various sources.

API Documentation: http://stitch.embl.de/
"""

import warnings
import requests
import urllib3
from typing import Dict, Any
from .base_tool import BaseTool
from .tool_registry import register_tool

# Base URL for STITCH REST API (chemical-protein interactions)
# NOTE: stitch.embl.de may have SSL certificate issues on some systems; use verify=False
STITCH_BASE_URL = "https://stitch.embl.de/api"




@register_tool("STITCHTool")
class STITCHTool(BaseTool):
    """
    Tool for querying STITCH database.

    STITCH provides chemical-protein interaction data including:
    - Known drug-target interactions
    - Predicted chemical-protein interactions
    - Interaction scores and evidence
    - Network analysis data

    No authentication required. Free for academic/research use.
    """



    def __init__(self, tool_config: Dict[str, Any]):
        super().__init__(tool_config)
        self.timeout = tool_config.get("timeout", 30)
        self.operation = tool_config.get("fields", {}).get(
            "operation", "get_interactions"
        )




    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Execute the STITCH API call."""
        operation = self.operation

        if operation == "get_interactions":
            return self._get_interactions(arguments)
        elif operation == "get_interactors":
            return self._get_interactors(arguments)
        elif operation == "resolve":
            return self._resolve_identifiers(arguments)
        else:
            return {"error": f"Unknown operation: {operation}"}




    def _get_interactions(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """
        Get chemical-protein interactions.

        Endpoint: GET /json/interactions
        """
        identifiers = arguments.get("identifiers", [])

        if not identifiers:
            return {
                "error": "identifiers parameter is required (chemical names or IDs)"
            }

        if isinstance(identifiers, str):
            identifiers = [identifiers]

        params = {
            "identifiers": "%0D".join(identifiers),  # URL-encoded newline separator
            "species": arguments.get("species", 9606),  # Default: human
            "limit": arguments.get("limit", 10),
            "required_score": arguments.get("required_score", 400),  # Medium confidence
        }

        try:
            with warnings.catch_warnings():
                warnings.simplefilter(
                    "ignore", urllib3.exceptions.InsecureRequestWarning
                )
                response = requests.get(
                    f"{STITCH_BASE_URL}/json/interactions",
                    params=params,
                    timeout=self.timeout,
                    verify=False,
                )
            if response.status_code == 404:
                return {
                    "error": f"No interactions found for {identifiers} in STITCH. "
                    "Try using CID identifiers (e.g., 'CIDm00002244' for aspirin) "
                    "or check compound names at http://stitch.embl.de/"
                }
            response.raise_for_status()
            return {"interactions": response.json()}
        except requests.RequestException as e:
            return {"error": f"STITCH API request failed: {str(e)}"}




    def _get_interactors(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """
        Get interaction partners for a chemical or protein.

        Endpoint: GET /json/interactors
        """
        identifiers = arguments.get("identifiers", [])

        if not identifiers:
            return {"error": "identifiers parameter is required"}

        if isinstance(identifiers, str):
            identifiers = [identifiers]

        params = {
            "identifiers": "%0D".join(identifiers),
            "species": arguments.get("species", 9606),
            "limit": arguments.get("limit", 10),
        }

        try:
            with warnings.catch_warnings():
                warnings.simplefilter(
                    "ignore", urllib3.exceptions.InsecureRequestWarning
                )
                response = requests.get(
                    f"{STITCH_BASE_URL}/json/network",
                    params=params,
                    timeout=self.timeout,
                    verify=False,
                )
            if response.status_code == 404:
                return {
                    "error": f"No interactors found for {identifiers} in STITCH. "
                    "Try using CID identifiers or check compound names at http://stitch.embl.de/"
                }
            response.raise_for_status()
            return {"interactors": response.json()}
        except requests.RequestException as e:
            return {"error": f"STITCH API request failed: {str(e)}"}




    def _resolve_identifiers(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """
        Resolve chemical/protein names to STITCH identifiers.

        Endpoint: GET /json/resolve
        """
        identifier = arguments.get("identifier", "")

        if not identifier:
            return {"error": "identifier parameter is required"}

        params = {"identifier": identifier, "species": arguments.get("species", 9606)}

        try:
            with warnings.catch_warnings():
                warnings.simplefilter(
                    "ignore", urllib3.exceptions.InsecureRequestWarning
                )
                response = requests.get(
                    f"{STITCH_BASE_URL}/json/resolve",
                    params=params,
                    timeout=self.timeout,
                    verify=False,
                )
            if response.status_code == 404:
                return {
                    "error": f"Identifier '{identifier}' not found in STITCH. "
                    "Try using CID identifiers or check at http://stitch.embl.de/"
                }
            response.raise_for_status()
            return {"matches": response.json()}
        except requests.RequestException as e:
            return {"error": f"STITCH API request failed: {str(e)}"}