Source code for tooluniverse.intact_tool

"""
IntAct Molecular Interaction Database Tool

This tool provides access to the IntAct database for protein-protein interactions,
molecular interactions, and interaction evidence.
"""

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




[docs]
@register_tool("IntActRESTTool")
class IntActRESTTool(BaseTool):
    """
    IntAct REST API tool.
    Generic wrapper for IntAct API endpoints defined in intact_tools.json.
    """



[docs]
    def __init__(self, tool_config: Dict):
        super().__init__(tool_config)
        self.base_url = "https://www.ebi.ac.uk/intact/ws"
        self.session = requests.Session()
        self.session.headers.update(
            {"Accept": "application/json", "User-Agent": "ToolUniverse/1.0"}
        )
        self.timeout = 30





[docs]
    def _build_url(self, args: Dict[str, Any]) -> str:
        """Build URL from endpoint template and arguments"""
        endpoint_template = self.tool_config["fields"].get("endpoint", "")
        tool_name = self.tool_config.get("name", "")

        if endpoint_template:
            url = endpoint_template
            # Replace placeholders in URL
            for k, v in args.items():
                url = url.replace(f"{{{k}}}", str(v))
            return url

        # Build URL based on tool name
        if tool_name == "intact_get_interactor":
            identifier = args.get("identifier", "")
            if identifier:
                return f"{self.base_url}/interactor/details/{identifier}"

        elif tool_name == "intact_get_interactions":
            identifier = args.get("identifier", "")
            if identifier:
                return f"{self.base_url}/interaction/findInteractions"

        elif tool_name == "intact_search_interactions":
            return f"{self.base_url}/interaction/find"

        elif tool_name == "intact_get_interaction_details":
            interaction_id = args.get("interaction_id", "")
            if interaction_id:
                return f"{self.base_url}/interaction/{interaction_id}"

        elif tool_name == "intact_get_interaction_network":
            identifier = args.get("identifier", "")
            if identifier:
                return f"{self.base_url}/interaction/network/{identifier}"

        return self.base_url





[docs]
    def _build_params(self, args: Dict[str, Any]) -> Dict[str, Any]:
        """Build query parameters for IntAct API"""
        params = {}
        tool_name = self.tool_config.get("name", "")

        # For search operations
        if tool_name == "intact_search_interactions":
            if "query" in args:
                params["query"] = args["query"]
            else:
                params["query"] = "*"
            params["format"] = args.get("format", "json")

        # For interaction retrieval by identifier
        elif tool_name == "intact_get_interactions":
            identifier = args.get("identifier", "")
            if identifier:
                params["query"] = identifier
            params["format"] = args.get("format", "json")

        # For interactor retrieval
        elif tool_name == "intact_get_interactor":
            params["format"] = args.get("format", "json")

        # For interaction details
        elif tool_name == "intact_get_interaction_details":
            params["format"] = args.get("format", "json")

        # For network
        elif tool_name == "intact_get_interaction_network":
            if "format" in args:
                params["format"] = args["format"]
            else:
                params["format"] = "json"
            if "depth" in args:
                params["depth"] = args["depth"]

        return params





