Download agent/tools/execute_code_tool.py from onewayto/water3: direct link, hf CLI and curl.
- Browser
- Download file 2.44 kB
-
https://huggingface.co/spaces/onewayto/water3/resolve/main/agent/tools/execute_code_tool.py
- Command line
-
hf download hf://spaces/onewayto/water3/agent/tools/execute_code_tool.py
-
curl -L -o execute_code_tool.py https://huggingface.co/spaces/onewayto/water3/resolve/main/agent/tools/execute_code_tool.py
2.44 kB
| """ | |
| 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 | |