[BREAKING] Python: Refactor workflow events to unified discriminated union pattern (#3690)

* Refactor events

* Merge main

* Fixes

* Cleanup

* Update samples and tests

* Remove unused imports

* PR feedback

* Merge main. Add properties for events to help typing

* Formatting

* Cleanup

* use builtins.type to avoid shadowing by WorkflowEvent.type attribute

* Final improvements

python/packages/core/tests/workflow/test_agent_executor.py

@@ -13,9 +13,7 @@ from agent_framework import (
     ChatMessageStore,
     Content,
     ResponseStream,
-    WorkflowOutputEvent,
     WorkflowRunState,
-    WorkflowStatusEvent,
 )
 from agent_framework._workflows._agent_executor import AgentExecutorResponse
 from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
@@ -77,9 +75,9 @@ async def test_agent_executor_checkpoint_stores_and_restores_state() -> None:
     # Run the workflow with a user message
     first_run_output: AgentExecutorResponse | None = None
     async for ev in wf.run("First workflow run", stream=True):
-        if isinstance(ev, WorkflowOutputEvent):
+        if ev.type == "output":
             first_run_output = ev.data  # type: ignore[assignment]
-        if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
+        if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
             break
 
     assert first_run_output is not None
@@ -131,9 +129,9 @@ async def test_agent_executor_checkpoint_stores_and_restores_state() -> None:
     # Resume from checkpoint
     resumed_output: AgentExecutorResponse | None = None
     async for ev in wf_resume.run(checkpoint_id=restore_checkpoint.checkpoint_id, stream=True):
-        if isinstance(ev, WorkflowOutputEvent):
+        if ev.type == "output":
             resumed_output = ev.data  # type: ignore[assignment]
-        if isinstance(ev, WorkflowStatusEvent) and ev.state in (
+        if ev.type == "status" and ev.state in (
             WorkflowRunState.IDLE,
             WorkflowRunState.IDLE_WITH_PENDING_REQUESTS,
         ):

python/packages/core/tests/workflow/test_agent_executor_tool_calls.py

@@ -19,11 +19,10 @@ from agent_framework import (
     ChatResponse,
     ChatResponseUpdate,
     Content,
-    RequestInfoEvent,
     ResponseStream,
     WorkflowBuilder,
     WorkflowContext,
-    WorkflowOutputEvent,
+    WorkflowEvent,
     executor,
     tool,
 )
@@ -100,9 +99,9 @@ async def test_agent_executor_emits_tool_calls_in_streaming_mode() -> None:
     workflow = WorkflowBuilder().set_start_executor(agent_exec).build()
 
     # Act: run in streaming mode
-    events: list[WorkflowOutputEvent] = []
+    events: list[WorkflowEvent[AgentResponseUpdate]] = []
     async for event in workflow.run("What's the weather?", stream=True):
-        if isinstance(event, WorkflowOutputEvent):
+        if event.type == "output" and isinstance(event.data, AgentResponseUpdate):
             events.append(event)
 
     # Assert: we should receive 4 events (text, function call, function result, text)
@@ -290,9 +289,9 @@ async def test_agent_executor_tool_call_with_approval_streaming() -> None:
     workflow = WorkflowBuilder().set_start_executor(agent).add_edge(agent, test_executor).build()
 
     # Act
-    request_info_events: list[RequestInfoEvent] = []
+    request_info_events: list[WorkflowEvent] = []
     async for event in workflow.run("Invoke tool requiring approval", stream=True):
-        if isinstance(event, RequestInfoEvent):
+        if event.type == "request_info":
             request_info_events.append(event)
 
     # Assert
@@ -307,7 +306,7 @@ async def test_agent_executor_tool_call_with_approval_streaming() -> None:
     async for event in workflow.send_responses_streaming({
         approval_request.request_id: approval_request.data.to_function_approval_response(True)
     }):
-        if isinstance(event, WorkflowOutputEvent):
+        if event.type == "output":
             output = event.data
 
     # Assert
@@ -367,9 +366,9 @@ async def test_agent_executor_parallel_tool_call_with_approval_streaming() -> No
     workflow = WorkflowBuilder().set_start_executor(agent).add_edge(agent, test_executor).build()
 
     # Act
-    request_info_events: list[RequestInfoEvent] = []
+    request_info_events: list[WorkflowEvent] = []
     async for event in workflow.run("Invoke tool requiring approval", stream=True):
-        if isinstance(event, RequestInfoEvent):
+        if event.type == "request_info":
             request_info_events.append(event)
 
     # Assert
@@ -387,7 +386,7 @@ async def test_agent_executor_parallel_tool_call_with_approval_streaming() -> No
 
     output: str | None = None
     async for event in workflow.send_responses_streaming(responses):
-        if isinstance(event, WorkflowOutputEvent):
+        if event.type == "output":
             output = event.data
 
     # Assert

python/packages/core/tests/workflow/test_agent_run_event_typing.py

@@ -1,27 +1,38 @@
 # Copyright (c) Microsoft. All rights reserved.
 
-"""Tests for agent run event typing."""
+"""Tests for WorkflowEvent[T] generic type annotations."""
 
 from agent_framework import AgentResponse, AgentResponseUpdate, ChatMessage
-from agent_framework._workflows._events import WorkflowOutputEvent
+from agent_framework._workflows._events import WorkflowEvent
 
 
-def test_agent_run_event_data_type() -> None:
-    """Verify WorkflowOutputEvent.data is typed as AgentResponse | None."""
+def test_workflow_event_with_agent_response_data_type() -> None:
+    """Verify WorkflowEvent[AgentResponse].data is typed as AgentResponse."""
     response = AgentResponse(messages=[ChatMessage(role="assistant", text="Hello")])
-    event = WorkflowOutputEvent(data=response, executor_id="test")
+    event: WorkflowEvent[AgentResponse] = WorkflowEvent.emit(executor_id="test", data=response)
 
     # This assignment should pass type checking without a cast
-    data: AgentResponse | None = event.data
+    data: AgentResponse = event.data
     assert data is not None
     assert data.text == "Hello"
 
 
-def test_agent_run_update_event_data_type() -> None:
-    """Verify WorkflowOutputEvent.data is typed as AgentResponseUpdate | None."""
+def test_workflow_event_with_agent_response_update_data_type() -> None:
+    """Verify WorkflowEvent[AgentResponseUpdate].data is typed as AgentResponseUpdate."""
     update = AgentResponseUpdate()
-    event = WorkflowOutputEvent(data=update, executor_id="test")
+    event: WorkflowEvent[AgentResponseUpdate] = WorkflowEvent.emit(executor_id="test", data=update)
 
     # This assignment should pass type checking without a cast
-    data: AgentResponseUpdate | None = event.data
+    data: AgentResponseUpdate = event.data
     assert data is not None
+
+
+def test_workflow_event_repr() -> None:
+    """Verify WorkflowEvent.__repr__ uses consistent format."""
+    response = AgentResponse(messages=[ChatMessage(role="assistant", text="Hello")])
+    event: WorkflowEvent[AgentResponse] = WorkflowEvent.emit(executor_id="test", data=response)
+
+    repr_str = repr(event)
+    assert "WorkflowEvent" in repr_str
+    assert "executor_id='test'" in repr_str
+    assert "data=" in repr_str

python/packages/core/tests/workflow/test_checkpoint_validation.py

@@ -8,7 +8,6 @@ from agent_framework import (
     WorkflowCheckpointException,
     WorkflowContext,
     WorkflowRunState,
-    WorkflowStatusEvent,
     handler,
 )
 from agent_framework._workflows._checkpoint import InMemoryCheckpointStorage
@@ -80,4 +79,4 @@ async def test_resume_succeeds_when_graph_matches() -> None:
         )
     ]
 
