def process_events()

in src/frontends/streamlit_adk/frontend/utils/stream_handler.py [0:0]
50 lines of code
14 McCabe index (conditional complexity)

    def process_events(self) -> None:
        """Process events from the stream, handling each event type appropriately."""
        messages = self.st.session_state.user_chats[
            self.st.session_state["session_id"]
        ]["messages"]

        session = self.st.session_state["session_id"]
        stream = self.client.stream_messages(
            data={
                "message": messages[-1]["content"],
                "events": messages[:-1],
                "user_id": self.st.session_state["user_id"],
                "session_id": self.st.session_state["session_id"],
            }
        )

        for message in stream:
            event = Event.model_validate(message)

            # Skip processing if event has no content or parts
            if not event.content or not event.content.parts:
                continue

            # Process each part in the event content
            for part in event.content.parts:
                # Case 1: Process function/tool calls
                if part.function_call:
                    # Extract tool call information
                    tool_call = {
                        "name": part.function_call.name,
                        "args": part.function_call.args,
                        "id": part.function_call.id,
                    }

                    # Track tool calls and update UI
                    self.tool_calls.append(event.model_dump())
                    tool_message = (
                        f"\n\nCalling tool: `{tool_call['name']}` "
                        f"with args: `{tool_call['args']}`"
                    )
                    self.stream_handler.new_status(tool_message)

                    # Add to conversation history
                    self.st.session_state.user_chats[session]["messages"].append(
                        event.model_dump(mode="json")
                    )
                # Case 2: Process function/tool responses
                elif part.function_response:
                    # Extract response content
                    content = str(part.function_response.response)

                    # Track responses and update UI
                    self.tool_calls.append(event.model_dump())
                    response_message = f"\n\nTool response: `{content}`"
                    self.stream_handler.new_status(response_message)

                    # Add to conversation history
                    self.st.session_state.user_chats[session]["messages"].append(
                        event.model_dump(mode="json")
                    )
                # Case 3: Process text responses
                elif part.text:
                    if event.is_final_response():
                        # Store the final response in conversation history
                        self.st.session_state.user_chats[session]["messages"].append(
                            event.model_dump(mode="json")
                        )
                        # Save the invocation ID - it might be used by the feedback functionality
                        self.st.session_state["invocation_id"] = event.invocation_id
                    else:
                        # For streaming chunks, accumulate text and update UI
                        self.final_content += part.text
                        self.stream_handler.new_token(part.text)