HostAgent: Desktop Orchestrator

async def context_provision(self, context: Context) -> None:
    """
    Provide the context for the agent.
    :param context: The context for the agent.
    """
    await self._load_mcp_context(context)

def create_subagent(self, context: Optional["Context"] = None) -> None:
    """
    Orchestrate creation of the appropriate sub-agent.
    Decides between third-party agent and built-in app/operator agent.
    :param context: The context for the agent and session.
    """
    mode = RunningMode(context.get(ContextNames.MODE))

    assigned_third_party_agent = self.processor.processing_context.get_local(
        "assigned_third_party_agent"
    )
    # if self.processor.assigned_third_party_agent:
    if assigned_third_party_agent:
        config = AgentConfigResolver.resolve_third_party_config(
            assigned_third_party_agent, mode
        )
    else:
        window_name = context.get(ContextNames.APPLICATION_PROCESS_NAME)
        root_name = context.get(ContextNames.APPLICATION_ROOT_NAME)

        if mode in {
            RunningMode.NORMAL,
            RunningMode.BATCH_NORMAL,
            RunningMode.FOLLOWER,
        }:
            config = AgentConfigResolver.resolve_app_agent_config(
                root_name, window_name, mode
            )
        elif mode in {RunningMode.NORMAL_OPERATOR, RunningMode.BATCH_OPERATOR}:
            config = AgentConfigResolver.resolve_operator_agent_config(
                root_name, window_name, mode
            )
        else:
            raise ValueError(f"Unsupported mode: {mode}")

    agent_name = config.get("name")
    agent_type = config.get("agent_type")
    process_name = config.get("process_name")

    self.logger.info(f"Creating sub agent with config: {config}")

    app_agent = self.agent_factory.create_agent(**config)
    self.appagent_dict[agent_name] = app_agent
    app_agent.host = self
    self._active_appagent = app_agent

    self.logger.info(
        f"Created sub agent: {agent_name} with type {agent_type} and process name {process_name}, class {app_agent.__class__.__name__}"
    )

    return app_agent

def get_active_appagent(self) -> AppAgent:
    """
    Get the active app agent.
    :return: The active app agent.
    """
    return self._active_appagent

def get_prompter(
    self,
    is_visual: bool,
    main_prompt: str,
    example_prompt: str,
    api_prompt: str,
) -> HostAgentPrompter:
    """
    Get the prompt for the agent.
    :param is_visual: The flag indicating whether the agent is visual or not.
    :param main_prompt: The main prompt file path.
    :param example_prompt: The example prompt file path.
    :param api_prompt: The API prompt file path.
    :return: The prompter instance.
    """
    return HostAgentPrompter(is_visual, main_prompt, example_prompt, api_prompt)

def message_constructor(
    self,
    image_list: List[str],
    os_info: str,
    plan: List[str],
    prev_subtask: List[Dict[str, str]],
    request: str,
    blackboard_prompt: List[Dict[str, str]],
) -> List[Dict[str, Union[str, List[Dict[str, str]]]]]:
    """
    Construct the message.
    :param image_list: The list of screenshot images.
    :param os_info: The OS information.
    :param prev_subtask: The previous subtask.
    :param plan: The plan.
    :param request: The request.
    :return: The message.
    """
    hostagent_prompt_system_message = self.prompter.system_prompt_construction()
    hostagent_prompt_user_message = self.prompter.user_content_construction(
        image_list=image_list,
        control_item=os_info,
        prev_subtask=prev_subtask,
        prev_plan=plan,
        user_request=request,
    )

    if blackboard_prompt:
        hostagent_prompt_user_message = (
            blackboard_prompt + hostagent_prompt_user_message
        )

    hostagent_prompt_message = self.prompter.prompt_construction(
        hostagent_prompt_system_message, hostagent_prompt_user_message
    )

    return hostagent_prompt_message

def print_response(self, response: HostAgentResponse) -> None:
    """
    Print the response using the presenter.
    :param response: The response object to print.
    """
    # Format the action string using get_command_string and pass to presenter
    function = response.function
    arguments = response.arguments

    action_str = None
    if function:
        action_str = self.get_command_string(function, arguments)

    # Pass formatted action string as parameter instead of modifying response
    self.presenter.present_host_agent_response(response, action_str=action_str)

async def process(self, context: Context) -> None:
    """
    Process the agent.
    :param context: The context.
    """
    # from ufo.agents.processors.host_agent_processor import HostAgentProcessor

    if not self._context_provision_executed:
        await self.context_provision(context=context)
        self._context_provision_executed = True
    self.processor = HostAgentProcessor(agent=self, global_context=context)
    # self.processor = HostAgentProcessor(agent=self, context=context)
    await self.processor.process()

    # Sync the status with the processor.
    # self.status = self.processor.status
    self.status = self.processor.processing_context.get_local("status")
    self.logger.info(f"Host agent status updated to: {self.status}")

def process_confirmation(self) -> None:
    """
    TODO: Process the confirmation.
    """
    pass