openhands.core.main

run_controller

async def run_controller(config: AppConfig,
                         initial_user_action: Action,
                         sid: str | None = None,
                         runtime: Runtime | None = None,
                         agent: Agent | None = None,
                         exit_on_message: bool = False,
                         fake_user_response_fn: FakeUserResponseFunc
                         | None = None,
                         headless_mode: bool = True) -> State | None

Main coroutine to run the agent controller with task input flexibility.

It's only used when you launch openhands backend directly via cmdline.

Arguments:

• config - The app config.
• initial_user_action - An Action object containing initial user input
• sid - (optional) The session id. IMPORTANT: please don't set this unless you know what you're doing. Set it to incompatible value will cause unexpected behavior on RemoteRuntime.
• runtime - (optional) A runtime for the agent to run on.
• agent - (optional) A agent to run.
• exit_on_message - quit if agent asks for a message from user (optional)
• fake_user_response_fn - An optional function that receives the current state (could be None) and returns a fake user response.
• headless_mode - Whether the agent is run in headless mode.

Returns:

The final state of the agent, or None if an error occurred.

Raises:

• AssertionError - If initial_user_action is not an Action instance.
• Exception - Various exceptions may be raised during execution and will be logged.

Notes:

• State persistence: If config.file_store is set, the agent's state will be saved between sessions.
• Trajectories: If config.trajectories_path is set, execution history will be saved as JSON for analysis.
• Budget control: Execution is limited by config.max_iterations and config.max_budget_per_task.

Example:

>>> config = load_app_config() >>> action = MessageAction(content="Write a hello world program") >>> state = await run_controller(config=config, initial_user_action=action)

auto_continue_response

def auto_continue_response(
        state: State,
        encapsulate_solution: bool = False,
        try_parse: Callable[[Action | None], str] | None = None) -> str

Default function to generate user responses. Tell the agent to proceed without asking for more input, or finish the interaction.

load_replay_log

def load_replay_log(trajectory_path: str) -> tuple[list[Event] | None, Action]

Load trajectory from given path, serialize it to a list of events, and return two things:

1. A list of events except the first action
2. First action (user message, a.k.a. initial task)