read_output

Instructions

Implementation Reference

• src/interactive_process_mcp/tools.py:42-54 (handler)

  The 'read_output' tool handler in the tools-based registration. It gets the session by ID, calls session.read_output with parameters (timeout, strip_ansi, max_lines), and returns output metadata.

  def read_output(args: dict) -> dict:
      session = _get_session(args["session_id"])
      output = session.read_output(
          timeout=args.get("timeout", 5.0),
          strip_ansi_flag=args.get("strip_ansi", True),
          max_lines=args.get("max_lines", 0),
      )
      return {
          "output": output,
          "has_more": False,
          "lines_returned": output.count("\n"),
          "bytes_returned": len(output.encode("utf-8")),
      }

• src/interactive_process_mcp/server.py:81-112 (handler)

  The 'read_output' tool handler in the FastMCP/server-based registration. Uses @mcp.tool() decorator with typed parameters (session_id, strip_ansi, timeout, max_lines) and returns dict with output, has_more, lines_returned, bytes_returned.

  @mcp.tool()
  def read_output(
      session_id: str,
      strip_ansi: bool = True,
      timeout: float = 5.0,
      max_lines: int = 0,
  ) -> dict:
      """Read new output from an interactive process since last read.
  
      If no new output is available, waits up to timeout seconds.
      Returns empty output on timeout (not an error).
  
      Args:
          session_id: The session ID returned by start_process.
          strip_ansi: Remove ANSI escape codes from output. Default True.
          timeout: Seconds to wait for new output. Default 5.
          max_lines: Max lines to return (0 = unlimited).
      """
      session = _mgr.get(session_id)
      if not session:
          return {"error": f"Session '{session_id}' not found"}
      output = session.read_output(
          timeout=timeout,
          strip_ansi_flag=strip_ansi,
          max_lines=max_lines,
      )
      return {
          "output": output,
          "has_more": False,
          "lines_returned": output.count("\n"),
          "bytes_returned": len(output.encode("utf-8")),
      }

• src/interactive_process_mcp/session.py:146-153 (handler)

  The Session.read_output() method — the core implementation that reads from the OutputBuffer, optionally strips ANSI codes, and limits to max_lines.

  def read_output(self, timeout: float = 5.0, strip_ansi_flag: bool = True, max_lines: int = 0) -> str:
      output = self._buffer.read_new(timeout=timeout)
      if strip_ansi_flag:
          output = strip_ansi(output)
      if max_lines > 0:
          lines = output.split("\n")
          output = "\n".join(lines[:max_lines])
      return output

• src/interactive_process_mcp/buffer.py:32-66 (helper)

  The OutputBuffer.read_new() method that implements the blocking read with timeout, returning unread buffered data since last read.

  def read_new(self, timeout: float = 0) -> str:
      deadline = time.monotonic() + timeout if timeout > 0 else 0
      while True:
          with self._lock:
              if self._read_pos < self._write_pos:
                  break
              if self._closed:
                  return ""
          remaining = deadline - time.monotonic() if deadline else 0
          if timeout > 0 and remaining <= 0:
              return ""
          if timeout > 0:
              self._new_data.wait(timeout=min(remaining, 0.1))
          else:
              return ""
      with self._lock:
          parts = []
          while self._chunks and self._read_pos < self._write_pos:
              parts.append(self._chunks.popleft())
              self._total_bytes -= len(parts[-1])
              self._read_pos += 1
          result = "".join(parts)
          self._stats["total_bytes_read"] += len(result)
          self._new_data.clear()
          return result
  
  def close(self) -> None:
      with self._lock:
          self._closed = True
      self._new_data.set()
  
  def stats(self) -> dict:
      with self._lock:
          return dict(self._stats)

• src/interactive_process_mcp/ansi.py:14-15 (helper)

  The strip_ansi() helper used by read_output to remove ANSI escape codes from output.

  def strip_ansi(text: str) -> str:
      return _ANSI_RE.sub('', text)

• src/interactive_process_mcp/tools.py:92-101 (registration)

  Registration of 'read_output' in the tools list returned by create_tools().

  return [
      ("start_process", start_process),
      ("send_input", send_input),
      ("read_output", read_output),
      ("send_and_read", send_and_read),
      ("list_sessions", list_sessions),
      ("terminate_process", terminate_process),
      ("resize_pty", resize_pty),
      ("get_session_info", get_session_info),
  ]

• src/interactive_process_mcp/server.py:81-112 (registration)

  Registration of 'read_output' via @mcp.tool() decorator in the FastMCP server.

  @mcp.tool()
  def read_output(
      session_id: str,
      strip_ansi: bool = True,
      timeout: float = 5.0,
      max_lines: int = 0,
  ) -> dict:
      """Read new output from an interactive process since last read.
  
      If no new output is available, waits up to timeout seconds.
      Returns empty output on timeout (not an error).
  
      Args:
          session_id: The session ID returned by start_process.
          strip_ansi: Remove ANSI escape codes from output. Default True.
          timeout: Seconds to wait for new output. Default 5.
          max_lines: Max lines to return (0 = unlimited).
      """
      session = _mgr.get(session_id)
      if not session:
          return {"error": f"Session '{session_id}' not found"}
      output = session.read_output(
          timeout=timeout,
          strip_ansi_flag=strip_ansi,
          max_lines=max_lines,
      )
      return {
          "output": output,
          "has_more": False,
          "lines_returned": output.count("\n"),
          "bytes_returned": len(output.encode("utf-8")),
      }