Python: preserve server RUN_ERROR code in AG-UI event conversion

python/packages/ag-ui/agent_framework_ag_ui/_event_converters.py

@@ -305,13 +305,15 @@ def _handle_run_finished(self, event: dict[str, Any]) -> ChatResponseUpdate:
     def _handle_run_error(self, event: dict[str, Any]) -> ChatResponseUpdate:
         """Handle RUN_ERROR event."""
         error_message = event.get("message", "Unknown error")
+        raw_code = event.get("code")
+        error_code = str(raw_code) if raw_code else "RUN_ERROR"
 
         return ChatResponseUpdate(
             role="assistant",
             contents=[
                 Content.from_error(
                     message=error_message,
-                    error_code="RUN_ERROR",
+                    error_code=error_code,
                 )
             ],
             additional_properties={

python/packages/ag-ui/tests/ag_ui/test_ag_ui_client.py

@@ -1703,3 +1703,36 @@ async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str
         )
 
         assert response is not None
+
+    async def test_run_error_event_preserves_specific_code(self, monkeypatch: MonkeyPatch) -> None:
+        """AGUIChatClient preserves the specific error code from RUN_ERROR event."""
+        mock_events = [
+            {"type": "RUN_STARTED", "threadId": "thread_1", "runId": "run_1"},
+            {
+                "type": "RUN_ERROR",
+                "threadId": "thread_1",
+                "runId": "run_1",
+                "message": "Function invocation is disabled",
+                "code": "APPROVAL_INVOCATION_DISABLED",
+            },
+        ]
+
+        async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str, Any], None]:
+            del args, kwargs
+            for event in mock_events:
+                yield event
+
+        client = StubAGUIChatClient(endpoint="http://localhost:8888/")
+        monkeypatch.setattr(client.http_service, "post_run", mock_post_run)
+
+        messages = [Message(role="user", contents=["Test"])]
+        stream = client.get_response(messages=messages, stream=True)
+        assert isinstance(stream, ResponseStream)
+        updates: list[ChatResponseUpdate] = []
+        async for update in stream:
+            updates.append(cast(ChatResponseUpdate, update))
+
+        error_contents = [c for update in updates for c in update.contents if c.type == "error"]
+        assert len(error_contents) == 1
+        assert error_contents[0].message == "Function invocation is disabled"
+        assert error_contents[0].error_code == "APPROVAL_INVOCATION_DISABLED"

python/packages/ag-ui/tests/ag_ui/test_event_converters.py

@@ -351,6 +351,27 @@ def test_run_error_event(self) -> None:
         assert update.contents[0].message == "Connection timeout"
         assert update.contents[0].error_code == "RUN_ERROR"
 
+    def test_run_error_event_preserves_specific_code(self) -> None:
+        """Test conversion of RUN_ERROR event preserves specific error code."""
+        converter = AGUIEventConverter()
+        converter.thread_id = "thread_123"
+        converter.run_id = "run_456"
+
+        event = {
+            "type": "RUN_ERROR",
+            "message": "Function invocation is disabled",
+            "code": "APPROVAL_INVOCATION_DISABLED",
+        }
+
+        update = converter.convert_event(event)
+
+        assert update is not None
+        assert update.role == "assistant"
+        assert update.finish_reason is None
+        assert len(update.contents) == 1
+        assert update.contents[0].message == "Function invocation is disabled"
+        assert update.contents[0].error_code == "APPROVAL_INVOCATION_DISABLED"
+
     def test_unknown_event_type(self) -> None:
         """Test handling of unknown event types."""
         converter = AGUIEventConverter()