codekingpro/portable-devtools

1import json2from typing import Any, Dict, List, Optional3 4import requests5from langchain_core.utils.env import get_from_dict_or_env6from pydantic import BaseModel, model_validator7 8from langchain_community.tools.connery.models import Action9from langchain_community.tools.connery.tool import ConneryAction10 11 12class ConneryService(BaseModel):13    """Service for interacting with the Connery Runner API.14 15    It gets the list of available actions from the Connery Runner,16    wraps them in ConneryAction Tools and returns them to the user.17    It also provides a method for running the actions.18    """19 20    runner_url: Optional[str] = None21    api_key: Optional[str] = None22 23    @model_validator(mode="before")24    @classmethod25    def validate_attributes(cls, values: Dict) -> Any:26        """27        Validate the attributes of the ConneryService class.28        Parameters:29            values (dict): The arguments to validate.30        Returns:31            dict: The validated arguments.32        """33 34        runner_url = get_from_dict_or_env(values, "runner_url", "CONNERY_RUNNER_URL")35        api_key = get_from_dict_or_env(values, "api_key", "CONNERY_RUNNER_API_KEY")36 37        if not runner_url:38            raise ValueError("CONNERY_RUNNER_URL environment variable must be set.")39        if not api_key:40            raise ValueError("CONNERY_RUNNER_API_KEY environment variable must be set.")41 42        values["runner_url"] = runner_url43        values["api_key"] = api_key44 45        return values46 47    def list_actions(self) -> List[ConneryAction]:48        """49        Returns the list of actions available in the Connery Runner.50        Returns:51            List[ConneryAction]: The list of actions available in the Connery Runner.52        """53 54        return [55            ConneryAction.create_instance(action, self)56            for action in self._list_actions()57        ]58 59    def get_action(self, action_id: str) -> ConneryAction:60        """61        Returns the specified action available in the Connery Runner.62        Parameters:63            action_id (str): The ID of the action to return.64        Returns:65            ConneryAction: The action with the specified ID.66        """67 68        return ConneryAction.create_instance(self._get_action(action_id), self)69 70    def run_action(self, action_id: str, input: Dict[str, str] = {}) -> Dict[str, str]:71        """72        Runs the specified Connery Action with the provided input.73        Parameters:74            action_id (str): The ID of the action to run.75            input (Dict[str, str]): The input object expected by the action.76        Returns:77            Dict[str, str]: The output of the action.78        """79 80        return self._run_action(action_id, input)81 82    def _list_actions(self) -> List[Action]:83        """84        Returns the list of actions available in the Connery Runner.85        Returns:86            List[Action]: The list of actions available in the Connery Runner.87        """88 89        response = requests.get(90            f"{self.runner_url}/v1/actions", headers=self._get_headers()91        )92 93        if not response.ok:94            raise ValueError(95                (96                    "Failed to list actions."97                    f"Status code: {response.status_code}."98                    f"Error message: {response.json()['error']['message']}"99                )100            )101 102        return [Action(**action) for action in response.json()["data"]]103 104    def _get_action(self, action_id: str) -> Action:105        """106        Returns the specified action available in the Connery Runner.107        Parameters:108            action_id (str): The ID of the action to return.109        Returns:110            Action: The action with the specified ID.111        """112 113        actions = self._list_actions()114        action = next((action for action in actions if action.id == action_id), None)115        if not action:116            raise ValueError(117                (118                    f"The action with ID {action_id} was not found in the list"119                    "of available actions in the Connery Runner."120                )121            )122        return action123 124    def _run_action(self, action_id: str, input: Dict[str, str] = {}) -> Dict[str, str]:125        """126        Runs the specified Connery Action with the provided input.127        Parameters:128            action_id (str): The ID of the action to run.129            prompt (str): This is a plain English prompt130            with all the information needed to run the action.131            input (Dict[str, str]): The input object expected by the action.132            If provided together with the prompt,133            the input takes precedence over the input specified in the prompt.134        Returns:135            Dict[str, str]: The output of the action.136        """137 138        response = requests.post(139            f"{self.runner_url}/v1/actions/{action_id}/run",140            headers=self._get_headers(),141            data=json.dumps({"input": input}),142        )143 144        if not response.ok:145            raise ValueError(146                (147                    "Failed to run action."148                    f"Status code: {response.status_code}."149                    f"Error message: {response.json()['error']['message']}"150                )151            )152 153        if not response.json()["data"]["output"]:154            return {}155        else:156            return response.json()["data"]["output"]157 158    def _get_headers(self) -> Dict[str, str]:159        """160        Returns a standard set of HTTP headers161        to be used in API calls to the Connery runner.162        Returns:163            Dict[str, str]: The standard set of HTTP headers.164        """165 166        return {"Content-Type": "application/json", "x-api-key": self.api_key or ""}167