start() — langchain Function Reference

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

Function python LangChainCore ApiManagement called by 2

Entity Profile

LangChainCore→ ApiManagement→ start() — langchain Function Reference

Dependency Diagram

graph TD
  a66768b0_c906_c61b_b24b_37fd445b1af9["start()"]
  cf4eeaa4_4bcc_69d9_a0f3_664b1eecb4e8["ShellSession"]
  a66768b0_c906_c61b_b24b_37fd445b1af9 -->|defined in| cf4eeaa4_4bcc_69d9_a0f3_664b1eecb4e8
  376fc508_1bee_23a7_7de6_267e13821247["restart()"]
  376fc508_1bee_23a7_7de6_267e13821247 -->|calls| a66768b0_c906_c61b_b24b_37fd445b1af9
  3d2455b7_eaeb_053d_fc17_21aaa32050ca["_create_resources()"]
  3d2455b7_eaeb_053d_fc17_21aaa32050ca -->|calls| a66768b0_c906_c61b_b24b_37fd445b1af9
  style a66768b0_c906_c61b_b24b_37fd445b1af9 fill:#6366f1,stroke:#818cf8,color:#fff

Relationship Graph

Source Code

libs/langchain_v1/langchain/agents/middleware/shell_tool.py lines 147–184

    def start(self) -> None:
        """Start the shell subprocess and reader threads.

        Raises:
            RuntimeError: If the shell session pipes cannot be initialized.
        """
        if self._process and self._process.poll() is None:
            return

        self._process = self._policy.spawn(
            workspace=self._workspace,
            env=self._environment,
            command=self._command,
        )
        if (
            self._process.stdin is None
            or self._process.stdout is None
            or self._process.stderr is None
        ):
            msg = "Failed to initialize shell session pipes."
            raise RuntimeError(msg)

        self._stdin = self._process.stdin
        self._terminated = False
        self._queue = queue.Queue()

        self._stdout_thread = threading.Thread(
            target=self._enqueue_stream,
            args=(self._process.stdout, "stdout"),
            daemon=True,
        )
        self._stderr_thread = threading.Thread(
            target=self._enqueue_stream,
            args=(self._process.stderr, "stderr"),
            daemon=True,
        )
        self._stdout_thread.start()
        self._stderr_thread.start()