Message

Message and content block primitives for agent graphs.

This module defines the core message representation, multimodal content blocks, token usage tracking, and utility functions for agent graph communication.

Classes:

Name	Description
TokenUsages	Tracks token usage statistics for a message or model response.
MediaRef	Reference to media content (image/audio/video/document/data).
AnnotationRef	Reference to annotation metadata.
Message	Represents a message in a conversation, including content, role, metadata, and token usage.

Functions:

Name	Description
generate_id	Generates a message or tool call ID based on DI context and type.

Attributes:

Name	Type	Description
ContentBlock
logger

Attributes

ContentBlock module-attribute

ContentBlock = Annotated[Union[TextBlock, ImageBlock, AudioBlock, VideoBlock, DocumentBlock, DataBlock, ToolCallBlock, ToolResultBlock, ReasoningBlock, AnnotationBlock, ErrorBlock], Field(discriminator='type')]

logger module-attribute

logger = getLogger(__name__)

Classes

AnnotationBlock

Bases: BaseModel

Annotation content block for messages.

Attributes:

Name	Type	Description
type	Literal['annotation']	Block type discriminator.
kind	Literal['citation', 'note']	Kind of annotation.
refs	list[AnnotationRef]	List of annotation references.
spans	list[tuple[int, int]] | None	Spans covered by the annotation.

Source code in pyagenity/utils/message.py

class AnnotationBlock(BaseModel):
    """
    Annotation content block for messages.

    Attributes:
        type (Literal["annotation"]): Block type discriminator.
        kind (Literal["citation", "note"]): Kind of annotation.
        refs (list[AnnotationRef]): List of annotation references.
        spans (list[tuple[int, int]] | None): Spans covered by the annotation.
    """

    type: Literal["annotation"] = "annotation"
    kind: Literal["citation", "note"] = "citation"
    refs: list[AnnotationRef] = Field(default_factory=list)
    spans: list[tuple[int, int]] | None = None

Attributes

kind class-attribute instance-attribute

kind = 'citation'

refs class-attribute instance-attribute

refs = Field(default_factory=list)

spans class-attribute instance-attribute

spans = None

type class-attribute instance-attribute

type = 'annotation'

AnnotationRef

Bases: BaseModel

Reference to annotation metadata (e.g., citation, note).

Attributes:

Name	Type	Description
url	str | None	URL to annotation source.
file_id	str | None	Provider-managed file ID.
page	int | None	Page number (if applicable).
index	int | None	Index within the annotation source.
title	str | None	Title of the annotation.

Source code in pyagenity/utils/message.py

class AnnotationRef(BaseModel):
    """
    Reference to annotation metadata (e.g., citation, note).

    Attributes:
        url (str | None): URL to annotation source.
        file_id (str | None): Provider-managed file ID.
        page (int | None): Page number (if applicable).
        index (int | None): Index within the annotation source.
        title (str | None): Title of the annotation.
    """

    url: str | None = None
    file_id: str | None = None
    page: int | None = None
    index: int | None = None
    title: str | None = None

Attributes

file_id class-attribute instance-attribute

file_id = None

index class-attribute instance-attribute

index = None

page class-attribute instance-attribute

page = None

title class-attribute instance-attribute

title = None

url class-attribute instance-attribute

url = None

AudioBlock

Bases: BaseModel

Audio content block for messages.

Attributes:

Name	Type	Description
type	Literal['audio']	Block type discriminator.
media	MediaRef	Reference to audio media.
transcript	str | None	Transcript of audio.
sample_rate	int | None	Sample rate in Hz.
channels	int | None	Number of audio channels.

Source code in pyagenity/utils/message.py

class AudioBlock(BaseModel):
    """
    Audio content block for messages.

    Attributes:
        type (Literal["audio"]): Block type discriminator.
        media (MediaRef): Reference to audio media.
        transcript (str | None): Transcript of audio.
        sample_rate (int | None): Sample rate in Hz.
        channels (int | None): Number of audio channels.
    """

    type: Literal["audio"] = "audio"
    media: MediaRef
    transcript: str | None = None
    sample_rate: int | None = None
    channels: int | None = None

