Compatible with Claude API streaming output

Author: LuckyYC

st_engine/engine/request_processor.py

@@ -65,6 +65,23 @@ def get_field_value(data: Dict[str, Any], path: str) -> Any:
         except (KeyError, IndexError, TypeError, ValueError):
             return ""
 
+    @staticmethod
+    def is_sse_event_line(chunk_str: str) -> bool:
+        """Check if the line is an SSE event line (not data line) that should be skipped.
+        Only skip event control lines, not data lines that start with 'data: '.
+        """
+        chunk_str = chunk_str.strip()
+
+        # Check for SSE event control prefixes that should be skipped
+        # Only 'event:', 'id:', 'retry:' are control lines, 'data:' contains actual data
+        control_prefixes = ["event:", "event: ", "id:", "id: ", "retry:", "retry: "]
+
+        for prefix in control_prefixes:
+            if chunk_str.startswith(prefix):
+                return True
+
+        return False
+
     @staticmethod
     def remove_chunk_prefix(chunk_str: str, field_mapping: FieldMapping) -> str:
         """Remove prefix from chunk string based on field mapping configuration."""
@@ -120,6 +137,12 @@ def extract_metrics_from_chunk(
             )
 
             if metrics.usage and isinstance(metrics.usage, dict):
+                # has_prompt_tokens = any(
+                #     "prompt" in key and value not in (None, 0)
+                #     for key, value in metrics.usage.items()
+                #     if isinstance(value, (int, float))
+                # )
+
                 has_completion_tokens = any(
                     "completion" in key and value not in (None, 0)
                     for key, value in metrics.usage.items()
@@ -131,6 +154,7 @@ def extract_metrics_from_chunk(
                     for key, value in metrics.usage.items()
                     if isinstance(value, (int, float))
                 )
+
                 if has_completion_tokens and has_total_tokens:
                     usage_extracted = True
 
@@ -238,6 +262,9 @@ def process_stream_chunk(
         if not chunk_str:
             return False, None, metrics
 
+        if StreamProcessor.is_sse_event_line(chunk_str):
+            return False, None, metrics
+
         # Remove prefix if present
         processed_chunk = StreamProcessor.remove_chunk_prefix(chunk_str, field_mapping)