Python: Preserve the AG-UI RUN_ERROR code in AGUIChatClient

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

@@ -243,13 +243,16 @@ 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")
+        # Keep the server's code so callers can tell failures apart; fall back when the event has none.
+        code = event.get("code")
+        error_code = str(code) if 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

@@ -703,6 +703,35 @@ async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str
         assert first_content.text == "Hello"
         assert second_content.text == " world"
 
+    async def test_get_response_streaming_preserves_run_error_code(self, monkeypatch: MonkeyPatch) -> None:
+        """Test that the server's RUN_ERROR code reaches the caller."""
+        mock_events = [
+            {"type": "RUN_STARTED", "threadId": "thread_1", "runId": "run_1"},
+            {"type": "RUN_ERROR", "message": "Approval not found", "code": "APPROVAL_RESUME_NOT_FOUND"},
+        ]
+
+        async def mock_post_run(*args: object, **kwargs: Any) -> AsyncGenerator[dict[str, Any], None]:
+            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 message"])]
+
+        stream = client.inner_get_response(messages=messages, stream=True, options=ChatOptions())
+        assert isinstance(stream, ResponseStream)
+        errors = [
+            content
+            async for update in stream
+            for content in cast(ChatResponseUpdate, update).contents
+            if content.type == "error"
+        ]
+
+        assert len(errors) == 1
+        assert errors[0].message == "Approval not found"
+        assert errors[0].error_code == "APPROVAL_RESUME_NOT_FOUND"
+
     async def test_get_response_non_streaming(self, monkeypatch: MonkeyPatch) -> None:
         """Test non-streaming response method."""
         mock_events = [

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

@@ -347,6 +347,23 @@ 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_server_code(self) -> None:
+        """Test that a RUN_ERROR code from the server is kept as the error code."""
+        converter = AGUIEventConverter()
+
+        event = {
+            "type": "RUN_ERROR",
+            "message": "Function invocation is disabled",
+            "code": "APPROVAL_INVOCATION_DISABLED",
+        }
+
+        update = converter.convert_event(event)
+
+        assert update is not 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()