feat(python): add optional subagent_id to message types

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
ag-ui-protocol · mme · Jul 7, 2026 · Jul 7, 2026 · Jul 7, 2026 · Jul 7, 2026
commit 1bb7368ccb43e31b701a49fb495ebde58c30c9db
@@ -47,6 +47,7 @@ class BaseMessage(ConfiguredBaseModel):
     content: Optional[str] = None
     name: Optional[str] = None
     encrypted_value: Optional[str] = None
+    subagent_id: Optional[str] = None
 
 
 class DeveloperMessage(BaseMessage):
@@ -199,6 +200,7 @@ class ToolMessage(ConfiguredBaseModel):
     tool_call_id: str
     error: Optional[str] = None
     encrypted_value: Optional[str] = None
+    subagent_id: Optional[str] = None
 
 
 class ActivityMessage(ConfiguredBaseModel):
@@ -210,6 +212,7 @@ class ActivityMessage(ConfiguredBaseModel):
     role: Literal["activity"] = "activity"  # pyright: ignore[reportIncompatibleVariableOverride]
     activity_type: str
     content: Dict[str, Any]
+    subagent_id: Optional[str] = None
 
 
 class ReasoningMessage(ConfiguredBaseModel):
@@ -221,6 +224,7 @@ class ReasoningMessage(ConfiguredBaseModel):
     role: Literal["reasoning"] = "reasoning"  # pyright: ignore[reportIncompatibleVariableOverride]
     content: str
     encrypted_value: Optional[str] = None
+    subagent_id: Optional[str] = None
 
 
 Message = Annotated[

@@ -0,0 +1,24 @@
+import unittest
+
+from ag_ui.core.types import AssistantMessage, ToolMessage, ReasoningMessage
+
+
+class TestSubagentMessageAttribution(unittest.TestCase):
+    def test_assistant_message_accepts_subagent_id(self):
+        msg = AssistantMessage(id="m1", role="assistant", content="hi", subagent_id="sub-1")
+        self.assertEqual(msg.subagent_id, "sub-1")
+        self.assertEqual(msg.model_dump(by_alias=True)["subagentId"], "sub-1")
+
+    def test_tool_and_reasoning_messages_accept_subagent_id(self):
+        tool = ToolMessage(id="t1", role="tool", content="ok", tool_call_id="tc1", subagent_id="sub-2")
+        self.assertEqual(tool.subagent_id, "sub-2")
+        reasoning = ReasoningMessage(id="r1", role="reasoning", content="x", subagent_id="sub-3")
+        self.assertEqual(reasoning.subagent_id, "sub-3")
+
+    def test_subagent_id_optional(self):
+        msg = AssistantMessage(id="m2", role="assistant", content="hi")
+        self.assertIsNone(msg.subagent_id)
+
+
+if __name__ == "__main__":
+    unittest.main()