codekingpro/portable-devtools

1from __future__ import annotations2 3import time4from typing import Any, cast5 6from langgraph.stream._types import ProtocolEvent, _ProtocolEventParams7from langgraph.types import StreamPart8 9 10def convert_to_protocol_event(part: StreamPart) -> ProtocolEvent:11    """Convert a v2 StreamPart to a ProtocolEvent.12 13    Args:14        part: A stream part with keys `type`, `ns`, `data`, and15            optionally `interrupts` (present on values events).16 17    Returns:18        The equivalent ProtocolEvent.19    """20    part_dict = cast(dict[str, Any], part)21    params: _ProtocolEventParams = {22        "namespace": list(part_dict["ns"]),23        "timestamp": int(time.time() * 1000),24        "data": part_dict["data"],25    }26    if "interrupts" in part_dict:27        params["interrupts"] = part_dict["interrupts"]28    return {29        "type": "event",30        "method": part_dict["type"],31        "params": params,32    }33