execute_shell
Execute shell commands to navigate directories, list contents, and run scripts for financial research workflows, enabling structured data extraction and analysis.
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
| Name | Required | Description | Default |
|---|---|---|---|
| command | Yes | command to be executed |
Input Schema (JSON Schema)
{
"properties": {
"command": {
"description": "command to be executed",
"type": "string"
}
},
"required": [
"command"
],
"type": "object"
}
Implementation Reference
- The handler function that performs the core logic of executing the shell command via run_shell_command, capturing stdout/stderr/exit code, and formatting 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)) async def async_default_execute(self, e: Exception = None, **_kwargs):
- Defines the tool schema including name 'ExecuteShell', description from prompt, and input schema requiring a 'command' string.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, which likely exposes it as the 'execute_shell' tool.@C.register_op() class ExecuteShellOp(BaseAsyncToolOp):
- Helper utility that executes the shell command using asyncio.create_subprocess_shell, handles timeout, decodes output streams, and returns stdout, stderr, returncode.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, )