Bases: BaseEvent
Carries the raw xAI GetChatCompletionResponse proto for multi-turn round-trip.
The xai-sdk Chat object is stateful: rebuilding an assistant turn with tool_calls requires passing the original Response proto back via chat.append(response). The proto is the only canonical source — there is no public helper that constructs an assistant message with tool_calls from primitives. Persisted (NOT transient) so it survives history storage.
Source code in autogen/beta/events/base.py
| def __init__(self, *args: Any, **kwargs: Any) -> None:
# MRO walk: map positional args and collect defaults.
positional_names: list[str] = []
defaults: dict[str, Any] = {}
seen: set[str] = set()
for klass in reversed(type(self).__mro__):
for name, f in getattr(klass, "_event_fields_", {}).items():
if name not in seen:
if not f.kw_only:
positional_names.append(name)
if name not in kwargs:
default = f.get_default()
if default is not Ellipsis:
defaults[name] = default
seen.add(name)
if args:
if len(args) > len(positional_names):
raise TypeError(
f"{type(self).__name__}() takes {len(positional_names)} "
f"positional argument(s) but {len(args)} were given"
)
for name, value in zip(positional_names, args):
if name in kwargs:
raise TypeError(f"{type(self).__name__}() got multiple values for argument '{name}'")
kwargs[name] = value
defaults.pop(name, None)
# Apply defaults first, then user-provided kwargs so that
# property setters (e.g. content -> _content) aren't overwritten
# by a field default applied afterwards.
for key, value in defaults.items():
setattr(self, key, value)
for key, value in kwargs.items():
setattr(self, key, value)
|
class-attribute instance-attribute
proto_bytes = Field(repr=False)
class-attribute instance-attribute
model = Field(default=None, compare=False)
class-attribute instance-attribute
finish_reason = Field(default=None, compare=False)
class-attribute instance-attribute
created_at = Field(default_factory=time, compare=False, repr=False)
classmethod
Source code in autogen/beta/config/xai/events.py
| @classmethod
def from_response(cls, response: XAIResponse) -> "XAIAssistantEvent":
# finish_reason comes back as "FINISH_REASON_STOP" / proto enum — normalise
# to "stop" so this metadata matches ModelResponse.finish_reason.
fr = response.finish_reason
finish_reason: str | None = None
if fr:
name = sample_pb2.FinishReason.Name(fr) if isinstance(fr, int) else str(fr)
finish_reason = name.removeprefix("FINISH_REASON_").removeprefix("REASON_").lower() or None
return cls(
proto_bytes=response.proto.SerializeToString(),
model=response.proto.model or None,
finish_reason=finish_reason,
)
|
Serialize this event to a JSON-compatible dictionary.
Source code in autogen/beta/events/base.py
| def to_dict(self) -> dict[str, Any]:
"""Serialize this event to a JSON-compatible dictionary."""
return event_to_dict(self)
|
classmethod
Reconstruct an event from a serialized dictionary.
Filters input to only fields known by this class (via MRO Field descriptors), then constructs via cls(**filtered).
Source code in autogen/beta/events/base.py
| @classmethod
def from_dict(cls, data: dict[str, Any]) -> "BaseEvent":
"""Reconstruct an event from a serialized dictionary.
Filters input to only fields known by this class (via MRO Field
descriptors), then constructs via ``cls(**filtered)``.
"""
# Collect known field names across the MRO
known_fields: set[str] = set()
for klass in cls.__mro__:
for name in getattr(klass, "_event_fields_", {}):
known_fields.add(name)
# Deserialize nested events/special types, then filter to known fields
deserialized = deserialize_payload(data)
filtered = {k: v for k, v in deserialized.items() if k in known_fields}
return cls(**filtered)
|