Source code for langchain_community.tools.connery.tool

import asyncio
from functools import partial
from typing import Any, Dict, List, Optional, Type

from langchain_core.callbacks.manager import (
    AsyncCallbackManagerForToolRun,
    CallbackManagerForToolRun,
)
from langchain_core.pydantic_v1 import BaseModel, Field, create_model, root_validator
from langchain_core.tools import BaseTool

from langchain_community.tools.connery.models import Action, Parameter


[docs]class ConneryAction(BaseTool): """Connery Action tool.""" name: str description: str args_schema: Type[BaseModel] action: Action connery_service: Any def _run( self, run_manager: Optional[CallbackManagerForToolRun] = None, **kwargs: Dict[str, str], ) -> Dict[str, str]: """ Runs the Connery Action with the provided input. Parameters: kwargs (Dict[str, str]): The input dictionary expected by the action. Returns: Dict[str, str]: The output of the action. """ return self.connery_service.run_action(self.action.id, kwargs) async def _arun( self, run_manager: Optional[AsyncCallbackManagerForToolRun] = None, **kwargs: Dict[str, str], ) -> Dict[str, str]: """ Runs the Connery Action asynchronously with the provided input. Parameters: kwargs (Dict[str, str]): The input dictionary expected by the action. Returns: Dict[str, str]: The output of the action. """ func = partial(self._run, **kwargs) return await asyncio.get_event_loop().run_in_executor(None, func)
[docs] def get_schema_json(self) -> str: """ Returns the JSON representation of the Connery Action Tool schema. This is useful for debugging. Returns: str: The JSON representation of the Connery Action Tool schema. """ return self.args_schema.schema_json(indent=2)
@root_validator() def validate_attributes(cls, values: dict) -> dict: """ Validate the attributes of the ConneryAction class. Parameters: values (dict): The arguments to validate. Returns: dict: The validated arguments. """ # Import ConneryService here and check if it is an instance # of ConneryService to avoid circular imports from .service import ConneryService if not isinstance(values.get("connery_service"), ConneryService): raise ValueError( "The attribute 'connery_service' must be an instance of ConneryService." ) if not values.get("name"): raise ValueError("The attribute 'name' must be set.") if not values.get("description"): raise ValueError("The attribute 'description' must be set.") if not values.get("args_schema"): raise ValueError("The attribute 'args_schema' must be set.") if not values.get("action"): raise ValueError("The attribute 'action' must be set.") if not values.get("connery_service"): raise ValueError("The attribute 'connery_service' must be set.") return values
[docs] @classmethod def create_instance(cls, action: Action, connery_service: Any) -> "ConneryAction": """ Creates a Connery Action Tool from a Connery Action. Parameters: action (Action): The Connery Action to wrap in a Connery Action Tool. connery_service (ConneryService): The Connery Service to run the Connery Action. We use Any here to avoid circular imports. Returns: ConneryAction: The Connery Action Tool. """ # Import ConneryService here and check if it is an instance # of ConneryService to avoid circular imports from .service import ConneryService if not isinstance(connery_service, ConneryService): raise ValueError( "The connery_service must be an instance of ConneryService." ) input_schema = cls._create_input_schema(action.inputParameters) description = action.title + ( ": " + action.description if action.description else "" ) instance = cls( name=action.id, description=description, args_schema=input_schema, action=action, connery_service=connery_service, ) return instance
@classmethod def _create_input_schema(cls, inputParameters: List[Parameter]) -> Type[BaseModel]: """ Creates an input schema for a Connery Action Tool based on the input parameters of the Connery Action. Parameters: inputParameters: List of input parameters of the Connery Action. Returns: Type[BaseModel]: The input schema for the Connery Action Tool. """ dynamic_input_fields: Dict[str, Any] = {} for param in inputParameters: default = ... if param.validation and param.validation.required else None title = param.title description = param.title + ( ": " + param.description if param.description else "" ) type = param.type dynamic_input_fields[param.key] = ( str, Field(default, title=title, description=description, type=type), ) InputModel = create_model("InputSchema", **dynamic_input_fields) return InputModel