Attributes

channels class-attribute instance-attribute

channels = None

media instance-attribute

media

sample_rate class-attribute instance-attribute

sample_rate = None

transcript class-attribute instance-attribute

transcript = None

type class-attribute instance-attribute

type = 'audio'

DataBlock

Bases: BaseModel

Data content block for messages.

Attributes:

Name	Type	Description
type	Literal['data']	Block type discriminator.
mime_type	str	MIME type of the data.
data_base64	str | None	Base64-encoded data.
media	MediaRef | None	Reference to associated media.

Source code in pyagenity/utils/message.py

class DataBlock(BaseModel):
    """
    Data content block for messages.

    Attributes:
        type (Literal["data"]): Block type discriminator.
        mime_type (str): MIME type of the data.
        data_base64 (str | None): Base64-encoded data.
        media (MediaRef | None): Reference to associated media.
    """

    type: Literal["data"] = "data"
    mime_type: str
    data_base64: str | None = None
    media: MediaRef | None = None

Attributes

data_base64 class-attribute instance-attribute

data_base64 = None

media class-attribute instance-attribute

media = None

mime_type instance-attribute

mime_type

type class-attribute instance-attribute

type = 'data'

DocumentBlock

Bases: BaseModel

Document content block for messages.

Attributes:

Name	Type	Description
type	Literal['document']	Block type discriminator.
media	MediaRef	Reference to document media.
pages	list[int] | None	List of page numbers.
excerpt	str | None	Excerpt from the document.

Source code in pyagenity/utils/message.py

class DocumentBlock(BaseModel):
    """
    Document content block for messages.

    Attributes:
        type (Literal["document"]): Block type discriminator.
        media (MediaRef): Reference to document media.
        pages (list[int] | None): List of page numbers.
        excerpt (str | None): Excerpt from the document.
    """

    type: Literal["document"] = "document"
    media: MediaRef
    pages: list[int] | None = None
    excerpt: str | None = None

Attributes

excerpt class-attribute instance-attribute

excerpt = None

media instance-attribute

media

pages class-attribute instance-attribute

pages = None

type class-attribute instance-attribute

type = 'document'

ErrorBlock

Bases: BaseModel

Error content block for messages.

Attributes:

Name	Type	Description
type	Literal['error']	Block type discriminator.
message	str	Error message.
code	str | None	Error code.
data	dict[str, Any] | None	Additional error data.

Source code in pyagenity/utils/message.py

class ErrorBlock(BaseModel):
    """
    Error content block for messages.

    Attributes:
        type (Literal["error"]): Block type discriminator.
        message (str): Error message.
        code (str | None): Error code.
        data (dict[str, Any] | None): Additional error data.
    """

    type: Literal["error"] = "error"
    message: str
    code: str | None = None
    data: dict[str, Any] | None = None

Attributes

code class-attribute instance-attribute

code = None

data class-attribute instance-attribute

data = None

message instance-attribute

message

type class-attribute instance-attribute

type = 'error'

ImageBlock

Bases: BaseModel

Image content block for messages.

Attributes:

Name	Type	Description
type	Literal['image']	Block type discriminator.
media	MediaRef	Reference to image media.
alt_text	str | None	Alternative text for accessibility.
bbox	list[float] | None	Bounding box coordinates [x1, y1, x2, y2].

Source code in pyagenity/utils/message.py

class ImageBlock(BaseModel):
    """
    Image content block for messages.

    Attributes:
        type (Literal["image"]): Block type discriminator.
        media (MediaRef): Reference to image media.
        alt_text (str | None): Alternative text for accessibility.
        bbox (list[float] | None): Bounding box coordinates [x1, y1, x2, y2].
    """

    type: Literal["image"] = "image"
    media: MediaRef
    alt_text: str | None = None
    bbox: list[float] | None = None  # [x1,y1,x2,y2] if applicable

