run_command

Instructions

Implementation Reference

• src/iterm2_mcp/server.py:224-273 (handler)

  The run_command tool handler: sends a command line to an iTerm2 session, waits for COMMAND_END via PromptMonitor with a configurable timeout, then returns the screen contents with a reason prefix.

  @mcp.tool()
  async def run_command(
      command: str,
      session_id: str | None = None,
      timeout_seconds: float = 30.0,
  ) -> str:
      """Run a command and wait for it to finish, returning the resulting screen.
  
      Completion is detected via iTerm2's ``PromptMonitor`` ``COMMAND_END``
      notification, which requires iTerm2 shell integration to be installed in
      the target shell. If the prompt-end event does not arrive within
      ``timeout_seconds``, the current screen is returned with a ``[timeout]``
      marker.
  
      :param command: The command line to run.
      :param session_id: Target session UUID. Defaults to the active session.
      :param timeout_seconds: How long to wait for ``COMMAND_END`` before
          giving up.
      """
      sess = await _session(session_id)
      conn = await _connection_singleton()
  
      try:
          async with iterm2.PromptMonitor(
              conn,
              sess.session_id,
              modes=[iterm2.PromptMonitor.Mode.COMMAND_END],
          ) as mon:
              # Subscribe before sending the command to avoid missing fast completions.
              await sess.async_send_text(command + "\n", suppress_broadcast=True)
  
              try:
                  mode, info = await asyncio.wait_for(mon.async_get(), timeout=timeout_seconds)
                  if mode == iterm2.PromptMonitor.Mode.COMMAND_END:
                      # For COMMAND_END, ``info`` is the exit-status int directly;
                      # see PromptMonitor.async_get docstring in the iterm2 package.
                      reason = f"prompt-end (exit {info})"
                  else:
                      reason = f"unexpected monitor mode {mode}"
              except (TimeoutError, asyncio.TimeoutError):
                  reason = f"timeout after {timeout_seconds}s"
      except (TimeoutError, asyncio.TimeoutError):
          # Catches timeouts from PromptMonitor setup itself (distinct from the
          # inner wait_for timeout which covers the command execution).
          reason = f"timeout after {timeout_seconds}s"
      except Exception as exc:  # noqa: BLE001 — surface as reason, don't crash the tool
          reason = f"prompt-monitor error ({exc!r}); is shell integration installed?"
  
      contents = await sess.async_get_screen_contents()
      return f"[{reason}]\n{_screen_text(contents)}"

• src/iterm2_mcp/server.py:224-224 (registration)

  Registration of run_command as an MCP tool via the @mcp.tool() decorator on line 224.

  @mcp.tool()

• src/iterm2_mcp/server.py:225-229 (schema)

  Input schema: command (str, required), session_id (str, optional), timeout_seconds (float, default 30.0). Returns str.

  async def run_command(
      command: str,
      session_id: str | None = None,
      timeout_seconds: float = 30.0,
  ) -> str:

• src/iterm2_mcp/server.py:72-77 (helper)

  Helper function _screen_text used by run_command to convert screen contents to plain text.

  def _screen_text(contents: iterm2.ScreenContents) -> str:
      """Flatten a ``ScreenContents`` into plain text, stripping trailing blank lines."""
      lines = [contents.line(i).string for i in range(contents.number_of_lines)]
      while lines and not lines[-1].strip():
          lines.pop()
      return "\n".join(lines)

• src/iterm2_mcp/server.py:48-69 (helper)

  Helper function _session used by run_command to resolve a session ID or fall back to the active session.

  async def _session(session_id: str | None) -> iterm2.Session:
      """Resolve a session by ID, falling back to the currently active session.
  
      :param session_id: A specific session UUID to target, or ``None`` to use the
          currently focused window/tab/pane.
      """
      app = await _app()
      if session_id:
          sess = app.get_session_by_id(session_id)
          if sess is None:
              raise ValueError(f"No session found with ID {session_id!r}")
          return sess
      win = app.current_terminal_window
      if win is None:
          raise RuntimeError("No active iTerm2 window.")
      tab = win.current_tab
      if tab is None:
          raise RuntimeError("No active tab in the current window.")
      sess = tab.current_session
      if sess is None:
          raise RuntimeError("No active session in the current tab.")
      return sess