Execute a command in the bash shell.
(self, command: str)
| 45 | self._process.terminate() |
| 46 | |
| 47 | async def run(self, command: str): |
| 48 | """Execute a command in the bash shell.""" |
| 49 | if not self._started: |
| 50 | raise ToolError("Session has not started.") |
| 51 | if self._process.returncode is not None: |
| 52 | return ToolResult( |
| 53 | system="tool must be restarted", |
| 54 | error=f"bash has exited with returncode {self._process.returncode}", |
| 55 | ) |
| 56 | if self._timed_out: |
| 57 | raise ToolError( |
| 58 | f"timed out: bash has not returned in {self._timeout} seconds and must be restarted", |
| 59 | ) |
| 60 | |
| 61 | # we know these are not None because we created the process with PIPEs |
| 62 | assert self._process.stdin |
| 63 | assert self._process.stdout |
| 64 | assert self._process.stderr |
| 65 | |
| 66 | # send command to the process |
| 67 | self._process.stdin.write( |
| 68 | command.encode() + f"; echo '{self._sentinel}'\n".encode() |
| 69 | ) |
| 70 | await self._process.stdin.drain() |
| 71 | |
| 72 | # read output from the process, until the sentinel is found |
| 73 | output = "" |
| 74 | try: |
| 75 | async with asyncio.timeout(self._timeout): |
| 76 | while True: |
| 77 | await asyncio.sleep(self._output_delay) |
| 78 | data = await self._process.stdout.readline() |
| 79 | if not data: |
| 80 | break |
| 81 | line = data.decode() |
| 82 | output += line |
| 83 | if self._sentinel in line: |
| 84 | output = output.replace(self._sentinel, "") |
| 85 | break |
| 86 | except asyncio.TimeoutError: |
| 87 | self._timed_out = True |
| 88 | raise ToolError( |
| 89 | f"timed out: bash has not returned in {self._timeout} seconds and must be restarted", |
| 90 | ) from None |
| 91 | |
| 92 | error = await self._process.stderr.read() |
| 93 | error = error.decode() |
| 94 | |
| 95 | return CLIResult(output=output.strip(), error=error.strip()) |
| 96 | |
| 97 | |
| 98 | class BashTool(BaseAnthropicTool): |
no test coverage detected