Attributes

alt_text class-attribute instance-attribute

alt_text = None

bbox class-attribute instance-attribute

bbox = None

media instance-attribute

media

type class-attribute instance-attribute

type = 'image'

MediaRef

Bases: BaseModel

Reference to media content (image/audio/video/document/data).

Prefer referencing by URL or provider file_id over inlining base64 for large payloads.

Attributes:

Name	Type	Description
kind	Literal['url', 'file_id', 'data']	Type of reference.
url	str | None	URL to media content.
file_id	str | None	Provider-managed file ID.
data_base64	str | None	Base64-encoded data (small payloads only).
mime_type	str | None	MIME type of the media.
size_bytes	int | None	Size in bytes.
sha256	str | None	SHA256 hash of the media.
filename	str | None	Filename of the media.
width	int | None	Image width (if applicable).
height	int | None	Image height (if applicable).
duration_ms	int | None	Duration in milliseconds (if applicable).
page	int | None	Page number (if applicable).

Source code in pyagenity/utils/message.py

class MediaRef(BaseModel):
    """
    Reference to media content (image/audio/video/document/data).

    Prefer referencing by URL or provider file_id over inlining base64 for large payloads.

    Attributes:
        kind (Literal["url", "file_id", "data"]): Type of reference.
        url (str | None): URL to media content.
        file_id (str | None): Provider-managed file ID.
        data_base64 (str | None): Base64-encoded data (small payloads only).
        mime_type (str | None): MIME type of the media.
        size_bytes (int | None): Size in bytes.
        sha256 (str | None): SHA256 hash of the media.
        filename (str | None): Filename of the media.
        width (int | None): Image width (if applicable).
        height (int | None): Image height (if applicable).
        duration_ms (int | None): Duration in milliseconds (if applicable).
        page (int | None): Page number (if applicable).
    """

    kind: Literal["url", "file_id", "data"] = "url"
    url: str | None = None  # http(s) or data: URL
    file_id: str | None = None  # provider-managed ID (e.g., OpenAI/Gemini)
    data_base64: str | None = None  # small payloads only
    mime_type: str | None = None
    size_bytes: int | None = None
    sha256: str | None = None
    filename: str | None = None
    # Media-specific hints
    width: int | None = None
    height: int | None = None
    duration_ms: int | None = None
    page: int | None = None

Attributes

data_base64 class-attribute instance-attribute

data_base64 = None

duration_ms class-attribute instance-attribute

duration_ms = None

file_id class-attribute instance-attribute

file_id = None

filename class-attribute instance-attribute

filename = None

height class-attribute instance-attribute

height = None

kind class-attribute instance-attribute

kind = 'url'

mime_type class-attribute instance-attribute

mime_type = None

page class-attribute instance-attribute

page = None

sha256 class-attribute instance-attribute

sha256 = None

size_bytes class-attribute instance-attribute

size_bytes = None

url class-attribute instance-attribute

url = None

width class-attribute instance-attribute

width = None

Message

Bases: BaseModel

Represents a message in a conversation, including content, role, metadata, and token usage.

Attributes:

Name	Type	Description
message_id	str | int	Unique identifier for the message.
role	Literal['user', 'assistant', 'system', 'tool']	The role of the message sender.
content	list[ContentBlock]	The message content blocks.
delta	bool	Indicates if this is a delta/partial message.
tools_calls	list[dict[str, Any]] | None	Tool call information, if any.
reasoning	str | None	Reasoning or explanation, if any.
timestamp	datetime | None	Timestamp of the message.
metadata	dict[str, Any]	Additional metadata.
usages	TokenUsages | None	Token usage statistics.
raw	dict[str, Any] | None	Raw data, if any.

