| 212 | |
| 213 | @override |
| 214 | async def run_async( |
| 215 | self, *, args: dict[str, Any], tool_context: ToolContext |
| 216 | ) -> Any: |
| 217 | # Preprocess arguments (includes Pydantic model conversion) |
| 218 | args_to_call = self._preprocess_args(args) |
| 219 | |
| 220 | signature = inspect.signature(self.func) |
| 221 | valid_params = {param for param in signature.parameters} |
| 222 | if self._context_param_name in valid_params: |
| 223 | args_to_call[self._context_param_name] = tool_context |
| 224 | |
| 225 | # Filter args_to_call to only include valid parameters for the function |
| 226 | args_to_call = {k: v for k, v in args_to_call.items() if k in valid_params} |
| 227 | |
| 228 | # Before invoking the function, we check for if the list of args passed in |
| 229 | # has all the mandatory arguments or not. |
| 230 | # If the check fails, then we don't invoke the tool and let the Agent know |
| 231 | # that there was a missing input parameter. This will basically help |
| 232 | # the underlying model fix the issue and retry. |
| 233 | mandatory_args = self._get_mandatory_args() |
| 234 | missing_mandatory_args = [ |
| 235 | arg for arg in mandatory_args if arg not in args_to_call |
| 236 | ] |
| 237 | |
| 238 | if missing_mandatory_args: |
| 239 | missing_mandatory_args_str = '\n'.join(missing_mandatory_args) |
| 240 | error_str = f"""Invoking `{self.name}()` failed as the following mandatory input parameters are not present: |
| 241 | {missing_mandatory_args_str} |
| 242 | You could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters.""" |
| 243 | return {'error': error_str} |
| 244 | |
| 245 | if isinstance(self._require_confirmation, Callable): |
| 246 | require_confirmation = await self._invoke_callable( |
| 247 | self._require_confirmation, args_to_call |
| 248 | ) |
| 249 | else: |
| 250 | require_confirmation = bool(self._require_confirmation) |
| 251 | |
| 252 | if require_confirmation: |
| 253 | if not tool_context.tool_confirmation: |
| 254 | args_to_show = args_to_call.copy() |
| 255 | if self._context_param_name in args_to_show: |
| 256 | args_to_show.pop(self._context_param_name) |
| 257 | |
| 258 | tool_context.request_confirmation( |
| 259 | hint=( |
| 260 | f'Please approve or reject the tool call {self.name}() by' |
| 261 | ' responding with a FunctionResponse with an expected' |
| 262 | ' ToolConfirmation payload.' |
| 263 | ), |
| 264 | ) |
| 265 | tool_context.actions.skip_summarization = True |
| 266 | return { |
| 267 | 'error': ( |
| 268 | 'This tool call requires confirmation, please approve or' |
| 269 | ' reject.' |
| 270 | ) |
| 271 | } |