File size: 2,438 Bytes
070daf8 | 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 | """
Local Code Execution Tool - Execute Python and bash scripts locally
Operations: run Python or bash commands with real-time output streaming
"""
import asyncio
import logging
from typing import Any, Dict
logger = logging.getLogger(__name__)
EXECUTE_CODE_TOOL_SPEC = {
"name": "execute_code",
"description": (
"Execute Python or bash commands locally. This is YOUR PRIMARY TOOL for running code. "
"Use this to: write files, run Python scripts, install packages, process data, and execute any commands. "
"Examples: "
"- Write a file: echo 'print(1+1)' > test.py OR cat > test.py << 'EOF'\\nprint(1+1)\\nEOF; "
"- Run Python: python test.py; "
"- Install packages: pip install numpy sympy; "
"- List files: ls -la; "
"- Read files: cat file.txt"
),
"parameters": {
"type": "object",
"properties": {
"command": {
"type": "string",
"description": "The command to execute. Can be any bash or Python command. Examples: 'python script.py', 'pip install sympy', 'cat > file.py << \\'EOF\\'\\ncode\\nEOF'",
},
"timeout": {
"type": "number",
"description": "Maximum execution time in seconds (default: 3600 = 1 hour)",
},
},
"required": ["command"],
},
}
async def execute_code_handler(arguments: Dict[str, Any]) -> tuple[str, bool]:
"""Handler for execute_code tool - calls backend endpoint."""
from agent.utils.api_client import api_client
try:
command = arguments.get("command")
timeout = arguments.get("timeout", 3600)
if not command:
return "Error: 'command' is required", False
# Call the backend execute endpoint
result = await api_client.post(
"/execute",
json={"command": command, "timeout": timeout}
)
if result.get("success"):
output = result.get("output", "")
return f"✅ Execution completed\n{output}", True
else:
error = result.get("error", "Unknown error")
return f"❌ Execution failed: {error}", False
except Exception as e:
logger.error(f"Execute code error: {e}")
return f"Error: {str(e)}", False
|