Example

msg = Message(message_id="abc123", role="user", content=[TextBlock(text="Hello!")])

Methods:

Name	Description
attach_media	Append a media block to the content.
text	Best-effort text extraction from content blocks.
text_message	Create a Message instance from plain text.
tool_message	Create a tool message, optionally marking it as an error.

Source code in pyagenity/utils/message.py

class Message(BaseModel):
    """
    Represents a message in a conversation, including content, role, metadata, and token usage.

    Attributes:
        message_id (str | int): Unique identifier for the message.
        role (Literal["user", "assistant", "system", "tool"]): The role of the message sender.
        content (list[ContentBlock]): The message content blocks.
        delta (bool): Indicates if this is a delta/partial message.
        tools_calls (list[dict[str, Any]] | None): Tool call information, if any.
        reasoning (str | None): Reasoning or explanation, if any.
        timestamp (datetime | None): Timestamp of the message.
        metadata (dict[str, Any]): Additional metadata.
        usages (TokenUsages | None): Token usage statistics.
        raw (dict[str, Any] | None): Raw data, if any.

    Example:
        >>> msg = Message(message_id="abc123", role="user", content=[TextBlock(text="Hello!")])
        {'message_id': 'abc123', 'role': 'user', 'content': [...], ...}
    """

    message_id: str | int = Field(default_factory=lambda: generate_id(None))
    role: Literal["user", "assistant", "system", "tool"]
    content: list[ContentBlock]
    delta: bool = False  # Indicates if this is a delta/partial message
    tools_calls: list[dict[str, Any]] | None = None
    reasoning: str | None = None  # Remove it
    timestamp: datetime | None = Field(default_factory=datetime.now)
    metadata: dict[str, Any] = Field(default_factory=dict)
    usages: TokenUsages | None = None
    raw: dict[str, Any] | None = None

    @classmethod
    def text_message(
        cls,
        content: str,
        role: Literal["user", "assistant", "system", "tool"] = "user",
        message_id: str | None = None,
    ) -> "Message":
        """
        Create a Message instance from plain text.

        Args:
            content (str): The message content.
            role (Literal["user", "assistant", "system", "tool"]): The role of the sender.
            message_id (str | None): Optional message ID.

        Returns:
            Message: The created Message instance.

        Example:
            >>> Message.text_message("Hello!", role="user")
        """
        logger.debug("Creating message from text with role: %s", role)
        return cls(
            message_id=generate_id(message_id),
            role=role,
            content=[TextBlock(text=content)],
            timestamp=datetime.now(),
            metadata={},
        )

    @classmethod
    def tool_message(
        cls,
        content: list[ContentBlock],
        message_id: str | None = None,
        meta: dict[str, Any] | None = None,
    ) -> "Message":
        """
        Create a tool message, optionally marking it as an error.

        Args:
            content (list[ContentBlock]): The message content blocks.
            message_id (str | None): Optional message ID.
            meta (dict[str, Any] | None): Optional metadata.

        Returns:
            Message: The created tool message instance.

        Example:
            >>> Message.tool_message([ToolResultBlock(...)], message_id="tool1")
        """
        res = content
        msg_id = generate_id(message_id)
        return cls(
            message_id=msg_id,
            role="tool",
            content=res,
            timestamp=datetime.now(),
            metadata=meta or {},
        )

    # --- Convenience helpers ---
    def text(self) -> str:
        """
        Best-effort text extraction from content blocks.

        Returns:
            str: Concatenated text from TextBlock and ToolResultBlock outputs.

        Example:
            >>> msg.text()
            'Hello!Result text.'
        """
        parts: list[str] = []
        for block in self.content:
            if isinstance(block, TextBlock):
                parts.append(block.text)
            elif isinstance(block, ToolResultBlock) and isinstance(block.output, str):
                parts.append(block.output)
        return "".join(parts)

    def attach_media(
        self,
        media: MediaRef,
        as_type: Literal["image", "audio", "video", "document"],
    ) -> None:
        """
        Append a media block to the content.

        If content was text, creates a block list. Supports image, audio, video, and document types.

        Args:
            media (MediaRef): Reference to media content.
            as_type (Literal["image", "audio", "video", "document"]): Type of media block to append.

        Returns:
            None

        Raises:
            ValueError: If an unsupported media type is provided.

        Example:
            >>> msg.attach_media(media_ref, as_type="image")
        """
        block: ContentBlock
        if as_type == "image":
            block = ImageBlock(media=media)
        elif as_type == "audio":
            block = AudioBlock(media=media)
        elif as_type == "video":
            block = VideoBlock(media=media)
        elif as_type == "document":
            block = DocumentBlock(media=media)
        else:
            raise ValueError(f"Unsupported media type: {as_type}")

        if isinstance(self.content, str):
            self.content = [TextBlock(text=self.content), block]
        elif isinstance(self.content, list):
            self.content.append(block)
        else:
            self.content = [block]

