tools/python_exec_tool.py

from langchain.tools import BaseTool
import subprocess
import sys
import tempfile
import os
import requests

class PythonExecTool(BaseTool):
    """
    Инструмент для выполнения прикрепленного Python-кода и возврата его вывода.
    Формат команды: URL или локальный путь к .py файлу.
    """
    name: str = "exec_python"
    description: str = (
        "Run Python scripts/files and return their output. Use for any question that provides a .py file and asks ‘what does this code print?’, ‘what is the return value?’, etc."
    )

    def _run(self, source: str) -> str:
        # определяем локальный путь к файлу
        if source.startswith("http://") or source.startswith("https://"):
            # скачиваем файл
            resp = requests.get(source)
            resp.raise_for_status()
            tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".py")
            tmp.write(resp.content)
            tmp.flush()
            file_path = tmp.name
        else:
            file_path = source

        # выполняем файл и захватываем вывод
        try:
            result = subprocess.run(
                [sys.executable, file_path],
                capture_output=True,
                text=True,
                check=False
            )
            # stdout содержит финальный вывод
            output = result.stdout.strip()
            if not output:
                # если stdout пуст, возвращаем stderr
                output = result.stderr.strip()
            return output
        finally:
            # удаляем временный файл, если он был скачан
            if source.startswith("http://") or source.startswith("https://"):
                os.remove(file_path)

    async def _arun(self, source: str) -> str:
        raise NotImplementedError("Async not supported.")