Python: [Breaking] Simplified Content types to a single class with classmethod constructors. (#3252)

* ported Content to a new model

* fixed linting

* fixes

* fixed data format handling

* fix for 3.10 mypy

* fix

* fix int test

python/packages/devui/tests/test_mapper.py

@@ -14,11 +14,8 @@ import pytest
 # Import Agent Framework types
 from agent_framework._types import (
     AgentResponseUpdate,
-    ErrorContent,
-    FunctionCallContent,
-    FunctionResultContent,
+    Content,
     Role,
-    TextContent,
 )
 
 # Import real workflow event classes - NOT mocks!
@@ -71,15 +68,17 @@ def test_request() -> AgentFrameworkRequest:
 def create_test_content(content_type: str, **kwargs: Any) -> Any:
     """Create test content objects."""
     if content_type == "text":
-        return TextContent(text=kwargs.get("text", "Hello, world!"))
+        return Content.from_text(text=kwargs.get("text", "Hello, world!"))
     if content_type == "function_call":
-        return FunctionCallContent(
+        return Content.from_function_call(
             call_id=kwargs.get("call_id", "test_call_id"),
             name=kwargs.get("name", "test_func"),
             arguments=kwargs.get("arguments", {"param": "value"}),
         )
     if content_type == "error":
-        return ErrorContent(message=kwargs.get("message", "Test error"), error_code=kwargs.get("code", "test_error"))
+        return Content.from_error(
+            message=kwargs.get("message", "Test error"), error_code=kwargs.get("code", "test_error")
+        )
     raise ValueError(f"Unknown content type: {content_type}")
 
 
@@ -162,7 +161,7 @@ async def test_function_result_content_with_string_result(
     mapper: MessageMapper, test_request: AgentFrameworkRequest
 ) -> None:
     """Test FunctionResultContent with plain string result (regular tools)."""
-    content = FunctionResultContent(
+    content = Content.from_function_result(
         call_id="test_call_123",
         result="Hello, World!",
     )
@@ -182,9 +181,9 @@ async def test_function_result_content_with_nested_content_objects(
     mapper: MessageMapper, test_request: AgentFrameworkRequest
 ) -> None:
     """Test FunctionResultContent with nested Content objects (MCP tools case)."""
-    content = FunctionResultContent(
+    content = Content.from_function_result(
         call_id="mcp_call_456",
-        result=[TextContent(text="Hello from MCP!")],
+        result=[Content.from_text(text="Hello from MCP!")],
     )
     update = create_test_agent_update([content])
 
@@ -451,12 +450,12 @@ async def test_magentic_agent_run_update_event_with_agent_delta_metadata(
     This tests the ACTUAL event format Magentic emits - not a fake MagenticAgentDeltaEvent class.
     Magentic uses AgentRunUpdateEvent with additional_properties containing magentic_event_type.
     """
-    from agent_framework._types import AgentResponseUpdate, Role, TextContent
+    from agent_framework._types import AgentResponseUpdate, Role
     from agent_framework._workflows._events import AgentRunUpdateEvent
 
     # Create the REAL event format that Magentic emits
     update = AgentResponseUpdate(
-        contents=[TextContent(text="Hello from agent")],
+        contents=[Content.from_text(text="Hello from agent")],
         role=Role.ASSISTANT,
         author_name="Writer",
         additional_properties={
@@ -482,12 +481,12 @@ async def test_magentic_orchestrator_message_event(mapper: MessageMapper, test_r
     Magentic emits orchestrator planning/instruction messages using AgentRunUpdateEvent
     with additional_properties containing magentic_event_type='orchestrator_message'.
     """
-    from agent_framework._types import AgentResponseUpdate, Role, TextContent
+    from agent_framework._types import AgentResponseUpdate, Role
     from agent_framework._workflows._events import AgentRunUpdateEvent
 
     # Create orchestrator message event (REAL format from Magentic)
     update = AgentResponseUpdate(
-        contents=[TextContent(text="Planning: First, the writer will create content...")],
+        contents=[Content.from_text(text="Planning: First, the writer will create content...")],
         role=Role.ASSISTANT,
         author_name="Orchestrator",
         additional_properties={
@@ -518,20 +517,20 @@ async def test_magentic_events_use_same_event_class_as_other_workflows(
     additional_properties. Any mapper code checking for 'MagenticAgentDeltaEvent'
     class names is dead code.
     """
-    from agent_framework._types import AgentResponseUpdate, Role, TextContent
+    from agent_framework._types import AgentResponseUpdate, Role
     from agent_framework._workflows._events import AgentRunUpdateEvent
 
     # Create events the way different workflows do it
     # 1. Regular workflow (no additional_properties)
     regular_update = AgentResponseUpdate(
-        contents=[TextContent(text="Regular workflow response")],
+        contents=[Content.from_text(text="Regular workflow response")],
         role=Role.ASSISTANT,
     )
     regular_event = AgentRunUpdateEvent(executor_id="regular_executor", data=regular_update)
 
     # 2. Magentic workflow (with additional_properties)
     magentic_update = AgentResponseUpdate(
-        contents=[TextContent(text="Magentic workflow response")],
+        contents=[Content.from_text(text="Magentic workflow response")],
         role=Role.ASSISTANT,
         additional_properties={"magentic_event_type": "agent_delta"},
     )
@@ -599,13 +598,13 @@ async def test_workflow_output_event(mapper: MessageMapper, test_request: AgentF
 
 async def test_workflow_output_event_with_list_data(mapper: MessageMapper, test_request: AgentFrameworkRequest) -> None:
     """Test WorkflowOutputEvent with list data (common for sequential/concurrent workflows)."""
-    from agent_framework import ChatMessage, Role, TextContent
+    from agent_framework import ChatMessage, Role
     from agent_framework._workflows._events import WorkflowOutputEvent
 
     # Sequential/Concurrent workflows often output list[ChatMessage]
     messages = [
-        ChatMessage(role=Role.USER, contents=[TextContent(text="Hello")]),
-        ChatMessage(role=Role.ASSISTANT, contents=[TextContent(text="World")]),
+        ChatMessage(role=Role.USER, contents=[Content.from_text(text="Hello")]),
+        ChatMessage(role=Role.ASSISTANT, contents=[Content.from_text(text="World")]),
     ]
     event = WorkflowOutputEvent(data=messages, executor_id="complete")
     events = await mapper.convert_event(event, test_request)