-    assert any(isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE for event in events)
+    assert any(event.type == "status" and event.state == WorkflowRunState.IDLE for event in events)

python/packages/core/tests/workflow/test_executor.py

@@ -8,11 +8,10 @@ from typing_extensions import Never
 from agent_framework import (
     ChatMessage,
     Executor,
-    ExecutorCompletedEvent,
-    ExecutorInvokedEvent,
     Message,
     WorkflowBuilder,
     WorkflowContext,
+    WorkflowEvent,
     executor,
     handler,
     response_handler,
@@ -139,7 +138,7 @@ def test_executor_handlers_with_output_types():
 
 
 async def test_executor_invoked_event_contains_input_data():
-    """Test that ExecutorInvokedEvent contains the input message data."""
+    """Test that executor_invoked event (type='executor_invoked') contains the input message data."""
 
     class UpperCaseExecutor(Executor):
         @handler
@@ -157,7 +156,7 @@ async def test_executor_invoked_event_contains_input_data():
     workflow = WorkflowBuilder().add_edge(upper, collector).set_start_executor(upper).build()
 
     events = await workflow.run("hello world")
-    invoked_events = [e for e in events if isinstance(e, ExecutorInvokedEvent)]
+    invoked_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "executor_invoked"]
 
     assert len(invoked_events) == 2
 
@@ -171,7 +170,7 @@ async def test_executor_invoked_event_contains_input_data():
 
 
 async def test_executor_completed_event_contains_sent_messages():
-    """Test that ExecutorCompletedEvent contains the messages sent via ctx.send_message()."""
+    """Test that event (type='executor_completed') contains the messages sent via ctx.send_message()."""
 
     class MultiSenderExecutor(Executor):
         @handler
@@ -194,7 +193,7 @@ async def test_executor_completed_event_contains_sent_messages():
     workflow = WorkflowBuilder().add_edge(sender, collector).set_start_executor(sender).build()
 
     events = await workflow.run("hello")
-    completed_events = [e for e in events if isinstance(e, ExecutorCompletedEvent)]
+    completed_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "executor_completed"]
 
     # Sender should have completed with the sent messages
     sender_completed = next(e for e in completed_events if e.executor_id == "sender")
@@ -210,9 +209,7 @@ async def test_executor_completed_event_contains_sent_messages():
 
 
 async def test_executor_completed_event_includes_yielded_outputs():
-    """Test that ExecutorCompletedEvent.data includes yielded outputs."""
-
-    from agent_framework import WorkflowOutputEvent
+    """Test that WorkflowEvent(type='executor_completed').data includes yielded outputs."""
 
     class YieldOnlyExecutor(Executor):
         @handler
@@ -223,15 +220,15 @@ async def test_executor_completed_event_includes_yielded_outputs():
     workflow = WorkflowBuilder().set_start_executor(executor).build()
 
     events = await workflow.run("test")
-    completed_events = [e for e in events if isinstance(e, ExecutorCompletedEvent)]
+    completed_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "executor_completed"]
 
     assert len(completed_events) == 1
     assert completed_events[0].executor_id == "yielder"
-    # Yielded outputs are now included in ExecutorCompletedEvent.data
+    # Yielded outputs are now included in executor_completed event (type='executor_completed').data
     assert completed_events[0].data == ["TEST"]
 
-    # Verify the output was also yielded as WorkflowOutputEvent
-    output_events = [e for e in events if isinstance(e, WorkflowOutputEvent)]
+    # Verify the output was also yielded as an output event (type='output')
+    output_events = [e for e in events if e.type == "output"]
     assert len(output_events) == 1
     assert output_events[0].data == "TEST"
 
@@ -268,8 +265,8 @@ async def test_executor_events_with_complex_message_types():
     input_request = Request(query="hello", limit=3)
     events = await workflow.run(input_request)
 
-    invoked_events = [e for e in events if isinstance(e, ExecutorInvokedEvent)]
-    completed_events = [e for e in events if isinstance(e, ExecutorCompletedEvent)]
+    invoked_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "executor_invoked"]
+    completed_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "executor_completed"]
 
     # Check processor invoked event has the Request object
     processor_invoked = next(e for e in invoked_events if e.executor_id == "processor")
@@ -531,7 +528,7 @@ def test_executor_response_handler_union_output_types():
 
 
 async def test_executor_invoked_event_data_not_mutated_by_handler():
-    """Test that ExecutorInvokedEvent.data captures original input, not mutated input."""
+    """Test that executor_invoked event (type='executor_invoked').data captures original input, not mutated input."""
 
     @executor(id="Mutator")
     async def mutator(messages: list[ChatMessage], ctx: WorkflowContext[list[ChatMessage]]) -> None:
@@ -549,7 +546,7 @@ async def test_executor_invoked_event_data_not_mutated_by_handler():
     events = await workflow.run(input_messages)
 
     # Find the invoked event for the Mutator executor
-    invoked_events = [e for e in events if isinstance(e, ExecutorInvokedEvent)]
+    invoked_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "executor_invoked"]
     assert len(invoked_events) == 1
     mutator_invoked = invoked_events[0]
 

python/packages/core/tests/workflow/test_full_conversation.py

@@ -20,7 +20,6 @@ from agent_framework import (
     WorkflowBuilder,
     WorkflowContext,
     WorkflowRunState,
-    WorkflowStatusEvent,
     handler,
 )
 from agent_framework.orchestrations import SequentialBuilder
@@ -149,7 +148,7 @@ async def test_sequential_adapter_uses_full_conversation() -> None:
 
     # Act
     async for ev in wf.run("hello seq", stream=True):
-        if isinstance(ev, WorkflowStatusEvent) and ev.state == WorkflowRunState.IDLE:
+        if ev.type == "status" and ev.state == WorkflowRunState.IDLE:
             break
 
     # Assert: second agent should have seen the user prompt and A1's assistant reply

python/packages/core/tests/workflow/test_request_info_and_response.py

@@ -4,11 +4,10 @@ from dataclasses import dataclass
 
 from agent_framework import (
     FileCheckpointStorage,
-    RequestInfoEvent,
     WorkflowBuilder,
     WorkflowContext,
+    WorkflowEvent,
     WorkflowRunState,
-    WorkflowStatusEvent,
     handler,
     response_handler,
 )
