python_repl: PythonREPL = Field(default_factory=_get_default_python_repl)
    sanitize_input: bool = True
    def _run(
        self,
        query: str,
        run_manager: Optional[CallbackManagerForToolRun] = None,
    ) -> Any:
        """Use the tool."""
        if self.sanitize_input:
            query = sanitize_input(query)
        return self.python_repl.run(query)
    async def _arun(
        self,
        query: str,
        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
    ) -> Any:
        """Use the tool asynchronously."""
        if self.sanitize_input:
            query = sanitize_input(query)
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(None, self.run, query)
        return result
[docs]class PythonInputs(BaseModel):
    query: str = Field(description="code snippet to run")
[docs]class PythonAstREPLTool(BaseTool):
    """A tool for running python code in a REPL."""
    name: str = "python_repl_ast"
    description: str = (
        "A Python shell. Use this to execute python commands. "
        "Input should be a valid python command. "
        "When using this tool, sometimes output is abbreviated - "
        "make sure it does not look abbreviated before using it in your answer."
    )
    globals: Optional[Dict] = Field(default_factory=dict)
    locals: Optional[Dict] = Field(default_factory=dict)
    sanitize_input: bool = True
    args_schema: Type[BaseModel] = PythonInputs
    @root_validator(pre=True)
    def validate_python_version(cls, values: Dict) -> Dict: