agui_converters

Class info

Classes

Name	Children	Inherits
ToolCallStartEvent llmling_agent.agent.events Event indicating a tool call has started with rich ACP metadata.		object
ToolCallProgressEvent llmling_agent.agent.events Enhanced tool call progress event with rich content support.		object

🛈 DocStrings

AG-UI to native event converters.

This module provides conversion from AG-UI protocol events to native llmling-agent streaming events, enabling AGUIAgent to yield the same event types as native agents.

agui_to_native_event

agui_to_native_event(event: Event) -> RichAgentStreamEvent[Any] | None

Convert AG-UI event to native streaming event.

Parameters:

Name	Type	Description	Default
event	Event	AG-UI Event from SSE stream	required

Returns:

Type	Description
RichAgentStreamEvent[Any] | None	Corresponding native event, or None if no mapping exists

Source code in src/llmling_agent/agent/agui_converters.py

def agui_to_native_event(event: Event) -> RichAgentStreamEvent[Any] | None:  # noqa: PLR0911
    """Convert AG-UI event to native streaming event.

    Args:
        event: AG-UI Event from SSE stream

    Returns:
        Corresponding native event, or None if no mapping exists
    """
    match event:
        # Text message content -> PartDeltaEvent with TextPartDelta
        case TextMessageContentEvent(delta=delta):
            return PartDeltaEvent(index=0, delta=TextPartDelta(content_delta=delta))

        # Text message chunks -> PartDeltaEvent with TextPartDelta
        case TextMessageChunkEvent(delta=delta) if delta:
            return PartDeltaEvent(index=0, delta=TextPartDelta(content_delta=delta))

        # Thinking message content -> PartDeltaEvent with ThinkingPartDelta
        case ThinkingTextMessageContentEvent(delta=delta):
            return PartDeltaEvent(index=0, delta=ThinkingPartDelta(content_delta=delta))

        # Message start/end events - could be used for metadata tracking
        case TextMessageStartEvent() | TextMessageEndEvent():
            return None

        case ThinkingTextMessageStartEvent() | ThinkingTextMessageEndEvent():
            return None

        # Tool call start -> NativeToolCallStartEvent
        case ToolCallStartEvent(
            tool_call_id=tool_call_id, tool_call_name=tool_name, parent_message_id=_
        ):
            return NativeToolCallStartEvent(
                tool_call_id=tool_call_id,
                tool_name=tool_name,
                title=tool_name,
                kind="other",
                content=[],
                locations=[],
                raw_input={},
            )

        # Tool call chunks -> NativeToolCallStartEvent if new, else progress
        case ToolCallChunkEvent(
            tool_call_id=tool_call_id,
            tool_call_name=tool_name,
            parent_message_id=_,
            delta=_,
        ) if tool_call_id and tool_name:
            # This is a new tool call
            return NativeToolCallStartEvent(
                tool_call_id=tool_call_id,
                tool_name=tool_name,
                title=tool_name,
                kind="other",
                content=[],
                locations=[],
                raw_input={},
            )

        # Tool call args accumulation - treat as progress
        case ToolCallArgsEvent(tool_call_id=tool_call_id, delta=_):
            return ToolCallProgressEvent(
                tool_call_id=tool_call_id,
                status="in_progress",
            )

        # Tool call result
        case ToolCallResultEvent(
            tool_call_id=tool_call_id,
            content=content,
            message_id=_,
        ):
            return ToolCallProgressEvent(
                tool_call_id=tool_call_id,
                status="completed",
                message=content,
            )

        # Tool call end
        case ToolCallEndEvent(tool_call_id=tool_call_id):
            return ToolCallProgressEvent(
                tool_call_id=tool_call_id,
                status="completed",
            )

        # Activity events - could map to plan or custom events
        case ActivitySnapshotEvent(message_id=_, activity_type=_, content=_, replace=_):
            # Could be used for plan updates if activity_type == "PLAN"
            return None

        case ActivityDeltaEvent(message_id=_, activity_type=_, patch=_):
            return None

        case _:
            return None

extract_text_from_event

extract_text_from_event(event: Event) -> str | None

Extract plain text content from an AG-UI event.

Parameters:

Name	Type	Description	Default
event	Event	AG-UI Event	required

Returns:

Type	Description
str | None	Text content if this is a text-bearing event, None otherwise

Source code in src/llmling_agent/agent/agui_converters.py

def extract_text_from_event(event: Event) -> str | None:
    """Extract plain text content from an AG-UI event.

    Args:
        event: AG-UI Event

    Returns:
        Text content if this is a text-bearing event, None otherwise
    """
    match event:
        case TextMessageContentEvent(delta=delta):
            return delta
        case TextMessageChunkEvent(delta=delta) if delta:
            return delta
        case ThinkingTextMessageContentEvent(delta=delta):
            return delta
        case _:
            return None

is_text_event

is_text_event(event: Event) -> bool

Check if this event contains text content.

Source code in src/llmling_agent/agent/agui_converters.py

def is_text_event(event: Event) -> bool:
    """Check if this event contains text content."""
    return extract_text_from_event(event) is not None