@@ -182,9 +181,9 @@ class TestRequestInfoAndResponse:
         workflow = WorkflowBuilder().set_start_executor(executor).build()
 
         # First run the workflow until it emits a request
-        request_info_event: RequestInfoEvent | None = None
+        request_info_event: WorkflowEvent | None = None
         async for event in workflow.run("test operation", stream=True):
-            if isinstance(event, RequestInfoEvent):
+            if event.type == "request_info":
                 request_info_event = event
 
         assert request_info_event is not None
@@ -194,7 +193,7 @@ class TestRequestInfoAndResponse:
         # Send response and continue workflow
         completed = False
         async for event in workflow.send_responses_streaming({request_info_event.request_id: True}):
-            if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+            if event.type == "status" and event.state == WorkflowRunState.IDLE:
                 completed = True
 
         assert completed
@@ -207,9 +206,9 @@ class TestRequestInfoAndResponse:
         workflow = WorkflowBuilder().set_start_executor(executor).build()
 
         # First run the workflow until it emits a calculation request
-        request_info_event: RequestInfoEvent | None = None
+        request_info_event: WorkflowEvent | None = None
         async for event in workflow.run("multiply 15.5 2.0", stream=True):
-            if isinstance(event, RequestInfoEvent):
+            if event.type == "request_info":
                 request_info_event = event
 
         assert request_info_event is not None
@@ -221,7 +220,7 @@ class TestRequestInfoAndResponse:
         calculated_result = 31.0
         completed = False
         async for event in workflow.send_responses_streaming({request_info_event.request_id: calculated_result}):
-            if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+            if event.type == "status" and event.state == WorkflowRunState.IDLE:
                 completed = True
 
         assert completed
@@ -234,18 +233,18 @@ class TestRequestInfoAndResponse:
         workflow = WorkflowBuilder().set_start_executor(executor).build()
 
         # Collect all request events by running the full stream
-        request_events: list[RequestInfoEvent] = []
+        request_events: list[WorkflowEvent] = []
         async for event in workflow.run("start batch", stream=True):
-            if isinstance(event, RequestInfoEvent):
+            if event.type == "request_info":
                 request_events.append(event)
 
         assert len(request_events) == 2
 
         # Find the approval and calculation requests
