| | """
|
| | 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
|
| |
|
| |
|
| | 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
|
| |
|