Attributes

content instance-attribute

content

delta class-attribute instance-attribute

delta = False

message_id class-attribute instance-attribute

message_id = Field(default_factory=lambda: generate_id(None))

metadata class-attribute instance-attribute

metadata = Field(default_factory=dict)

raw class-attribute instance-attribute

raw = None

reasoning class-attribute instance-attribute

reasoning = None

role instance-attribute

role

timestamp class-attribute instance-attribute

timestamp = Field(default_factory=now)

tools_calls class-attribute instance-attribute

tools_calls = None

usages class-attribute instance-attribute

usages = None

Functions

attach_media

attach_media(media, as_type)

Append a media block to the content.

If content was text, creates a block list. Supports image, audio, video, and document types.

Parameters:

Name	Type	Description	Default
media	MediaRef	Reference to media content.	required
as_type	Literal['image', 'audio', 'video', 'document']	Type of media block to append.	required

Returns:

Type	Description
None	None

Raises:

Type	Description
ValueError	If an unsupported media type is provided.

Example

msg.attach_media(media_ref, as_type="image")

Source code in pyagenity/utils/message.py

def attach_media(
    self,
    media: MediaRef,
    as_type: Literal["image", "audio", "video", "document"],
) -> None:
    """
    Append a media block to the content.

    If content was text, creates a block list. Supports image, audio, video, and document types.

    Args:
        media (MediaRef): Reference to media content.
        as_type (Literal["image", "audio", "video", "document"]): Type of media block to append.

    Returns:
        None

    Raises:
        ValueError: If an unsupported media type is provided.

    Example:
        >>> msg.attach_media(media_ref, as_type="image")
    """
    block: ContentBlock
    if as_type == "image":
        block = ImageBlock(media=media)
    elif as_type == "audio":
        block = AudioBlock(media=media)
    elif as_type == "video":
        block = VideoBlock(media=media)
    elif as_type == "document":
        block = DocumentBlock(media=media)
    else:
        raise ValueError(f"Unsupported media type: {as_type}")

    if isinstance(self.content, str):
        self.content = [TextBlock(text=self.content), block]
    elif isinstance(self.content, list):
        self.content.append(block)
    else:
        self.content = [block]

text

text()

Best-effort text extraction from content blocks.

Returns:

Name	Type	Description
str	str	Concatenated text from TextBlock and ToolResultBlock outputs.

Example

msg.text() 'Hello!Result text.'

Source code in pyagenity/utils/message.py

def text(self) -> str:
    """
    Best-effort text extraction from content blocks.

    Returns:
        str: Concatenated text from TextBlock and ToolResultBlock outputs.

    Example:
        >>> msg.text()
        'Hello!Result text.'
    """
    parts: list[str] = []
    for block in self.content:
        if isinstance(block, TextBlock):
            parts.append(block.text)
        elif isinstance(block, ToolResultBlock) and isinstance(block.output, str):
            parts.append(block.output)
    return "".join(parts)

