跳到主要内容

openhands.runtime.impl.eventstream.eventstream_runtime

LogBuffer Objects​

class LogBuffer()

Synchronous buffer for Docker container logs.

This class provides a thread-safe way to collect, store, and retrieve logs from a Docker container. It uses a list to store log lines and provides methods for appending, retrieving, and clearing logs.

stream_logs​

def stream_logs()

Stream logs from the Docker container in a separate thread.

This method runs in its own thread to handle the blocking operation of reading log lines from the Docker SDK's synchronous generator.

EventStreamRuntime Objects​

class EventStreamRuntime(Runtime)

This runtime will subscribe the event stream. When receive an event, it will send the event to runtime-client which run inside the docker environment.

Arguments:

  • config AppConfig - The application configuration.
  • event_stream EventStream - The event stream to subscribe to.
  • sid str, optional - The session ID. Defaults to 'default'.
  • plugins list[PluginRequirement] | None, optional - List of plugin requirements. Defaults to None.
  • env_vars dict[str, str] | None, optional - Environment variables to set. Defaults to None.

close​

def close(rm_all_containers: bool = True)

Closes the EventStreamRuntime and associated objects

Arguments:

  • rm_all_containers (bool): Whether to remove all containers with the 'openhands-sandbox-' prefix

list_files​

def list_files(path: str | None = None) -> list[str]

List files in the sandbox.

If path is None, list files in the sandbox's initial working directory (e.g., /workspace).

copy_from​

def copy_from(path: str) -> Path

Zip all files in the sandbox and return as a stream of bytes.