Function on_execution_request

examples/xpander/coding_agent.py:119–156

(execution_task: AgentExecution)

# === Define Execution Handler ===
async def on_execution_request(execution_task: AgentExecution) -> AgentExecutionResult:
    logger.info(f"[on_execution_request] Called with execution_task: {execution_task}")
    my_agent = MyAgent()
    logger.info(f"[on_execution_request] Instantiated MyAgent: {my_agent}")

    user_info = ""
    user = getattr(execution_task.input, "user", None)

    if user:
        name = f"{user.first_name} {user.last_name}".strip()
        email = getattr(user, "email", "")
        user_info = f"👤 From user: {name}\n📧 Email: {email}"

    IncomingEvent = f"\n📨 Incoming message: {execution_task.input.text}\n{user_info}"

    logger.info(f"[on_execution_request] IncomingEvent: {IncomingEvent}")
    logger.info(f"[on_execution_request] Calling agent_backend.init_task with execution={execution_task.model_dump()}")
    my_agent.agent_backend.init_task(execution=execution_task.model_dump())

    # extract just the text input for quick start purpose. for more robust use the object
    user_txt_input = execution_task.input.text
    logger.info(f"[on_execution_request] Running agent with user_txt_input: {user_txt_input}")
    try:
        await my_agent.run(user_txt_input)
        logger.info("[on_execution_request] Agent run completed")
        execution_result = my_agent.agent_backend.retrieve_execution_result()
        logger.info(f"[on_execution_request] Execution result: {execution_result}")
        result_obj = AgentExecutionResult(
            result=execution_result.result,
            is_success=execution_result.status == ExecutionStatus.COMPLETED,
        )
        logger.info(f"[on_execution_request] Returning AgentExecutionResult: {result_obj}")
        return result_obj
    except Exception as e:
        logger.error(f"[on_execution_request] Exception: {e}")
        raise
    finally:
        logger.info("[on_execution_request] Exiting handler")


# === Register Callback ===