text_message classmethod

text_message(content, role='user', message_id=None)

Create a Message instance from plain text.

Parameters:

Name	Type	Description	Default
content	str	The message content.	required
role	Literal['user', 'assistant', 'system', 'tool']	The role of the sender.	'user'
message_id	str | None	Optional message ID.	None

Returns:

Name	Type	Description
Message	Message	The created Message instance.

Example

Message.text_message("Hello!", role="user")

Source code in pyagenity/utils/message.py

@classmethod
def text_message(
    cls,
    content: str,
    role: Literal["user", "assistant", "system", "tool"] = "user",
    message_id: str | None = None,
) -> "Message":
    """
    Create a Message instance from plain text.

    Args:
        content (str): The message content.
        role (Literal["user", "assistant", "system", "tool"]): The role of the sender.
        message_id (str | None): Optional message ID.

    Returns:
        Message: The created Message instance.

    Example:
        >>> Message.text_message("Hello!", role="user")
    """
    logger.debug("Creating message from text with role: %s", role)
    return cls(
        message_id=generate_id(message_id),
        role=role,
        content=[TextBlock(text=content)],
        timestamp=datetime.now(),
        metadata={},
    )

tool_message classmethod

tool_message(content, message_id=None, meta=None)

Create a tool message, optionally marking it as an error.

Parameters:

Name	Type	Description	Default
content	list[ContentBlock]	The message content blocks.	required
message_id	str | None	Optional message ID.	None
meta	dict[str, Any] | None	Optional metadata.	None

Returns:

Name	Type	Description
Message	Message	The created tool message instance.

Example

Message.tool_message([ToolResultBlock(...)], message_id="tool1")

Source code in pyagenity/utils/message.py

@classmethod
def tool_message(
    cls,
    content: list[ContentBlock],
    message_id: str | None = None,
    meta: dict[str, Any] | None = None,
) -> "Message":
    """
    Create a tool message, optionally marking it as an error.

    Args:
        content (list[ContentBlock]): The message content blocks.
        message_id (str | None): Optional message ID.
        meta (dict[str, Any] | None): Optional metadata.

    Returns:
        Message: The created tool message instance.

    Example:
        >>> Message.tool_message([ToolResultBlock(...)], message_id="tool1")
    """
    res = content
    msg_id = generate_id(message_id)
    return cls(
        message_id=msg_id,
        role="tool",
        content=res,
        timestamp=datetime.now(),
        metadata=meta or {},
    )

ReasoningBlock

Bases: BaseModel

Reasoning content block for messages.

Attributes:

Name	Type	Description
type	Literal['reasoning']	Block type discriminator.
summary	str	Summary of reasoning.
details	list[str] | None	Detailed reasoning steps.

Source code in pyagenity/utils/message.py

class ReasoningBlock(BaseModel):
    """
    Reasoning content block for messages.

    Attributes:
        type (Literal["reasoning"]): Block type discriminator.
        summary (str): Summary of reasoning.
        details (list[str] | None): Detailed reasoning steps.
    """

    type: Literal["reasoning"] = "reasoning"
    summary: str
    details: list[str] | None = None

Attributes

details class-attribute instance-attribute

details = None

summary instance-attribute

summary

type class-attribute instance-attribute

type = 'reasoning'

TextBlock

Bases: BaseModel

Text content block for messages.

Attributes:

Name	Type	Description
type	Literal['text']	Block type discriminator.
text	str	Text content.
annotations	list[AnnotationRef]	List of annotation references.

Source code in pyagenity/utils/message.py

class TextBlock(BaseModel):
    """
    Text content block for messages.

    Attributes:
        type (Literal["text"]): Block type discriminator.
        text (str): Text content.
        annotations (list[AnnotationRef]): List of annotation references.
    """

    type: Literal["text"] = "text"
    text: str
    annotations: list[AnnotationRef] = Field(default_factory=list)

