executor_events.py
"""
Executor Events
===============
Demonstrates filtering internal executor events during streamed workflow runs.
"""
from agno.agent import Agent
from agno.models.openai import OpenAIChat
from agno.workflow.step import Step
from agno.workflow.workflow import Workflow
# ---------------------------------------------------------------------------
# Create Agent
# ---------------------------------------------------------------------------
agent = Agent(
name="ResearchAgent",
model=OpenAIChat(id="gpt-4o"),
instructions="You are a helpful research assistant. Be concise.",
)
# ---------------------------------------------------------------------------
# Create Workflow
# ---------------------------------------------------------------------------
workflow = Workflow(
name="Research Workflow",
steps=[Step(name="Research", agent=agent)],
stream=True,
stream_executor_events=False,
)
# ---------------------------------------------------------------------------
# Run Workflow
# ---------------------------------------------------------------------------
def main() -> None:
print("\n" + "=" * 70)
print("Workflow Streaming Example: stream_executor_events=False")
print("=" * 70)
print(
"\nThis will show only workflow and step events and will not yield RunContent and TeamRunContent events"
)
print("Filtering out internal agent/team events for cleaner output.\n")
for event in workflow.run(
"What is Python?",
stream=True,
stream_events=True,
):
event_name = event.event if hasattr(event, "event") else type(event).__name__
print(f" -> {event_name}")
if __name__ == "__main__":
main()
Run the Example
1
Set up your virtual environment
uv venv --python 3.12
source .venv/bin/activate
uv venv --python 3.12
.venv\Scripts\activate
2
Install dependencies
uv pip install -U agno fastapi openai
3
Export your OpenAI API key
export OPENAI_API_KEY="your_openai_api_key_here"
$Env:OPENAI_API_KEY="your_openai_api_key_here"
4
Run the example
Save the code above as
executor_events.py, then run:python executor_events.py