-        approval_event: RequestInfoEvent | None = next(
+        approval_event: WorkflowEvent | None = next(
             (e for e in request_events if isinstance(e.data, UserApprovalRequest)), None
         )
-        calc_event: RequestInfoEvent | None = next(
+        calc_event: WorkflowEvent | None = next(
             (e for e in request_events if isinstance(e.data, CalculationRequest)), None
         )
 
@@ -256,7 +255,7 @@ class TestRequestInfoAndResponse:
         responses = {approval_event.request_id: True, calc_event.request_id: 50.0}
         completed = False
         async for event in workflow.send_responses_streaming(responses):
-            if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+            if event.type == "status" and event.state == WorkflowRunState.IDLE:
                 completed = True
 
         assert completed
@@ -268,9 +267,9 @@ class TestRequestInfoAndResponse:
         workflow = WorkflowBuilder().set_start_executor(executor).build()
 
         # First run the workflow until it emits a request
-        request_info_event: RequestInfoEvent | None = None
+        request_info_event: WorkflowEvent | None = None
         async for event in workflow.run("sensitive operation", stream=True):
-            if isinstance(event, RequestInfoEvent):
+            if event.type == "request_info":
                 request_info_event = event
 
         assert request_info_event is not None
@@ -278,7 +277,7 @@ class TestRequestInfoAndResponse:
         # Deny the request
         completed = False
         async for event in workflow.send_responses_streaming({request_info_event.request_id: False}):
-            if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+            if event.type == "status" and event.state == WorkflowRunState.IDLE:
                 completed = True
 
         assert completed
@@ -291,12 +290,12 @@ class TestRequestInfoAndResponse:
         workflow = WorkflowBuilder().set_start_executor(executor).build()
 
         # Run workflow until idle with pending requests
-        request_info_event: RequestInfoEvent | None = None
+        request_info_event: WorkflowEvent | None = None
         idle_with_pending = False
         async for event in workflow.run("test operation", stream=True):
-            if isinstance(event, RequestInfoEvent):
+            if event.type == "request_info":
                 request_info_event = event
-            elif isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS:
+            elif event.type == "status" and event.state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS:
                 idle_with_pending = True
 
         assert request_info_event is not None
@@ -305,7 +304,7 @@ class TestRequestInfoAndResponse:
         # Continue with response
         completed = False
         async for event in workflow.send_responses_streaming({request_info_event.request_id: True}):
-            if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+            if event.type == "status" and event.state == WorkflowRunState.IDLE:
                 completed = True
 
         assert completed
@@ -318,7 +317,7 @@ class TestRequestInfoAndResponse:
         # Send invalid input (no numbers)
         completed = False
         async for event in workflow.run("invalid input", stream=True):
-            if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+            if event.type == "status" and event.state == WorkflowRunState.IDLE:
                 completed = True
 
         assert completed
@@ -338,9 +337,9 @@ class TestRequestInfoAndResponse:
             workflow = WorkflowBuilder().set_start_executor(executor).with_checkpointing(storage).build()
 
             # Step 1: Run workflow to completion to ensure checkpoints are created
-            request_info_event: RequestInfoEvent | None = None
+            request_info_event: WorkflowEvent | None = None
             async for event in workflow.run("checkpoint test operation", stream=True):
-                if isinstance(event, RequestInfoEvent):
+                if event.type == "request_info":
                     request_info_event = event
 
             # Verify request was emitted
@@ -377,15 +376,12 @@ class TestRequestInfoAndResponse:
 
             # Step 5: Resume from checkpoint and verify the request can be continued
             completed = False
-            restored_request_event: RequestInfoEvent | None = None
+            restored_request_event: WorkflowEvent | None = None
             async for event in restored_workflow.run(checkpoint_id=checkpoint_with_request.checkpoint_id, stream=True):
                 # Should re-emit the pending request info event
-                if isinstance(event, RequestInfoEvent) and event.request_id == request_info_event.request_id:
+                if event.type == "request_info" and event.request_id == request_info_event.request_id:
                     restored_request_event = event
-                elif (
-                    isinstance(event, WorkflowStatusEvent)
-                    and event.state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
-                ):
+                elif event.type == "status" and event.state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS:
                     completed = True
 
             assert completed, "Workflow should reach idle with pending requests state after restoration"
@@ -402,7 +398,7 @@ class TestRequestInfoAndResponse:
             async for event in restored_workflow.send_responses_streaming({
                 request_info_event.request_id: True  # Approve the request
             }):
-                if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+                if event.type == "status" and event.state == WorkflowRunState.IDLE:
                     final_completed = True
 
             assert final_completed, "Workflow should complete after providing response to restored request"

python/packages/core/tests/workflow/test_request_info_event_rehydrate.py

@@ -9,7 +9,7 @@ import pytest
 from agent_framework import InMemoryCheckpointStorage, InProcRunnerContext
 from agent_framework._workflows._checkpoint_encoding import DATACLASS_MARKER, encode_checkpoint_value
 from agent_framework._workflows._checkpoint_summary import get_checkpoint_summary
-from agent_framework._workflows._events import RequestInfoEvent
+from agent_framework._workflows._events import WorkflowEvent
 from agent_framework._workflows._state import State
 
 
@@ -36,7 +36,7 @@ class TimedApproval:
 
 async def test_rehydrate_request_info_event() -> None:
     """Rehydration should succeed for valid request info events."""
-    request_info_event = RequestInfoEvent(
+    request_info_event = WorkflowEvent.request_info(
         request_id="request-123",
         source_executor_id="review_gateway",
         request_data=MockRequest(),
@@ -69,7 +69,7 @@ async def test_rehydrate_request_info_event() -> None:
 
 async def test_rehydrate_fails_when_request_type_missing() -> None:
     """Rehydration should fail is the request type is missing or fails to import."""
-    request_info_event = RequestInfoEvent(
+    request_info_event = WorkflowEvent.request_info(
         request_id="request-123",
         source_executor_id="review_gateway",
         request_data=MockRequest(),
@@ -97,7 +97,7 @@ async def test_rehydrate_fails_when_request_type_missing() -> None:
 
 async def test_rehydrate_fails_when_request_type_mismatch() -> None:
     """Rehydration should fail if the request type is mismatched."""
-    request_info_event = RequestInfoEvent(
+    request_info_event = WorkflowEvent.request_info(
         request_id="request-123",
         source_executor_id="review_gateway",
         request_data=MockRequest(),
@@ -127,7 +127,7 @@ async def test_rehydrate_fails_when_request_type_mismatch() -> None:
 
 async def test_pending_requests_in_summary() -> None:
     """Test that pending requests are correctly summarized in the checkpoint summary."""
-    request_info_event = RequestInfoEvent(
+    request_info_event = WorkflowEvent.request_info(
         request_id="request-123",
         source_executor_id="review_gateway",
         request_data=MockRequest(),
@@ -148,7 +148,8 @@ async def test_pending_requests_in_summary() -> None:
 
     assert len(summary.pending_request_info_events) == 1
     pending_event = summary.pending_request_info_events[0]
-    assert isinstance(pending_event, RequestInfoEvent)
+    assert isinstance(pending_event, WorkflowEvent)
+    assert pending_event.type == "request_info"
     assert pending_event.request_id == "request-123"
 
     assert pending_event.source_executor_id == "review_gateway"
@@ -158,13 +159,13 @@ async def test_pending_requests_in_summary() -> None:
 
 
 async def test_request_info_event_serializes_non_json_payloads() -> None:
-    req_1 = RequestInfoEvent(
+    req_1 = WorkflowEvent.request_info(
         request_id="req-1",
         source_executor_id="source",
         request_data=TimedApproval(issued_at=datetime(2024, 5, 4, 12, 30, 45)),
         response_type=bool,
     )
-    req_2 = RequestInfoEvent(
+    req_2 = WorkflowEvent.request_info(
         request_id="req-2",
         source_executor_id="source",
         request_data=SlottedApproval(note="slot-based"),

python/packages/core/tests/workflow/test_runner.py

@@ -12,10 +12,8 @@ from agent_framework import (
     WorkflowContext,
     WorkflowConvergenceException,
     WorkflowEvent,
-    WorkflowOutputEvent,
     WorkflowRunnerException,
     WorkflowRunState,
-    WorkflowStatusEvent,
     handler,
 )
 from agent_framework._workflows._edge import SingleEdgeGroup
@@ -97,7 +95,7 @@ async def test_runner_run_until_convergence():
     )
     async for event in runner.run_until_convergence():
         assert isinstance(event, WorkflowEvent)
-        if isinstance(event, WorkflowOutputEvent):
+        if event.type == "output":
             result = event.data
 
     assert result is not None and result == 10
@@ -137,7 +135,7 @@ async def test_runner_run_until_convergence_not_completed():
         match="Runner did not converge after 5 iterations.",
     ):
         async for event in runner.run_until_convergence():
-            assert not isinstance(event, WorkflowStatusEvent) or event.state != WorkflowRunState.IDLE
+            assert event.type != "status" or event.state != WorkflowRunState.IDLE
 
 
 async def test_runner_already_running():

python/packages/core/tests/workflow/test_sub_workflow.py

@@ -8,12 +8,12 @@ from typing_extensions import Never
 
 from agent_framework import (
     Executor,
-    RequestInfoEvent,
     SubWorkflowRequestMessage,
     SubWorkflowResponseMessage,
     Workflow,
     WorkflowBuilder,
     WorkflowContext,
+    WorkflowEvent,
     WorkflowExecutor,
     handler,
     response_handler,
@@ -592,7 +592,7 @@ async def test_sub_workflow_checkpoint_restore_no_duplicate_requests() -> None:
 
     first_request_id: str | None = None
     async for event in workflow1.run("test_value", stream=True):
-        if isinstance(event, RequestInfoEvent):
+        if event.type == "request_info":
             first_request_id = event.request_id
 
     assert first_request_id is not None
@@ -606,15 +606,15 @@ async def test_sub_workflow_checkpoint_restore_no_duplicate_requests() -> None:
 
     resumed_first_request_id: str | None = None
     async for event in workflow2.run(checkpoint_id=checkpoint_id, stream=True):
-        if isinstance(event, RequestInfoEvent):
+        if event.type == "request_info":
             resumed_first_request_id = event.request_id
 
     assert resumed_first_request_id is not None
     assert resumed_first_request_id == first_request_id
 
-    request_events: list[RequestInfoEvent] = []
+    request_events: list[WorkflowEvent] = []
     async for event in workflow2.send_responses_streaming({resumed_first_request_id: "first_answer"}):
-        if isinstance(event, RequestInfoEvent):
+        if event.type == "request_info":
             request_events.append(event)
 
     # Key assertion: Only the second request should be received, not a duplicate of the first

python/packages/core/tests/workflow/test_typing_utils.py

@@ -5,7 +5,7 @@ from typing import Any, Generic, Optional, TypeVar, Union
 
 import pytest
 
-from agent_framework import RequestInfoEvent
+from agent_framework import WorkflowEvent
 from agent_framework._workflows._typing_utils import (
     deserialize_type,
     is_instance_of,
@@ -308,18 +308,19 @@ def test_serialize_deserialize_roundtrip() -> None:
 
     # Test agent framework type roundtrip
 
-    serialized = serialize_type(RequestInfoEvent)
+    serialized = serialize_type(WorkflowEvent)
     deserialized = deserialize_type(serialized)
-    assert deserialized is RequestInfoEvent
+    assert deserialized is WorkflowEvent
 
-    # Verify we can instantiate the deserialized type
-    instance = deserialized(
+    # Verify we can instantiate the deserialized type via factory method
+    instance = WorkflowEvent.request_info(
         request_id="request-123",
         source_executor_id="executor_1",
         request_data="test",
         response_type=str,
     )
-    assert isinstance(instance, RequestInfoEvent)
+    assert isinstance(instance, WorkflowEvent)
+    assert instance.type == "request_info"
 
 
 def test_deserialize_type_error_handling() -> None:

python/packages/core/tests/workflow/test_workflow.py

@@ -20,16 +20,13 @@ from agent_framework import (
     Executor,
     FileCheckpointStorage,
     Message,
-    RequestInfoEvent,
     ResponseStream,
     WorkflowBuilder,
     WorkflowCheckpointException,
     WorkflowContext,
     WorkflowConvergenceException,
     WorkflowEvent,
-    WorkflowOutputEvent,
     WorkflowRunState,
-    WorkflowStatusEvent,
     handler,
     response_handler,
 )
@@ -123,7 +120,7 @@ async def test_workflow_run_streaming() -> None:
     result: int | None = None
     async for event in workflow.run(NumberMessage(data=0), stream=True):
         assert isinstance(event, WorkflowEvent)
-        if isinstance(event, WorkflowOutputEvent):
+        if event.type == "output":
             result = event.data
 
     assert result is not None and result == 10
@@ -197,9 +194,10 @@ async def test_fan_out():
 
     events = await workflow.run(NumberMessage(data=0))
 
-    # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent
-    # executor_b will also emit a WorkflowOutputEvent (no WorkflowCompletedEvent anymore)
-    # Each superstep will emit also emit a WorkflowStartedEvent and WorkflowCompletedEvent
+    # Each executor will emit two events: executor_invoked (type='executor_invoked')
+    # and executor_completed (type='executor_completed')
+    # executor_b will also emit an output event (type='output')
+    # Each superstep will emit a started event (type='started') and status event (type='status')
     # This workflow will converge in 2 supersteps because executor_c will send one more message
     # after executor_b completes
     assert len(events) == 11
@@ -221,9 +219,10 @@ async def test_fan_out_multiple_completed_events():
 
     events = await workflow.run(NumberMessage(data=0))
 
-    # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent
-    # executor_b and executor_c will also emit a WorkflowOutputEvent (no WorkflowCompletedEvent anymore)
-    # Each superstep will emit also emit a WorkflowStartedEvent and WorkflowCompletedEvent
+    # Each executor will emit two events: executor_invoked (type='executor_invoked')
+    # and executor_completed (type='executor_completed')
+    # executor_b and executor_c will also emit an output event (type='output')
+    # Each superstep will emit a started event (type='started') and status event (type='status')
     # This workflow will converge in 1 superstep because executor_a and executor_b will not send further messages
     assert len(events) == 10
 
@@ -249,9 +248,10 @@ async def test_fan_in():
 
     events = await workflow.run(NumberMessage(data=0))
 
-    # Each executor will emit two events: ExecutorInvokedEvent and ExecutorCompletedEvent
-    # aggregator will also emit a WorkflowOutputEvent (no WorkflowCompletedEvent anymore)
-    # Each superstep will emit also emit a WorkflowStartedEvent and WorkflowCompletedEvent
+    # Each executor will emit two events: executor_invoked (type='executor_invoked')
+    # and executor_completed (type='executor_completed')
+    # aggregator will also emit an output event (type='output')
+    # Each superstep will emit a started event (type='started') and status event (type='status')
     assert len(events) == 13
 
     assert events.get_final_state() == WorkflowRunState.IDLE
@@ -427,7 +427,7 @@ async def test_workflow_run_from_checkpoint_non_streaming(simple_executor: Execu
 async def test_workflow_run_stream_from_checkpoint_with_responses(
     simple_executor: Executor,
 ):
-    """Test that workflow can be resumed from checkpoint with pending RequestInfoEvents."""
+    """Test that workflow can be resumed from checkpoint with pending request_info events."""
     with tempfile.TemporaryDirectory() as temp_dir:
         storage = FileCheckpointStorage(temp_dir)
 
@@ -439,7 +439,7 @@ async def test_workflow_run_stream_from_checkpoint_with_responses(
             messages={},
             state={},
             pending_request_info_events={
-                "request_123": RequestInfoEvent(
+                "request_123": WorkflowEvent.request_info(
                     request_id="request_123",
                     source_executor_id=simple_executor.id,
                     request_data="Mock",
@@ -465,9 +465,7 @@ async def test_workflow_run_stream_from_checkpoint_with_responses(
             events.append(event)
 
         # Verify that the pending request event was emitted
-        assert next(
-            event for event in events if isinstance(event, RequestInfoEvent) and event.request_id == "request_123"
-        )
+        assert next(event for event in events if event.type == "request_info" and event.request_id == "request_123")
 
         assert len(events) > 0  # Just ensure we processed some events
 
@@ -730,10 +728,12 @@ async def test_workflow_with_simple_cycle_and_exit_condition():
     assert outputs[0] is not None and outputs[0] >= 6  # Should complete when executor_a reaches its limit
 
     # Verify cycling occurred (should have events from both executors)
-    # Check for ExecutorInvokedEvent and ExecutorCompletedEvent types that have executor_id
-    from agent_framework import ExecutorCompletedEvent, ExecutorInvokedEvent
+    # Check for executor events that have executor_id
+    from agent_framework import WorkflowEvent
 
-    executor_events = [e for e in events if isinstance(e, (ExecutorInvokedEvent, ExecutorCompletedEvent))]
+    executor_events = [
+        e for e in events if isinstance(e, WorkflowEvent) and e.type in ("executor_invoked", "executor_completed")
+    ]
     executor_ids = {e.executor_id for e in executor_events}
     assert "exec_a" in executor_ids, "Should have events from executor A"
     assert "exec_b" in executor_ids, "Should have events from executor B"
@@ -880,7 +880,7 @@ class _StreamingTestAgent(BaseAgent):
 
 
 async def test_agent_streaming_vs_non_streaming() -> None:
-    """Test that stream=True/False both emits WorkflowOutputEvents correctly with the right data types."""
+    """Test that stream=True/False both emit output events (type='output') with the right data types."""
     agent = _StreamingTestAgent(id="test_agent", name="TestAgent", reply_text="Hello World")
     agent_exec = AgentExecutor(agent, id="agent_exec")
 
@@ -890,17 +890,15 @@ async def test_agent_streaming_vs_non_streaming() -> None:
     result = await workflow.run("test message")
 
     # Filter for agent events (result is a list of events)
-    agent_response = [e for e in result if isinstance(e, WorkflowOutputEvent) and isinstance(e.data, AgentResponse)]
-    agent_response_updates = [
-        e for e in result if isinstance(e, WorkflowOutputEvent) and isinstance(e.data, AgentResponseUpdate)
-    ]
+    agent_run_events = [e for e in result if e.type == "output" and isinstance(e.data, AgentResponse)]
+    agent_update_events = [e for e in result if e.type == "output" and isinstance(e.data, AgentResponseUpdate)]
 
-    # In non-streaming mode, should have AgentResponse, no AgentResponseUpdate
-    assert len(agent_response) == 1, "Expected exactly one AgentResponse in non-streaming mode"
-    assert len(agent_response_updates) == 0, "Expected no AgentResponseUpdate in non-streaming mode"
-    assert agent_response[0].executor_id == "agent_exec"
-    assert agent_response[0].data is not None
-    assert agent_response[0].data.messages[0].text == "Hello World"
+    # In non-streaming mode, should have output event with AgentResponse, no AgentResponseUpdate
+    assert len(agent_run_events) == 1, "Expected exactly one output event with AgentResponse in non-streaming mode"
+    assert len(agent_update_events) == 0, "Expected no output event with AgentResponseUpdate in non-streaming mode"
+    assert agent_run_events[0].executor_id == "agent_exec"
+    assert agent_run_events[0].data is not None
+    assert agent_run_events[0].data.messages[0].text == "Hello World"
 
     # Test streaming mode with run(stream=True)
     stream_events: list[WorkflowEvent] = []
@@ -909,12 +907,10 @@ async def test_agent_streaming_vs_non_streaming() -> None:
 
     # Filter for agent events
     agent_response = [
-        cast(AgentResponse, e.data)  # type: ignore
-        for e in stream_events
-        if isinstance(e, WorkflowOutputEvent) and isinstance(e.data, AgentResponse)
+        cast(AgentResponse, e.data) for e in stream_events if e.type == "output" and isinstance(e.data, AgentResponse)
     ]
     agent_response_updates = [
-        e.data for e in stream_events if isinstance(e, WorkflowOutputEvent) and isinstance(e.data, AgentResponseUpdate)
+        e.data for e in stream_events if e.type == "output" and isinstance(e.data, AgentResponseUpdate)
     ]
 
     # In streaming mode, should have AgentResponseUpdate, no AgentResponse
@@ -977,7 +973,7 @@ async def test_workflow_run_stream_parameter_validation(
     events: list[WorkflowEvent] = []
     async for event in workflow.run(test_message, stream=True):
         events.append(event)
-    assert any(isinstance(e, WorkflowStatusEvent) and e.state == WorkflowRunState.IDLE for e in events)
+    assert any(e.type == "status" and e.state == WorkflowRunState.IDLE for e in events)
 
     # Invalid combinations already tested in test_workflow_run_parameter_validation
     # This test ensures streaming works correctly for valid parameters
@@ -1027,7 +1023,7 @@ async def test_output_executors_empty_yields_all_outputs() -> None:
     assert len(outputs) == 2
     assert outputs == [10, 20]
 
-    output_events = [event for event in result if isinstance(event, WorkflowOutputEvent)]
+    output_events = [event for event in result if event.type == "output"]
     assert len(output_events) == 2
     assert output_events[0].executor_id == "executor_a"
     assert output_events[1].executor_id == "executor_b"
@@ -1055,7 +1051,7 @@ async def test_output_executors_filters_outputs_non_streaming() -> None:
     assert len(outputs) == 1
     assert outputs[0] == 20
 
-    output_events = [event for event in result if isinstance(event, WorkflowOutputEvent)]
+    output_events = [event for event in result if event.type == "output"]
     assert len(output_events) == 1
     assert output_events[0].executor_id == "executor_b"
 
@@ -1076,9 +1072,9 @@ async def test_output_executors_filters_outputs_streaming() -> None:
     )
 
     # Collect outputs from streaming
-    output_events: list[WorkflowOutputEvent] = []
+    output_events: list[WorkflowEvent] = []
     async for event in workflow.run(NumberMessage(data=0), stream=True):
-        if isinstance(event, WorkflowOutputEvent):
+        if event.type == "output":
             output_events.append(event)
 
     # Only executor_a's output should be present
@@ -1213,7 +1209,7 @@ async def test_output_executors_filtering_with_send_responses_streaming() -> Non
         events_list.append(event)
 
     # Get request info events
-    request_events = [e for e in events_list if isinstance(e, RequestInfoEvent)]
+    request_events = [e for e in events_list if e.type == "request_info"]
     assert len(request_events) == 1
 
     # Set output_executors to exclude the approval executor
@@ -1221,9 +1217,9 @@ async def test_output_executors_filtering_with_send_responses_streaming() -> Non
 
     # Send approval response via streaming
     responses = {request_events[0].request_id: ApprovalMessage(approved=True)}
-    output_events: list[WorkflowOutputEvent] = []
+    output_events: list[WorkflowEvent] = []
     async for event in workflow.send_responses_streaming(responses):
-        if isinstance(event, WorkflowOutputEvent):
+        if event.type == "output":
             output_events.append(event)
 
     # No outputs should be yielded since approval_executor is not in output_executors

python/packages/core/tests/workflow/test_workflow_agent.py

@@ -218,7 +218,7 @@ class TestWorkflowAgent:
         assert "Streaming2: Streaming1: Test input" in second_content.text
 
     async def test_end_to_end_request_info_handling(self):
-        """Test end-to-end workflow with RequestInfoEvent handling."""
+        """Test end-to-end workflow with request_info event (type='request_info') handling."""
         # Create workflow with requesting executor -> request info executor (no cycle)
         simple_executor = SimpleExecutor(id="simple", response_text="SimpleResponse", streaming=False)
         requesting_executor = RequestingExecutor(id="requester", streaming=False)
@@ -331,7 +331,7 @@ class TestWorkflowAgent:
     async def test_workflow_as_agent_yield_output_surfaces_as_agent_response(self) -> None:
         """Test that ctx.yield_output() in a workflow executor surfaces as agent output when using .as_agent().
 
-        This validates the fix for issue #2813: WorkflowOutputEvent should be converted to
+        This validates the fix for issue #2813: output event (type='output') should be converted to
         AgentResponseUpdate when the workflow is wrapped via .as_agent().
         """
 
@@ -343,7 +343,7 @@ class TestWorkflowAgent:
 
         workflow = WorkflowBuilder().set_start_executor(yielding_executor).build()
 
-        # Run directly - should return WorkflowOutputEvent in result
+        # Run directly - should return output event (type='output') in result
         direct_result = await workflow.run([ChatMessage(role="user", text="hello")])
         direct_outputs = direct_result.get_outputs()
         assert len(direct_outputs) == 1
@@ -779,7 +779,7 @@ class TestWorkflowAgent:
         # Count occurrences of the unique response text
         unique_text_count = sum(1 for msg in result.messages if msg.text and "Unique response text" in msg.text)
 
-        # Should appear exactly once (not duplicated from both streaming and WorkflowOutputEvent)
+        # Should appear exactly once (not duplicated from both streaming and output event)
         assert unique_text_count == 1, f"Response should appear exactly once, but appeared {unique_text_count} times"
 
 
@@ -793,7 +793,7 @@ class TestWorkflowAgentAuthorName:
         identification of which agent produced them in multi-agent workflows.
         """
         # Create workflow with executor that emits AgentResponseUpdate without author_name
-        executor1 = SimpleExecutor(id="my_executor_id", response_text="Response")
+        executor1 = SimpleExecutor(id="my_executor_id", response_text="Response", streaming=True)
         workflow = WorkflowBuilder().set_start_executor(executor1).build()
         agent = WorkflowAgent(workflow=workflow, name="Test Agent")
 

python/packages/core/tests/workflow/test_workflow_context.py

@@ -13,7 +13,6 @@ from agent_framework import (
     WorkflowContext,
     WorkflowEvent,
     WorkflowRunState,
-    WorkflowStatusEvent,
     executor,
     handler,
 )
@@ -62,15 +61,15 @@ async def test_executor_cannot_emit_framework_lifecycle_event(caplog: "LogCaptur
     async with make_context() as (ctx, runner_ctx):
         caplog.clear()
         with caplog.at_level("WARNING"):
-            await ctx.add_event(WorkflowStatusEvent(state=WorkflowRunState.IN_PROGRESS))
+            await ctx.add_event(WorkflowEvent.status(state=WorkflowRunState.IN_PROGRESS))
 
         events: list[WorkflowEvent] = await runner_ctx.drain_events()
         assert len(events) == 1
-        assert type(events[0]).__name__ == "WorkflowWarningEvent"
-        data = getattr(events[0], "data", None)
+        assert events[0].type == "warning"
+        data = events[0].data
         assert isinstance(data, str)
         assert "reserved for framework lifecycle notifications" in data
-        assert any("attempted to emit WorkflowStatusEvent" in message for message in list(caplog.messages))
+        assert any("attempted to emit" in message and "'status'" in message for message in list(caplog.messages))
 
 
 async def test_executor_emits_normal_event() -> None:
@@ -84,7 +83,8 @@ async def test_executor_emits_normal_event() -> None:
 
 
 class _TestEvent(WorkflowEvent):
-    pass
+    def __init__(self, data: Any = None) -> None:
+        super().__init__("test_event", data=data)
 
 
 async def test_workflow_context_type_annotations_no_parameter() -> None:

python/packages/core/tests/workflow/test_workflow_kwargs.py

@@ -14,7 +14,6 @@ from agent_framework import (
     Content,
     ResponseStream,
     WorkflowRunState,
-    WorkflowStatusEvent,
     tool,
 )
 from agent_framework._workflows._const import WORKFLOW_RUN_KWARGS_KEY
@@ -90,7 +89,7 @@ async def test_sequential_kwargs_flow_to_agent() -> None:
         custom_data=custom_data,
         user_token=user_token,
     ):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # Verify agent received kwargs
@@ -111,7 +110,7 @@ async def test_sequential_kwargs_flow_to_multiple_agents() -> None:
     custom_data = {"key": "value"}
 
     async for event in workflow.run("test", custom_data=custom_data, stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # Both agents should have received kwargs
@@ -153,7 +152,7 @@ async def test_concurrent_kwargs_flow_to_agents() -> None:
         custom_data=custom_data,
         user_token=user_token,
     ):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # Both agents should have received kwargs
@@ -200,7 +199,7 @@ async def test_groupchat_kwargs_flow_to_agents() -> None:
     custom_data = {"session_id": "group123"}
 
     async for event in workflow.run("group chat test", custom_data=custom_data, stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # At least one agent should have received kwargs
@@ -234,7 +233,7 @@ async def test_kwargs_stored_in_state() -> None:
     workflow = SequentialBuilder().participants([inspector]).build()
 
     async for event in workflow.run("test", my_kwarg="my_value", another=123, stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     assert stored_kwargs is not None, "kwargs should be stored in State"
@@ -260,7 +259,7 @@ async def test_empty_kwargs_stored_as_empty_dict() -> None:
 
     # Run without any kwargs
     async for event in workflow.run("test", stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # State should have empty dict when no kwargs provided
@@ -279,7 +278,7 @@ async def test_kwargs_with_none_values() -> None:
     workflow = SequentialBuilder().participants([agent]).build()
 
     async for event in workflow.run("test", optional_param=None, other_param="value", stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     assert len(agent.captured_kwargs) >= 1
@@ -306,7 +305,7 @@ async def test_kwargs_with_complex_nested_data() -> None:
     }
 
     async for event in workflow.run("test", complex_data=complex_data, stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     assert len(agent.captured_kwargs) >= 1
@@ -324,12 +323,12 @@ async def test_kwargs_preserved_across_workflow_reruns() -> None:
 
     # First run
     async for event in workflow1.run("run1", run_id="first", stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # Second run with different kwargs (using fresh workflow)
     async for event in workflow2.run("run2", run_id="second", stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     assert len(agent.captured_kwargs) >= 2
@@ -361,7 +360,7 @@ async def test_handoff_kwargs_flow_to_agents() -> None:
     custom_data = {"session_id": "handoff123"}
 
     async for event in workflow.run("handoff test", custom_data=custom_data, stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # Coordinator agent should have received kwargs
@@ -419,7 +418,7 @@ async def test_magentic_kwargs_flow_to_agents() -> None:
     custom_data = {"session_id": "magentic123"}
 
     async for event in workflow.run("magentic test", custom_data=custom_data, stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # The workflow completes immediately via prepare_final_answer without invoking agents
@@ -470,7 +469,7 @@ async def test_magentic_kwargs_stored_in_state() -> None:
     custom_data = {"magentic_key": "magentic_value"}
 
     async for event in magentic_workflow.run("test task", custom_data=custom_data, stream=True):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # Verify the workflow completed (kwargs were stored, even if agent wasn't invoked)
@@ -626,7 +625,7 @@ async def test_subworkflow_kwargs_propagation() -> None:
         custom_data=custom_data,
         user_token=user_token,
     ):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # Verify that the inner agent was called
@@ -686,7 +685,7 @@ async def test_subworkflow_kwargs_accessible_via_state() -> None:
         my_custom_kwarg="should_be_propagated",
         another_kwarg=42,
     ):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # Verify the state reader was invoked
@@ -732,7 +731,7 @@ async def test_nested_subworkflow_kwargs_propagation() -> None:
         stream=True,
         deep_kwarg="should_reach_inner",
     ):
-        if isinstance(event, WorkflowStatusEvent) and event.state == WorkflowRunState.IDLE:
+        if event.type == "status" and event.state == WorkflowRunState.IDLE:
             break
 
     # Verify inner agent was called

python/packages/core/tests/workflow/test_workflow_states.py

@@ -5,18 +5,14 @@ from typing_extensions import Never
 
 from agent_framework import (
     Executor,
-    ExecutorFailedEvent,
     InProcRunnerContext,
-    RequestInfoEvent,
     Workflow,
     WorkflowBuilder,
     WorkflowContext,
+    WorkflowEvent,
     WorkflowEventSource,
-    WorkflowFailedEvent,
     WorkflowRunResult,
     WorkflowRunState,
-    WorkflowStartedEvent,
-    WorkflowStatusEvent,
     handler,
 )
 from agent_framework._workflows._state import State
@@ -39,24 +35,26 @@ async def test_executor_failed_and_workflow_failed_events_streaming():
         async for ev in wf.run(0, stream=True):
             events.append(ev)
 
-    # ExecutorFailedEvent should be emitted before WorkflowFailedEvent
-    executor_failed_events = [e for e in events if isinstance(e, ExecutorFailedEvent)]
-    assert executor_failed_events, "ExecutorFailedEvent should be emitted when start executor fails"
+    # executor_failed event (type='executor_failed') should be emitted before workflow failed event
+    executor_failed_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "executor_failed"]
+    assert executor_failed_events, "executor_failed event should be emitted when start executor fails"
     assert executor_failed_events[0].executor_id == "f"
     assert executor_failed_events[0].origin is WorkflowEventSource.FRAMEWORK
 
     # Workflow-level failure and FAILED status should be surfaced
-    failed_events = [e for e in events if isinstance(e, WorkflowFailedEvent)]
+    failed_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "failed"]
     assert failed_events
     assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in failed_events)
-    status = [e for e in events if isinstance(e, WorkflowStatusEvent)]
+    status = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "status"]
     assert status and status[-1].state == WorkflowRunState.FAILED
     assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in status)
 
-    # Verify ExecutorFailedEvent comes before WorkflowFailedEvent
+    # Verify executor_failed event comes before workflow failed event
     executor_failed_idx = events.index(executor_failed_events[0])
     workflow_failed_idx = events.index(failed_events[0])
-    assert executor_failed_idx < workflow_failed_idx, "ExecutorFailedEvent should be emitted before WorkflowFailedEvent"
+    assert executor_failed_idx < workflow_failed_idx, (
+        "executor_failed event should be emitted before workflow failed event"
+    )
 
 
 async def test_executor_failed_event_emitted_on_direct_execute():
@@ -71,7 +69,7 @@ async def test_executor_failed_event_emitted_on_direct_execute():
             ctx,
         )
     drained = await ctx.drain_events()
-    failed = [e for e in drained if isinstance(e, ExecutorFailedEvent)]
+    failed = [e for e in drained if isinstance(e, WorkflowEvent) and e.type == "executor_failed"]
     assert failed
     assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in failed)
 
@@ -85,7 +83,7 @@ class PassthroughExecutor(Executor):
 
 
 async def test_executor_failed_event_from_second_executor_in_chain():
-    """Test that ExecutorFailedEvent is emitted when a non-start executor fails."""
+    """Test that executor_failed event is emitted when a non-start executor fails."""
     passthrough = PassthroughExecutor(id="passthrough")
     failing = FailingExecutor(id="failing")
     wf: Workflow = WorkflowBuilder().set_start_executor(passthrough).add_edge(passthrough, failing).build()
@@ -95,21 +93,23 @@ async def test_executor_failed_event_from_second_executor_in_chain():
         async for ev in wf.run(0, stream=True):
             events.append(ev)
 
-    # ExecutorFailedEvent should be emitted for the failing executor
-    executor_failed_events = [e for e in events if isinstance(e, ExecutorFailedEvent)]
-    assert executor_failed_events, "ExecutorFailedEvent should be emitted when second executor fails"
+    # executor_failed event should be emitted for the failing executor
+    executor_failed_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "executor_failed"]
+    assert executor_failed_events, "executor_failed event should be emitted when second executor fails"
     assert executor_failed_events[0].executor_id == "failing"
     assert executor_failed_events[0].origin is WorkflowEventSource.FRAMEWORK
 
     # Workflow-level failure should also be surfaced
-    failed_events = [e for e in events if isinstance(e, WorkflowFailedEvent)]
+    failed_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "failed"]
     assert failed_events
     assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in failed_events)
 
-    # Verify ExecutorFailedEvent comes before WorkflowFailedEvent
+    # Verify executor_failed event comes before workflow failed event
     executor_failed_idx = events.index(executor_failed_events[0])
     workflow_failed_idx = events.index(failed_events[0])
-    assert executor_failed_idx < workflow_failed_idx, "ExecutorFailedEvent should be emitted before WorkflowFailedEvent"
+    assert executor_failed_idx < workflow_failed_idx, (
+        "executor_failed event should be emitted before workflow failed event"
+    )
 
 
 class SimpleExecutor(Executor):
@@ -136,8 +136,8 @@ async def test_idle_with_pending_requests_status_streaming():
     events = [ev async for ev in wf.run("start", stream=True)]  # Consume stream fully
 
     # Ensure a request was emitted
-    assert any(isinstance(e, RequestInfoEvent) for e in events)
-    status_events = [e for e in events if isinstance(e, WorkflowStatusEvent)]
+    assert any(isinstance(e, WorkflowEvent) and e.type == "request_info" for e in events)
+    status_events = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "status"]
     assert len(status_events) >= 3
     assert status_events[-2].state == WorkflowRunState.IN_PROGRESS_PENDING_REQUESTS
     assert status_events[-1].state == WorkflowRunState.IDLE_WITH_PENDING_REQUESTS
@@ -156,7 +156,7 @@ async def test_completed_status_streaming():
     wf = WorkflowBuilder().set_start_executor(c).build()
     events = [ev async for ev in wf.run("ok", stream=True)]  # no raise
     # Last status should be IDLE
-    status = [e for e in events if isinstance(e, WorkflowStatusEvent)]
+    status = [e for e in events if isinstance(e, WorkflowEvent) and e.type == "status"]
     assert status and status[-1].state == WorkflowRunState.IDLE
     assert all(e.origin is WorkflowEventSource.FRAMEWORK for e in status)
 
@@ -166,12 +166,13 @@ async def test_started_and_completed_event_origins():
     wf = WorkflowBuilder().set_start_executor(c).build()
     events = [ev async for ev in wf.run("payload", stream=True)]
 
-    started = next(e for e in events if isinstance(e, WorkflowStartedEvent))
+    started = next(e for e in events if isinstance(e, WorkflowEvent) and e.type == "started")
     assert started.origin is WorkflowEventSource.FRAMEWORK
 
     # Check for IDLE status indicating completion
     idle_status = next(
-        (e for e in events if isinstance(e, WorkflowStatusEvent) and e.state == WorkflowRunState.IDLE), None
+        (e for e in events if isinstance(e, WorkflowEvent) and e.type == "status" and e.state == WorkflowRunState.IDLE),
+        None,
     )
     assert idle_status is not None
     assert idle_status.origin is WorkflowEventSource.FRAMEWORK