Attributes

annotations class-attribute instance-attribute

annotations = Field(default_factory=list)

text instance-attribute

text

type class-attribute instance-attribute

type = 'text'

TokenUsages

Bases: BaseModel

Tracks token usage statistics for a message or model response.

Attributes:

Name	Type	Description
completion_tokens	int	Number of completion tokens used.
prompt_tokens	int	Number of prompt tokens used.
total_tokens	int	Total tokens used.
reasoning_tokens	int	Reasoning tokens used (optional).
cache_creation_input_tokens	int	Cache creation input tokens (optional).
cache_read_input_tokens	int	Cache read input tokens (optional).
image_tokens	int | None	Image tokens for multimodal models (optional).
audio_tokens	int | None	Audio tokens for multimodal models (optional).

Example

usage = TokenUsages(completion_tokens=10, prompt_tokens=20, total_tokens=30)

Source code in pyagenity/utils/message.py

class TokenUsages(BaseModel):
    """
    Tracks token usage statistics for a message or model response.

    Attributes:
        completion_tokens (int): Number of completion tokens used.
        prompt_tokens (int): Number of prompt tokens used.
        total_tokens (int): Total tokens used.
        reasoning_tokens (int): Reasoning tokens used (optional).
        cache_creation_input_tokens (int): Cache creation input tokens (optional).
        cache_read_input_tokens (int): Cache read input tokens (optional).
        image_tokens (int | None): Image tokens for multimodal models (optional).
        audio_tokens (int | None): Audio tokens for multimodal models (optional).

    Example:
        >>> usage = TokenUsages(completion_tokens=10, prompt_tokens=20, total_tokens=30)
        {'completion_tokens': 10, 'prompt_tokens': 20, 'total_tokens': 30, ...}
    """

    completion_tokens: int
    prompt_tokens: int
    total_tokens: int
    reasoning_tokens: int = 0
    cache_creation_input_tokens: int = 0
    cache_read_input_tokens: int = 0
    # Optional modality-specific usage fields for multimodal models
    image_tokens: int | None = 0
    audio_tokens: int | None = 0

Attributes

audio_tokens class-attribute instance-attribute

audio_tokens = 0

cache_creation_input_tokens class-attribute instance-attribute

cache_creation_input_tokens = 0

cache_read_input_tokens class-attribute instance-attribute

cache_read_input_tokens = 0

completion_tokens instance-attribute

completion_tokens

image_tokens class-attribute instance-attribute

image_tokens = 0

prompt_tokens instance-attribute

prompt_tokens

reasoning_tokens class-attribute instance-attribute

reasoning_tokens = 0

total_tokens instance-attribute

total_tokens

ToolCallBlock

Bases: BaseModel

Tool call content block for messages.

Attributes:

Name	Type	Description
type	Literal['tool_call']	Block type discriminator.
id	str	Tool call ID.
name	str	Tool name.
args	dict[str, Any]	Arguments for the tool call.
tool_type	str | None	Type of tool (e.g., web_search, file_search).

Source code in pyagenity/utils/message.py

class ToolCallBlock(BaseModel):
    """
    Tool call content block for messages.

    Attributes:
        type (Literal["tool_call"]): Block type discriminator.
        id (str): Tool call ID.
        name (str): Tool name.
        args (dict[str, Any]): Arguments for the tool call.
        tool_type (str | None): Type of tool (e.g., web_search, file_search).
    """

    type: Literal["tool_call"] = "tool_call"
    id: str
    name: str
    args: dict[str, Any] = Field(default_factory=dict)
    tool_type: str | None = None  # e.g., web_search, file_search, computer_use

Attributes

args class-attribute instance-attribute

args = Field(default_factory=dict)

id instance-attribute

id

name instance-attribute

name

tool_type class-attribute instance-attribute

tool_type = None

