Tool that takes in function or coroutine directly.
| 568 | |
| 569 | |
| 570 | class Tool(BaseTool): |
| 571 | """Tool that takes in function or coroutine directly.""" |
| 572 | |
| 573 | description: str = "" |
| 574 | func: Optional[Callable[..., str]] |
| 575 | """The function to run when the tool is called.""" |
| 576 | coroutine: Optional[Callable[..., Awaitable[str]]] = None |
| 577 | """The asynchronous version of the function.""" |
| 578 | |
| 579 | # --- Runnable --- |
| 580 | |
| 581 | async def ainvoke( |
| 582 | self, |
| 583 | input: Union[str, Dict], |
| 584 | config: Optional[RunnableConfig] = None, |
| 585 | **kwargs: Any, |
| 586 | ) -> Any: |
| 587 | if not self.coroutine: |
| 588 | # If the tool does not implement async, fall back to default implementation |
| 589 | return await run_in_executor(config, self.invoke, input, config, **kwargs) |
| 590 | |
| 591 | return await super().ainvoke(input, config, **kwargs) |
| 592 | |
| 593 | # --- Tool --- |
| 594 | |
| 595 | @property |
| 596 | def args(self) -> dict: |
| 597 | """The tool's input arguments.""" |
| 598 | if self.args_schema is not None: |
| 599 | return self.args_schema.schema()["properties"] |
| 600 | # For backwards compatibility, if the function signature is ambiguous, |
| 601 | # assume it takes a single string input. |
| 602 | return {"tool_input": {"type": "string"}} |
| 603 | |
| 604 | def _to_args_and_kwargs(self, tool_input: Union[str, Dict]) -> Tuple[Tuple, Dict]: |
| 605 | """Convert tool input to pydantic model.""" |
| 606 | args, kwargs = super()._to_args_and_kwargs(tool_input) |
| 607 | # For backwards compatibility. The tool must be run with a single input |
| 608 | all_args = list(args) + list(kwargs.values()) |
| 609 | if len(all_args) != 1: |
| 610 | raise ToolException( |
| 611 | f"""Too many arguments to single-input tool {self.name}. |
| 612 | Consider using StructuredTool instead.""" |
| 613 | f" Args: {all_args}" |
| 614 | ) |
| 615 | return tuple(all_args), {} |
| 616 | |
| 617 | def _run( |
| 618 | self, |
| 619 | *args: Any, |
| 620 | run_manager: Optional[CallbackManagerForToolRun] = None, |
| 621 | **kwargs: Any, |
| 622 | ) -> Any: |
| 623 | """Use the tool.""" |
| 624 | if self.func: |
| 625 | new_argument_supported = signature(self.func).parameters.get("callbacks") |
| 626 | return ( |
| 627 | self.func( |
no outgoing calls