Python: Executor with decorator doesn't generate `AgentRunUpdateEvent` event

[a1exwang]

For my workflow, if a workflow node is created with the decorator syntax, it's missing the AgentRunUpdateEvent. But if a node is a plain agent, it will correctly generate AgentRunUpdateEvent.

Repro steps

1. Run the app code
2. Get the result

WorkflowStartedEvent(origin=WorkflowEventSource.FRAMEWORK, data=None)
WorkflowStatusEvent(state=WorkflowRunState.IN_PROGRESS, data=None, origin=WorkflowEventSource.FRAMEWORK)
ExecutorInvokedEvent(executor_id=ContentWriter, data=None)
AgentRunUpdateEvent(executor_id=ContentWriter, messages=Thank)
AgentRunUpdateEvent(executor_id=ContentWriter, messages= you)
...
ExecutorCompletedEvent(executor_id=ContentWriter, data=None)
ExecutorInvokedEvent(executor_id=CodeWriter, data=None)   # <---- HERE, the CodeWriter executor only has executor level events, no agent events.
ExecutorCompletedEvent(executor_id=CodeWriter, data=None)
WorkflowStatusEvent(state=WorkflowRunState.IDLE, data=None, origin=WorkflowEventSource.FRAMEWORK)

Expected behavior

For the latter syntax can also generate AgentRunUpdateEvent which contains the output text.

My app code

import asyncio
import os
from typing import Any, Never

from agent_framework import (
    AgentExecutorResponse,
    WorkflowBuilder,
    Executor, WorkflowContext, handler, ChatAgent, ChatMessage,
    executor,
)
from agent_framework.azure import AzureOpenAIChatClient
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()

async def main():
    chat_client = AzureOpenAIChatClient(api_key=os.environ["AZURE_OPENAI_API_KEY"])
    writer_agent = chat_client.create_agent(
        name="ContentWriter",
        instructions=("You are an excellent content writer. "),
    )
    class CodeWriter(Executor):
        agent: ChatAgent
        def __init__(self, chat_client: AzureOpenAIChatClient):
            self.agent = chat_client.create_agent(instructions=("You are an excellent programmer. Rewrite your input with Python code."))
            super().__init__(id="CodeWriter")

        @handler
        async def handle(self, last_response: AgentExecutorResponse, ctx: WorkflowContext[list[ChatMessage]]) -> None:
            response = await self.agent.run(last_response.agent_run_response.text)
            await ctx.send_message(response)

    writer2 = CodeWriter(chat_client=chat_client)

    workflow = (
        WorkflowBuilder(
            name="Test Workflow",
            description="Test Workflow",
        )
        .set_start_executor(writer_agent)
        .add_edge(writer_agent, writer2)
        .build()
    )

    async for event in workflow.run_stream([ChatMessage(role="user", content="Hello world")]):
        print(event)

if __name__ == "__main__":
    asyncio.run(main())

package version:

> uv pip freeze | grep agent-framework
agent-framework-azure-ai==1.0.0b251007
agent-framework-core==1.0.0b251007

[moonbox3]

@a1exwang when writing a custom executor that manages an agent instance, it is the dev's responsibility to emit any events. If you simply pass in an agent as an edge, we wrap it in an AgentExecutor on your behalf, and then emit events as produced by the agent run. For your current code:

response = await self.agent.run(...)
    await ctx.add_event(AgentRunEvent(self.id, response))

In your code, you're getting the full response from the agent.run(...) and then passing that response as a message to a non-existent edge. If writer2 is indeed your terminal node, then you'd want to call ctx.yield_output(response and annotate ctx as WorkflowContext[Never, list[ChatMessage]] if calling response.messages otherwise WorkflowContext[Never, AgentRunResponse] if using your current code.

To handle it streaming, you'd want:

async for update in self.agent.run_stream(...):
    await ctx.add_event(AgentRunUpdateEvent(self.id, update))

[a1exwang]

Thanks for the clarification. The raw problem I'd like to solve find a possible way to get each executor's input / output (I mean the executor handler's argument and data sent by ctx.send_message) after or while running via any means without modifying executor code, given I have a workflow object already constructed. This is due to the architecture of our app.

For agent executor, I can use the events returned from workflow.run_stream. But is there a generic way for all executor types. @moonbox3

[moonbox3]

Thanks for the added context. I see that in dotnet, we're including the input messages as part of ExecutorInvokedEvent.data and the messages sent via ExecutorCompletedEvent.data.

I've included this update to do the same in Python in #2591. I also added a sample showing the behavior to get the I/O types.

[a1exwang]

Thanks for the added context. I see that in dotnet, we're including the input messages as part of ExecutorInvokedEvent.data and the messages sent via ExecutorCompletedEvent.data.

I've included this update to do the same in Python in #2591. I also added a sample showing the behavior to get the I/O types.

Thanks, this helps a lot!