[docs]classConneryAction(BaseTool):# type: ignore[override, override]"""Connery Action tool."""name:strdescription:strargs_schema:Type[BaseModel]action:Actionconnery_service:Anydef_run(self,run_manager:Optional[CallbackManagerForToolRun]=None,**kwargs:Any,)->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. """returnself.connery_service.run_action(self.action.id,kwargs)asyncdef_arun(self,run_manager:Optional[AsyncCallbackManagerForToolRun]=None,**kwargs:Any,)->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)returnawaitasyncio.get_event_loop().run_in_executor(None,func)
[docs]defget_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. """returnself.args_schema.schema_json(indent=2)
@model_validator(mode="before")@classmethoddefvalidate_attributes(cls,values:dict)->Any:""" 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 importsfrom.serviceimportConneryServiceifnotisinstance(values.get("connery_service"),ConneryService):raiseValueError("The attribute 'connery_service' must be an instance of ConneryService.")ifnotvalues.get("name"):raiseValueError("The attribute 'name' must be set.")ifnotvalues.get("description"):raiseValueError("The attribute 'description' must be set.")ifnotvalues.get("args_schema"):raiseValueError("The attribute 'args_schema' must be set.")ifnotvalues.get("action"):raiseValueError("The attribute 'action' must be set.")ifnotvalues.get("connery_service"):raiseValueError("The attribute 'connery_service' must be set.")returnvalues
[docs]@classmethoddefcreate_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 importsfrom.serviceimportConneryServiceifnotisinstance(connery_service,ConneryService):raiseValueError("The connery_service must be an instance of ConneryService.")input_schema=cls._create_input_schema(action.inputParameters)description=action.title+(": "+action.descriptionifaction.descriptionelse"")instance=cls(name=action.id,description=description,args_schema=input_schema,action=action,connery_service=connery_service,)returninstance
@classmethoddef_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]={}forparamininputParameters:default=...ifparam.validationandparam.validation.requiredelseNonetitle=param.titledescription=param.title+(": "+param.descriptionifparam.descriptionelse"")type=param.typedynamic_input_fields[param.key]=(type,Field(default,title=title,description=description),)InputModel=create_model("InputSchema",**dynamic_input_fields)returnInputModel