Skip to content

Models

EventEnvelopeWithMetadata #

Bases: BaseModel

Client readable representation of an Event. Includes class metadata in order to support matching event types semantically in an extendable manner (e.g. "StartEvent", "StopEvent", etc.).

Parameters:

Name Type Description Default
value dict[str, Any]
required
qualified_name str | None
required
type str
required
types list[str] | None
required
Source code in llama_agents/client/protocol/serializable_events.py
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
class EventEnvelopeWithMetadata(BaseModel):
    """
    Client readable representation of an Event. Includes class metadata in order to support
    matching event types semantically in an extendable manner (e.g. "StartEvent", "StopEvent", etc.).
    """

    value: dict[str, Any]

    # deprecated, use type instead
    qualified_name: str | None

    # New metadata
    type: str
    types: list[str] | None

    def load_event(
        self,
        registry: Sequence[type[Event]] = (),
        serializer: JsonSerializer | None = None,
    ) -> Event:
        """
        Load the event data using the given serializer when provided.
        A non-empty registry also permits framework event classes.
        With neither, a default serializer resolves classes by qualified name.
        """
        registry_lookup = {e.__name__: e for e in registry}
        if serializer is None:
            if registry:
                registry_lookup = {
                    **{event.__name__: event for event in FRAMEWORK_EVENT_TYPES},
                    **registry_lookup,
                }
                serializer = JsonSerializer(
                    allowed_types=list(registry_lookup.values())
                )
            else:
                serializer = JsonSerializer()
        as_event_envelope = EventEnvelope(
            value=self.value, type=self.type, qualified_name=self.qualified_name
        ).model_dump()
        return EventEnvelope.parse(
            client_data=as_event_envelope,
            registry=registry_lookup,
            serializer=serializer,
        )

    @classmethod
    def from_event(
        cls, event: Event, include_qualified_name: bool = True
    ) -> EventEnvelopeWithMetadata:
        """
        Build a backward-compatible envelope for an Event, preserving existing
        fields (e.g., qualified_name, value) while adding metadata useful for
        type-safe clients.

        """
        # Start with the existing JSON-serializable structure
        value = event.model_dump(mode="json")

        envelope = EventEnvelopeWithMetadata(
            value=value,
            qualified_name=_get_qualified_name(type(event))
            if include_qualified_name
            else None,
            types=_get_event_subtypes(type(event)),
            type=type(event).__name__,
        )
        return envelope

load_event #

load_event(registry: Sequence[type[Event]] = (), serializer: JsonSerializer | None = None) -> Event

Load the event data using the given serializer when provided. A non-empty registry also permits framework event classes. With neither, a default serializer resolves classes by qualified name.

Source code in llama_agents/client/protocol/serializable_events.py
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def load_event(
    self,
    registry: Sequence[type[Event]] = (),
    serializer: JsonSerializer | None = None,
) -> Event:
    """
    Load the event data using the given serializer when provided.
    A non-empty registry also permits framework event classes.
    With neither, a default serializer resolves classes by qualified name.
    """
    registry_lookup = {e.__name__: e for e in registry}
    if serializer is None:
        if registry:
            registry_lookup = {
                **{event.__name__: event for event in FRAMEWORK_EVENT_TYPES},
                **registry_lookup,
            }
            serializer = JsonSerializer(
                allowed_types=list(registry_lookup.values())
            )
        else:
            serializer = JsonSerializer()
    as_event_envelope = EventEnvelope(
        value=self.value, type=self.type, qualified_name=self.qualified_name
    ).model_dump()
    return EventEnvelope.parse(
        client_data=as_event_envelope,
        registry=registry_lookup,
        serializer=serializer,
    )

from_event classmethod #

from_event(event: Event, include_qualified_name: bool = True) -> EventEnvelopeWithMetadata

Build a backward-compatible envelope for an Event, preserving existing fields (e.g., qualified_name, value) while adding metadata useful for type-safe clients.

Source code in llama_agents/client/protocol/serializable_events.py
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
@classmethod
def from_event(
    cls, event: Event, include_qualified_name: bool = True
) -> EventEnvelopeWithMetadata:
    """
    Build a backward-compatible envelope for an Event, preserving existing
    fields (e.g., qualified_name, value) while adding metadata useful for
    type-safe clients.

    """
    # Start with the existing JSON-serializable structure
    value = event.model_dump(mode="json")

    envelope = EventEnvelopeWithMetadata(
        value=value,
        qualified_name=_get_qualified_name(type(event))
        if include_qualified_name
        else None,
        types=_get_event_subtypes(type(event)),
        type=type(event).__name__,
    )
    return envelope

HandlerData #

Bases: BaseModel

Parameters:

Name Type Description Default
handler_id str
required
workflow_name str
required
run_id str | None
required
error str | None
required
result EventEnvelopeWithMetadata | None
required
result_unreadable bool
False
status Literal['running', 'completed', 'failed', 'cancelled']
required
started_at str
required
updated_at str | None
required
completed_at str | None
required
Source code in llama_agents/client/protocol/__init__.py
20
21
22
23
24
25
26
27
28
29
30
class HandlerData(BaseModel):
    handler_id: str
    workflow_name: str
    run_id: str | None
    error: str | None
    result: EventEnvelopeWithMetadata | None
    result_unreadable: bool = False
    status: Status
    started_at: str
    updated_at: str | None
    completed_at: str | None

HandlersListResponse #

Bases: BaseModel

Parameters:

Name Type Description Default
handlers list[HandlerData]
required
Source code in llama_agents/client/protocol/__init__.py
33
34
class HandlersListResponse(BaseModel):
    handlers: list[HandlerData]

SendEventResponse #

Bases: BaseModel

Parameters:

Name Type Description Default
status Literal['sent']
required
Source code in llama_agents/client/protocol/__init__.py
45
46
class SendEventResponse(BaseModel):
    status: Literal["sent"]

CancelHandlerResponse #

Bases: BaseModel

Parameters:

Name Type Description Default
status Literal['deleted', 'cancelled']
required
Source code in llama_agents/client/protocol/__init__.py
49
50
class CancelHandlerResponse(BaseModel):
    status: Literal["deleted", "cancelled"]