Method arun

libs/core/langchain_core/tools.py:457–562

Run the tool asynchronously.

(
        self,
        tool_input: Union[str, Dict],
        verbose: Optional[bool] = None,
        start_color: Optional[str] = "green",
        color: Optional[str] = "green",
        callbacks: Callbacks = None,
        *,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        run_name: Optional[str] = None,
        run_id: Optional[uuid.UUID] = None,
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    )

            return observation

    async def arun(
        self,
        tool_input: Union[str, Dict],
        verbose: Optional[bool] = None,
        start_color: Optional[str] = "green",
        color: Optional[str] = "green",
        callbacks: Callbacks = None,
        *,
        tags: Optional[List[str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
        run_name: Optional[str] = None,
        run_id: Optional[uuid.UUID] = None,
        config: Optional[RunnableConfig] = None,
        **kwargs: Any,
    ) -> Any:
        """Run the tool asynchronously."""
        if not self.verbose and verbose is not None:
            verbose_ = verbose
        else:
            verbose_ = self.verbose
        callback_manager = AsyncCallbackManager.configure(
            callbacks,
            self.callbacks,
            verbose_,
            tags,
            self.tags,
            metadata,
            self.metadata,
        )
        new_arg_supported = signature(self._arun).parameters.get("run_manager")
        run_manager = await callback_manager.on_tool_start(
            {"name": self.name, "description": self.description},
            tool_input if isinstance(tool_input, str) else str(tool_input),
            color=start_color,
            name=run_name,
            inputs=tool_input,
            run_id=run_id,
            **kwargs,
        )
        try:
            parsed_input = self._parse_input(tool_input)
            # We then call the tool on the tool input to get an observation
            tool_args, tool_kwargs = self._to_args_and_kwargs(parsed_input)
            child_config = patch_config(
                config,
                callbacks=run_manager.get_child(),
            )
            context = copy_context()
            context.run(_set_config_context, child_config)
            coro = (
                context.run(
                    self._arun, *tool_args, run_manager=run_manager, **tool_kwargs
                )
                if new_arg_supported
                else context.run(self._arun, *tool_args, **tool_kwargs)
            )
            if accepts_context(asyncio.create_task):
                observation = await asyncio.create_task(coro, context=context)  # type: ignore