[docs]
    def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Execute the IntAct API call"""
        tool_name = self.tool_config.get("name", "")

        # Use Complex Web Service for complex queries
        if tool_name == "intact_get_interactions_by_complex":
            return self._use_complex_web_service(arguments)

        elif tool_name == "intact_get_complex_details":
            return self._get_complex_details(arguments)
        # Use EBI Search API as primary method since IntAct direct API is unreliable
        # EBI Search has an 'intact' domain that works reliably
        if tool_name in [
            "intact_get_interactions",
            "intact_search_interactions",
            "intact_get_interactor",
            "intact_get_interactions_by_organism",
        ]:
            return self._use_ebi_search(arguments, tool_name)

        try:
            # Build URL
            url = self._build_url(arguments)

            # Build parameters
            params = self._build_params(arguments)

            # Make API request
            response = self.session.get(url, params=params, timeout=self.timeout)

            # Check if response is HTML (API endpoint not available)
            content_type = response.headers.get("content-type", "")
            if "text/html" in content_type:
                # Fallback to EBI Search
                return self._use_ebi_search(arguments, tool_name)

            response.raise_for_status()

            # Parse JSON response
            data = response.json()

            # Build response
            response_data = {
                "status": "success",
                "data": data,
                "url": response.url,
            }

            # Add count for list results
            if isinstance(data, list):
                response_data["count"] = len(data)
            elif isinstance(data, dict) and "data" in data:
                if isinstance(data["data"], list):
                    response_data["count"] = len(data["data"])

            return response_data

        except requests.exceptions.RequestException:
            # Fallback to EBI Search if direct API fails
            return self._use_ebi_search(arguments, tool_name)
        except Exception as e:
            return {
                "status": "error",
                "error": f"Unexpected error: {str(e)}",
                "url": url if "url" in locals() else None,
            }








[docs]
    def _use_complex_web_service(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Use IntAct Complex Web Service for complex queries"""
        try:
            complex_id = arguments.get("complex_id", "")
            if not complex_id:
                return {
                    "status": "error",
                    "error": "complex_id parameter is required",
                }

            # Complex Web Service endpoint - query goes in path, not params
            from urllib.parse import quote

            complex_id_encoded = quote(complex_id, safe="")
            complex_url = (
                f"https://www.ebi.ac.uk/intact/complex-ws/search/{complex_id_encoded}"
            )
            params = {
                "format": "json",
                "first": arguments.get("first", 0),
                "number": arguments.get("size", 25),
            }

            response = self.session.get(
                complex_url, params=params, timeout=self.timeout
            )
            response.raise_for_status()
            data = response.json()

            # Extract complex information
            elements = data.get("elements", [])
            total = data.get("totalNumberOfResults", 0)

            # Extract complex ACs for reference
            complex_ac_list = []
            for element in elements:
                complex_ac = element.get("complexAC", "")
                if complex_ac:
                    complex_ac_list.append(complex_ac)

            return {
                "status": "success",
                "data": elements,
                "url": response.url,
                "count": len(elements),
                "totalNumberOfResults": total,
                "complex_ac_list": complex_ac_list,
                "note": "Data retrieved via IntAct Complex Web Service. Use complex_ac_list to reference specific complexes.",
            }
        except Exception as e:
            return {
                "status": "error",
                "error": f"IntAct Complex Web Service query failed: {str(e)}",
            }





[docs]
    def _get_complex_details(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
        """Get detailed information about a specific complex by complex AC"""
        try:
            complex_ac = arguments.get("complex_ac", "")
            if not complex_ac:
                return {
                    "status": "error",
                    "error": "complex_ac parameter is required (e.g., 'CPX-915')",
                }

            # Complex Web Service details endpoint
            complex_url = (
                f"https://www.ebi.ac.uk/intact/complex-ws/complex/{complex_ac}"
            )
            params = {
                "format": "json",
            }

            response = self.session.get(
                complex_url, params=params, timeout=self.timeout
            )
            response.raise_for_status()
            data = response.json()

            return {
                "status": "success",
                "data": data,
                "url": response.url,
                "complex_ac": data.get("complexAc", complex_ac),
                "complex_name": data.get("name", ""),
                "note": "Data retrieved via IntAct Complex Web Service. Includes complex details, participants, functions, properties, and related information.",
            }
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 404:
                return {
                    "status": "error",
                    "error": f"Complex '{complex_ac}' not found. Verify the complex AC is correct (e.g., 'CPX-915'). Use intact_get_interactions_by_complex to search for complexes.",
                }
            return {
                "status": "error",
                "error": f"IntAct Complex Web Service query failed: {str(e)}",
            }
        except Exception as e:
            return {
                "status": "error",
                "error": f"IntAct Complex Web Service query failed: {str(e)}",
            }