Spaces:
Running
Running
File size: 3,903 Bytes
8dcf472 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 | """
CodeInterpreterTool — local sandboxed Python execution.
Replaces the crewai_tools.CodeInterpreterTool removed in crewai v1.x.
Runs Python code in a subprocess with a strict timeout and whitelist.
"""
from __future__ import annotations
import json
import subprocess
import sys
import textwrap
from pathlib import Path
from crewai.tools import BaseTool
from loguru import logger
# ─── Allowlist of safe imports ────────────────────────────────────────────────
_ALLOWED_IMPORTS = frozenset(
[
"pandas", "numpy", "math", "statistics", "json", "csv", "io",
"datetime", "re", "collections", "itertools", "functools",
"matplotlib", "matplotlib.pyplot",
]
)
_BLOCKED_KEYWORDS = frozenset(
[
"__import__", "exec", "eval", "compile", "open", "os", "sys",
"subprocess", "socket", "urllib", "requests", "shutil", "glob",
"importlib", "ctypes", "pickle", "marshal",
]
)
def _is_safe_code(code: str) -> tuple[bool, str]:
"""Basic static check — not a full sandbox, just a sanity layer."""
for kw in _BLOCKED_KEYWORDS:
if kw in code:
return False, f"Blocked keyword in code: '{kw}'"
return True, ""
class CodeInterpreterTool(BaseTool):
"""
Sandboxed Python code execution tool for financial calculations.
Implements the interface of the deprecated crewai_tools.CodeInterpreterTool.
Accepts Python code as a string. Returns stdout + last expression result.
"""
name: str = "code_interpreter"
description: str = (
"Execute Python code for financial calculations. "
"Supports pandas, numpy, math, statistics, datetime. "
"Input: a string of Python code. "
"Output: stdout output and/or the result of the last expression. "
"Use for: revenue projections, unit economics, CAC/LTV, burn rate, market sizing."
)
timeout_seconds: int = 30
def _run(self, code: str) -> str:
"""Execute Python code in a subprocess and return the output."""
if not code or not code.strip():
return "Error: no code provided."
safe, reason = _is_safe_code(code)
if not safe:
return f"Error: unsafe code detected — {reason}"
# Wrap in a try/except and auto-print last expression
wrapped = textwrap.dedent(f"""
import pandas as pd
import numpy as np
import math
import statistics
import json
import csv
import io
import datetime
import re
import collections
import itertools
import functools
try:
import matplotlib
matplotlib.use('Agg')
import matplotlib.pyplot as plt
except Exception:
pass
# ─── User code ────────────────────────
{code}
""").strip()
try:
result = subprocess.run(
[sys.executable, "-c", wrapped],
capture_output=True,
text=True,
timeout=self.timeout_seconds,
)
output = result.stdout.strip()
err = result.stderr.strip()
if result.returncode != 0:
# Surface the traceback but strip local paths
safe_err = "\n".join(
ln for ln in err.splitlines()
if "/home/" not in ln and "/tmp/" not in ln
)
return f"Code execution error:\n{safe_err}" if safe_err else f"Exit code {result.returncode}"
if output:
return output
return "(code ran successfully, no output)"
except subprocess.TimeoutExpired:
return f"Error: code execution timed out after {self.timeout_seconds}s."
except Exception as exc:
logger.error(f"CodeInterpreterTool: {exc}")
return f"Error: {exc}"
|