execute_shell
Execute shell commands to navigate directories, list contents, and run scripts for financial data processing and analysis workflows.
Instructions
A tool capable of executing shell commands can use pwd to check the current location, cd to navigate to a new directory, ls to view the contents of a directory, and execute scripts.
Note that the starting directory is always the same each time the tool is invoked. If you need to perform multiple operations within a specific directory, you must include the full path in each command, for example: cd aa/bb && bash xxx.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| command | Yes | command to be executed |
Implementation Reference
- Executes the shell command asynchronously using the run_shell_command helper, captures stdout, stderr, and exit code, then formats and sets the output.
async def async_execute(self): """Execute the shell command operation.""" command: str = self.input_dict.get("command", "").strip() assert command, "The 'command' parameter cannot be empty." # Execute using run_shell_command from common_utils stdout, stderr, return_code = await run_shell_command(command) # Build result message result_parts = [ f"Command: {command}", f"Output: {stdout if stdout else '(empty)'}", f"Error: {stderr if stderr else '(none)'}", f"Exit Code: {return_code if return_code is not None else '(none)'}", ] self.set_output("\n".join(result_parts)) - Defines the tool schema including name 'ExecuteShell', description, and required 'command' string input.
def build_tool_call(self) -> ToolCall: """Build the tool call schema for executing shell commands.""" return ToolCall( **{ "name": "ExecuteShell", "description": self.get_prompt("tool_description"), "input_schema": { "command": { "type": "string", "description": "command to be executed", "required": True, }, }, }, ) - finance_mcp/core/gallery/execute_shell_op.py:15-16 (registration)Registers the ExecuteShellOp class as an operation using the @C.register_op() decorator.
@C.register_op() class ExecuteShellOp(BaseAsyncToolOp): - Helper function that runs shell commands using asyncio.create_subprocess_shell, handles timeout, and returns decoded stdout, stderr, and return code.
async def run_shell_command(cmd: str, timeout: Optional[float] = 30) -> Tuple[str, str, int]: """Run a shell command asynchronously and return its output. Args: cmd: Full command string to execute in a system shell. timeout: Maximum time in seconds to wait for completion. ``None`` disables the timeout and waits indefinitely. Returns: A tuple ``(stdout, stderr, return_code)`` where both streams are UTF-8 decoded strings. Raises: asyncio.TimeoutError: If command execution exceeds ``timeout``. """ process = await asyncio.create_subprocess_shell( cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) if timeout: stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout) else: stdout, stderr = await process.communicate() return ( stdout.decode("utf-8", errors="ignore"), stderr.decode("utf-8", errors="ignore"), process.returncode, )