type class-attribute instance-attribute

type = 'tool_call'

ToolResultBlock

Bases: BaseModel

Tool result content block for messages.

Attributes:

Name	Type	Description
type	Literal['tool_result']	Block type discriminator.
call_id	str	Tool call ID.
output	Any	Output from the tool (str, dict, MediaRef, or list of blocks).
is_error	bool	Whether the result is an error.
status	Literal['completed', 'failed'] | None	Status of the tool call.

Source code in pyagenity/utils/message.py

class ToolResultBlock(BaseModel):
    """
    Tool result content block for messages.

    Attributes:
        type (Literal["tool_result"]): Block type discriminator.
        call_id (str): Tool call ID.
        output (Any): Output from the tool (str, dict, MediaRef, or list of blocks).
        is_error (bool): Whether the result is an error.
        status (Literal["completed", "failed"] | None): Status of the tool call.
    """

    type: Literal["tool_result"] = "tool_result"
    call_id: str
    output: Any = None  # str | dict | MediaRef | list[ContentBlock-like]
    is_error: bool = False
    status: Literal["completed", "failed"] | None = None

Attributes

call_id instance-attribute

call_id

is_error class-attribute instance-attribute

is_error = False

output class-attribute instance-attribute

output = None

status class-attribute instance-attribute

status = None

type class-attribute instance-attribute

type = 'tool_result'

VideoBlock

Bases: BaseModel

Video content block for messages.

Attributes:

Name	Type	Description
type	Literal['video']	Block type discriminator.
media	MediaRef	Reference to video media.
thumbnail	MediaRef | None	Reference to thumbnail image.

Source code in pyagenity/utils/message.py

class VideoBlock(BaseModel):
    """
    Video content block for messages.

    Attributes:
        type (Literal["video"]): Block type discriminator.
        media (MediaRef): Reference to video media.
        thumbnail (MediaRef | None): Reference to thumbnail image.
    """

    type: Literal["video"] = "video"
    media: MediaRef
    thumbnail: MediaRef | None = None

Attributes

media instance-attribute

media

thumbnail class-attribute instance-attribute

thumbnail = None

type class-attribute instance-attribute

type = 'video'

Functions

generate_id

generate_id(default_id)

Generate a message or tool call ID based on DI context and type.

Parameters:

Name	Type	Description	Default
default_id	str | int | None	Default ID to use if provided and matches type.	required

Returns:

Type	Description
str | int	str | int: Generated or provided ID, type determined by DI context.

Example

generate_id("abc123") 'abc123' generate_id(None) 'a-uuid-string'

Source code in pyagenity/utils/message.py

def generate_id(default_id: str | int | None) -> str | int:
    """
    Generate a message or tool call ID based on DI context and type.

    Args:
        default_id (str | int | None): Default ID to use if provided and matches type.

    Returns:
        str | int: Generated or provided ID, type determined by DI context.

    Raises:
        None

    Example:
        >>> generate_id("abc123")
        'abc123'
        >>> generate_id(None)
        'a-uuid-string'
    """
    id_type = InjectQ.get_instance().try_get("generated_id_type", "string")
    generated_id = InjectQ.get_instance().try_get("generated_id", None)

    # if user provided an awaitable, resolve it
    if isinstance(generated_id, Awaitable):

        async def wait_for_id():
            return await generated_id

        generated_id = asyncio.run(wait_for_id())

    if generated_id:
        return generated_id

    if default_id:
        if id_type == "string" and isinstance(default_id, str):
            return default_id
        if id_type in ("int", "bigint") and isinstance(default_id, int):
            return default_id

    # if not matched or default_id is None, generate new id
    logger.debug(
        "Generating new id of type: %s. Default ID not provided or not matched %s",
        id_type,
        default_id,
    )

    if id_type == "int":
        return uuid4().int >> 32
    if id_type == "bigint":
        return uuid4().int >> 64
    return str(uuid4())