_run_shell_tool() — langchain Function Reference

Architecture documentation for the _run_shell_tool() function in shell_tool.py from the langchain codebase.

Function python LangChainCore ApiManagement calls 5 called by 1

Entity Profile

LangChainCore→ ApiManagement→ _run_shell_tool() — langchain Function Reference

Dependency Diagram

graph TD
  8b644660_40e4_465a_3e51_e1a3166de4f9["_run_shell_tool()"]
  4e57a0ff_f1a5_e2e9_6949_569d7cd6d70a["ShellToolMiddleware"]
  8b644660_40e4_465a_3e51_e1a3166de4f9 -->|defined in| 4e57a0ff_f1a5_e2e9_6949_569d7cd6d70a
  bb8cdf05_0b11_e3e7_3c28_2a6ae5bfecb9["__init__()"]
  bb8cdf05_0b11_e3e7_3c28_2a6ae5bfecb9 -->|calls| 8b644660_40e4_465a_3e51_e1a3166de4f9
  c6011545_45bb_9aa6_7568_fa418072d163["_run_startup_commands()"]
  8b644660_40e4_465a_3e51_e1a3166de4f9 -->|calls| c6011545_45bb_9aa6_7568_fa418072d163
  b1223200_cf76_0d5b_1be6_44041f4fd7cf["_format_tool_message()"]
  8b644660_40e4_465a_3e51_e1a3166de4f9 -->|calls| b1223200_cf76_0d5b_1be6_44041f4fd7cf
  5160a5db_aeea_2d69_de2b_e5d653052a9d["_apply_redactions()"]
  8b644660_40e4_465a_3e51_e1a3166de4f9 -->|calls| 5160a5db_aeea_2d69_de2b_e5d653052a9d
  376fc508_1bee_23a7_7de6_267e13821247["restart()"]
  8b644660_40e4_465a_3e51_e1a3166de4f9 -->|calls| 376fc508_1bee_23a7_7de6_267e13821247
  da18dc22_e4fc_2b4a_2829_9217c9f5d8f5["execute()"]
  8b644660_40e4_465a_3e51_e1a3166de4f9 -->|calls| da18dc22_e4fc_2b4a_2829_9217c9f5d8f5
  style 8b644660_40e4_465a_3e51_e1a3166de4f9 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/shell_tool.py lines 761–854

    def _run_shell_tool(
        self,
        resources: _SessionResources,
        payload: dict[str, Any],
        *,
        tool_call_id: str | None,
    ) -> Any:
        session = resources.session

        if payload.get("restart"):
            LOGGER.info("Restarting shell session on request.")
            try:
                session.restart()
                self._run_startup_commands(session)
            except BaseException as err:
                LOGGER.exception("Restarting shell session failed; session remains unavailable.")
                msg = "Failed to restart shell session."
                raise ToolException(msg) from err
            message = "Shell session restarted."
            return self._format_tool_message(message, tool_call_id, status="success")

        command = payload.get("command")
        if not command or not isinstance(command, str):
            msg = "Shell tool expects a 'command' string when restart is not requested."
            raise ToolException(msg)

        LOGGER.info("Executing shell command: %s", command)
        result = session.execute(command, timeout=self._execution_policy.command_timeout)

        if result.timed_out:
            timeout_seconds = self._execution_policy.command_timeout
            message = f"Error: Command timed out after {timeout_seconds:.1f} seconds."
            return self._format_tool_message(
                message,
                tool_call_id,
                status="error",
                artifact={
                    "timed_out": True,
                    "exit_code": None,
                },
            )

        try:
            sanitized_output, matches = self._apply_redactions(result.output)
        except PIIDetectionError as error:
            LOGGER.warning("Blocking command output due to detected %s.", error.pii_type)
            message = f"Output blocked: detected {error.pii_type}."
            return self._format_tool_message(
                message,
                tool_call_id,
                status="error",
                artifact={
                    "timed_out": False,
                    "exit_code": result.exit_code,
                    "matches": {error.pii_type: error.matches},
                },
            )

        sanitized_output = sanitized_output or "<no output>"
        if result.truncated_by_lines:
            sanitized_output = (
                f"{sanitized_output.rstrip()}\n\n"
                f"... Output truncated at {self._execution_policy.max_output_lines} lines "
                f"(observed {result.total_lines})."
            )
        if result.truncated_by_bytes and self._execution_policy.max_output_bytes is not None:
            sanitized_output = (
                f"{sanitized_output.rstrip()}\n\n"
                f"... Output truncated at {self._execution_policy.max_output_bytes} bytes "
                f"(observed {result.total_bytes})."
            )

        if result.exit_code not in {0, None}:
            sanitized_output = f"{sanitized_output.rstrip()}\n\nExit code: {result.exit_code}"
            final_status: Literal["success", "error"] = "error"
        else:
            final_status = "success"

        artifact = {
            "timed_out": False,
            "exit_code": result.exit_code,