ChatMessage

Base classes

Name	Children	Inherits
Generic typing Abstract base class for generic types.	Protocol IO _IteratorToAsyncIterator UpdatableLRU BaseAsyncFileSystem BaseFileSystem Registry Choice BaseConfigOption BasePlugin CombinedEvent ...	object

⋔ Inheritance diagram

graph TD
  94096746884240["messages.ChatMessage"]
  94096692677376["typing.Generic"]
  140089511254496["builtins.object"]
  94096692677376 --> 94096746884240
  140089511254496 --> 94096692677376

🛈 DocStrings

Common message format for all UI types.

Generically typed with: ChatMessage[Type of Content] The type can either be str or a BaseModel subclass.

Source code in src/llmling_agent/messaging/messages.py

@dataclass
class ChatMessage[TContent]:
    """Common message format for all UI types.

    Generically typed with: ChatMessage[Type of Content]
    The type can either be str or a BaseModel subclass.
    """

    content: TContent
    """Message content, typed as TContent (either str or BaseModel)."""

    role: MessageRole
    """Role of the message sender (user/assistant)."""

    metadata: SimpleJsonType = field(default_factory=dict)
    """Additional metadata about the message."""

    timestamp: datetime = field(default_factory=get_now)
    """When this message was created."""

    cost_info: TokenCost | None = None
    """Token usage and costs for this specific message if available."""

    message_id: str = field(default_factory=lambda: str(uuid4()))
    """Unique identifier for this message."""

    conversation_id: str | None = None
    """ID of the conversation this message belongs to."""

    response_time: float | None = None
    """Time it took the LLM to respond."""

    associated_messages: list[ChatMessage[Any]] = field(default_factory=list)
    """List of messages which were generated during the the creation of this messsage."""

    name: str | None = None
    """Display name for the message sender in UI."""

    forwarded_from: list[str] = field(default_factory=list)
    """List of agent names (the chain) that forwarded this message to the sender."""

    provider_details: dict[str, Any] = field(default_factory=dict)
    """Provider specific metadata / extra information."""

    messages: list[ModelMessage] = field(default_factory=list)
    """List of messages which were generated during the the creation of this messsage."""

    usage: RequestUsage = field(default_factory=RequestUsage)
    """Usage information for the request.

    This has a default to make tests easier,
    and to support loading old messages where usage will be missing.
    """

    model_name: str | None = None
    """The name of the model that generated the response."""

    provider_name: str | None = None
    """The name of the LLM provider that generated the response."""

    provider_response_id: str | None = None
    """request ID as specified by the model provider.

    This can be used to track the specific request to the model."""

    finish_reason: FinishReason | None = None
    """Reason the model finished generating the response.

    Normalized to OpenTelemetry values."""

    __repr__ = dataclasses_no_defaults_repr

    @property
    def last_message(self) -> ModelMessage | None:
        """Return the last message from the message history."""
        # The response may not be the very last item if it contained an output tool call.
        # See `CallToolsNode._handle_final_result`.
        for message in reversed(self.messages):
            if isinstance(message, ModelRequest | ModelResponse):
                return message
        return None

    @property
    def parts(self) -> Sequence[ModelRequestPart] | Sequence[ModelResponsePart]:
        """The parts of the last model message."""
        if not self.last_message:
            return []
        return self.last_message.parts

    @property
    def kind(self) -> Literal["request", "response"]:
        """Role of the message."""
        match self.role:
            case "assistant":
                return "response"
            case "user":
                return "request"

    def to_pydantic_ai(self) -> Sequence[ModelMessage]:
        """Convert this message to a Pydantic model."""
        if self.messages:
            return self.messages
        match self.kind:
            case "request":
                return [
                    ModelRequest(
                        parts=self.parts,  # type: ignore
                        instructions=None,
                        run_id=self.message_id,
                    )
                ]
            case "response":
                return [
                    ModelResponse(
                        parts=self.parts,  # type: ignore
                        usage=self.usage,
                        model_name=self.model_name,
                        timestamp=self.timestamp,
                        provider_name=self.provider_name,
                        provider_details=self.provider_details,
                        finish_reason=self.finish_reason,
                        provider_response_id=self.provider_response_id,
                        run_id=self.message_id,
                    )
                ]

    @classmethod
    def user_prompt[TPromptContent: str | Sequence[UserContent] = str](
        cls,
        message: TPromptContent,
        conversation_id: str | None = None,
        instructions: str | None = None,
    ) -> ChatMessage[TPromptContent]:
        """Create a user prompt message."""
        part = UserPromptPart(content=message)
        request = ModelRequest(parts=[part], instructions=instructions)
        return ChatMessage(
            messages=[request],
            role="user",
            content=message,
            conversation_id=conversation_id or str(uuid4()),
        )

    @classmethod
    def from_pydantic_ai[TContentType](
        cls,
        content: TContentType,
        message: ModelMessage,
        conversation_id: str | None = None,
        name: str | None = None,
        forwarded_from: list[str] | None = None,
    ) -> ChatMessage[TContentType]:
        """Convert a Pydantic model to a ChatMessage."""
        match message:
            case ModelRequest(instructions=_instructions, run_id=run_id):
                return ChatMessage(
                    messages=[message],
                    content=content,
                    role="user" if message.kind == "request" else "assistant",
                    message_id=run_id or str(uuid.uuid4()),
                    # instructions=instructions,
                    forwarded_from=forwarded_from or [],
                    name=name,
                )
            case ModelResponse(
                usage=usage,
                model_name=model_name,
                timestamp=timestamp,
                provider_name=provider_name,
                provider_details=provider_details,
                finish_reason=finish_reason,
                provider_response_id=provider_response_id,
                run_id=run_id,
            ):
                return ChatMessage(
                    role="user" if message.kind == "request" else "assistant",
                    content=content,
                    messages=[message],
                    usage=usage,
                    message_id=run_id or str(uuid.uuid4()),
                    conversation_id=conversation_id,
                    model_name=model_name,
                    timestamp=timestamp,
                    provider_name=provider_name,
                    provider_details=provider_details or {},
                    finish_reason=finish_reason,
                    provider_response_id=provider_response_id,
                    name=name,
                    forwarded_from=forwarded_from or [],
                )
            case _ as unreachable:
                assert_never(unreachable)

    @classmethod
    async def from_run_result[OutputDataT](
        cls,
        result: AgentRunResult[OutputDataT],
        *,
        agent_name: str | None = None,
        message_id: str | None = None,
        conversation_id: str | None = None,
        response_time: float,
    ) -> ChatMessage[OutputDataT]:
        """Create a ChatMessage from a PydanticAI run result.

        Args:
            result: The PydanticAI run result
            agent_name: Name of the agent that generated this response
            message_id: Unique message identifier
            conversation_id: Conversation identifier
            response_time: Total time taken for the response

        Returns:
            A ChatMessage with all fields populated from the result
        """
        # Calculate costs
        run_usage = result.usage()
        usage = result.response.usage
        cost_info = await TokenCost.from_usage(
            model=result.response.model_name or "", usage=run_usage
        )

        return ChatMessage[OutputDataT](
            content=result.output,
            role="assistant",
            name=agent_name,
            model_name=result.response.model_name,
            finish_reason=result.response.finish_reason,
            messages=result.new_messages(),
            provider_response_id=result.response.provider_response_id,
            usage=usage,
            provider_name=result.response.provider_name,
            message_id=message_id or str(uuid4()),
            conversation_id=conversation_id,
            cost_info=cost_info,
            response_time=response_time,
            provider_details={},
        )

    def forwarded(self, previous_message: ChatMessage[Any]) -> Self:
        """Create new message showing it was forwarded from another message.

        Args:
            previous_message: The message that led to this one's creation

        Returns:
            New message with updated chain showing the path through previous message
        """
        from_ = [*previous_message.forwarded_from, previous_message.name or "unknown"]
        return replace(self, forwarded_from=from_)

    @property
    def response(self) -> ModelResponse:
        """Return the last response from the message history."""
        # The response may not be the very last item if it contained an output tool call.
        #  See `CallToolsNode._handle_final_result`.
        for message in reversed(self.messages):
            if isinstance(message, ModelResponse):
                return message
        msg = "No response found in the message history"
        raise ValueError(msg)

    def to_request(self) -> Self:
        """Convert this message to a request message.

        If the message is already a request (user role), this is a no-op.
        If it's a response (assistant role), converts response parts to user content.

        Returns:
            New ChatMessage with role='user' and converted parts
        """
        if self.role == "user":
            return self  # Already a request, return as-is

        user_content: list[UserContent] = []  # Convert response parts to user content
        for part in self.parts:
            match part:
                case TextPart(content=text_content):
                    # Text parts become user content strings
                    user_content.append(text_content)
                case FilePart(content=binary_content):
                    # File parts (images, etc.) become user content directly
                    user_content.append(binary_content)
                case BaseToolReturnPart(
                    content=(
                        ImageUrl()
                        | AudioUrl()
                        | DocumentUrl()
                        | VideoUrl()
                        | BinaryContent()
                        | str(),
                    ) as content
                ):
                    user_content.extend(content)
                case BaseToolReturnPart(
                    content=(
                        str()
                        | ImageUrl()
                        | AudioUrl()
                        | DocumentUrl()
                        | VideoUrl()
                        | BinaryContent()
                    ) as content
                ):
                    user_content.append(content)
                case BaseToolReturnPart():
                    # Tool return parts become user content strings
                    user_content.append(part.model_response_str())
                case ToolCallPart():
                    # Tool return parts become user content strings
                    user_content.append(part.args_as_json_str())
                case _:
                    pass

        # Create new UserPromptPart with converted content
        if user_content:
            converted_parts = [UserPromptPart(content=user_content)]
        else:
            converted_parts = [UserPromptPart(content=str(self.content))]

        return replace(
            self,
            role="user",
            messages=[ModelRequest(parts=converted_parts)],
            cost_info=None,
            # TODO: what about message_id?
        )

    @property
    def data(self) -> TContent:
        """Get content as typed data. Provides compat to AgentRunResult."""
        return self.content

    def get_tool_calls(
        self,
        tools: dict[str, Any] | None = None,
        agent_name: str | None = None,
    ) -> list[ToolCallInfo]:
        """Extract tool call information from all messages lazily.

        Args:
            tools: Original Tool set to enrich ToolCallInfos with additional info
            agent_name: Name of the caller
        """
        from uuid import uuid4

        from pydantic_ai import ToolReturnPart

        from llmling_agent.tools import ToolCallInfo

        tools = tools or {}
        parts = [part for message in self.messages for part in message.parts]
        call_parts = {
            part.tool_call_id: part
            for part in parts
            if isinstance(part, ToolCallPart) and part.tool_call_id
        }

        tool_calls = []
        for part in parts:
            if isinstance(part, ToolReturnPart) and part.tool_call_id in call_parts:
                call_part = call_parts[part.tool_call_id]
                tool_info = ToolCallInfo(
                    tool_name=call_part.tool_name,
                    args=call_part.args_as_dict(),
                    agent_name=agent_name or "UNSET",
                    result=part.content,
                    tool_call_id=call_part.tool_call_id or str(uuid4()),
                    timestamp=part.timestamp,
                    agent_tool_name=(
                        t.agent_name if (t := tools.get(part.tool_name)) else None
                    ),
                )
                tool_calls.append(tool_info)

        return tool_calls

    def format(
        self,
        style: FormatStyle = "simple",
        *,
        template: str | None = None,
        variables: dict[str, Any] | None = None,
        show_metadata: bool = False,
        show_costs: bool = False,
    ) -> str:
        """Format message with configurable style.

        Args:
            style: Predefined style or "custom" for custom template
            template: Custom Jinja template (required if style="custom")
            variables: Additional variables for template rendering
            show_metadata: Whether to include metadata
            show_costs: Whether to include cost information

        Raises:
            ValueError: If style is "custom" but no template provided
                    or if style is invalid
        """
        from jinjarope import Environment
        import yamling

        env = Environment(trim_blocks=True, lstrip_blocks=True)
        env.filters["to_yaml"] = yamling.dump_yaml

        match style:
            case "custom":
                if not template:
                    msg = "Custom style requires a template"
                    raise ValueError(msg)
                template_str = template
            case _ if style in MESSAGE_TEMPLATES:
                template_str = MESSAGE_TEMPLATES[style]
            case _:
                msg = f"Invalid style: {style}"
                raise ValueError(msg)
        template_obj = env.from_string(template_str)
        vars_ = {
            **(self.__dict__),
            "show_metadata": show_metadata,
            "show_costs": show_costs,
        }

        if variables:
            vars_.update(variables)

        return template_obj.render(**vars_)

associated_messages class-attribute instance-attribute

associated_messages: list[ChatMessage[Any]] = field(default_factory=list)

List of messages which were generated during the the creation of this messsage.

content instance-attribute

content: TContent

Message content, typed as TContent (either str or BaseModel).

conversation_id class-attribute instance-attribute

conversation_id: str | None = None

ID of the conversation this message belongs to.

cost_info class-attribute instance-attribute

cost_info: TokenCost | None = None

Token usage and costs for this specific message if available.

data property

data: TContent

Get content as typed data. Provides compat to AgentRunResult.

finish_reason class-attribute instance-attribute

finish_reason: FinishReason | None = None

Reason the model finished generating the response.

Normalized to OpenTelemetry values.

forwarded_from class-attribute instance-attribute

forwarded_from: list[str] = field(default_factory=list)

List of agent names (the chain) that forwarded this message to the sender.

kind property

kind: Literal['request', 'response']

Role of the message.

last_message property

last_message: ModelMessage | None

Return the last message from the message history.

message_id class-attribute instance-attribute

message_id: str = field(default_factory=lambda: str(uuid4()))

Unique identifier for this message.

messages class-attribute instance-attribute

messages: list[ModelMessage] = field(default_factory=list)

List of messages which were generated during the the creation of this messsage.

metadata class-attribute instance-attribute

metadata: SimpleJsonType = field(default_factory=dict)

Additional metadata about the message.

model_name class-attribute instance-attribute

model_name: str | None = None

The name of the model that generated the response.

name class-attribute instance-attribute

name: str | None = None

Display name for the message sender in UI.

parts property

parts: Sequence[ModelRequestPart] | Sequence[ModelResponsePart]

The parts of the last model message.

provider_details class-attribute instance-attribute

provider_details: dict[str, Any] = field(default_factory=dict)

Provider specific metadata / extra information.

provider_name class-attribute instance-attribute

provider_name: str | None = None

The name of the LLM provider that generated the response.

provider_response_id class-attribute instance-attribute

provider_response_id: str | None = None

request ID as specified by the model provider.

This can be used to track the specific request to the model.

response property

response: ModelResponse

Return the last response from the message history.

response_time class-attribute instance-attribute

response_time: float | None = None

Time it took the LLM to respond.

role instance-attribute

role: MessageRole

Role of the message sender (user/assistant).

timestamp class-attribute instance-attribute

timestamp: datetime = field(default_factory=get_now)

When this message was created.

usage class-attribute instance-attribute

usage: RequestUsage = field(default_factory=RequestUsage)

Usage information for the request.

This has a default to make tests easier, and to support loading old messages where usage will be missing.

format

format(
    style: FormatStyle = "simple",
    *,
    template: str | None = None,
    variables: dict[str, Any] | None = None,
    show_metadata: bool = False,
    show_costs: bool = False
) -> str

Format message with configurable style.

Parameters:

Name	Type	Description	Default
style	FormatStyle	Predefined style or "custom" for custom template	'simple'
template	str | None	Custom Jinja template (required if style="custom")	None
variables	dict[str, Any] | None	Additional variables for template rendering	None
show_metadata	bool	Whether to include metadata	False
show_costs	bool	Whether to include cost information	False

Raises:

Type	Description
ValueError	If style is "custom" but no template provided or if style is invalid

Source code in src/llmling_agent/messaging/messages.py

def format(
    self,
    style: FormatStyle = "simple",
    *,
    template: str | None = None,
    variables: dict[str, Any] | None = None,
    show_metadata: bool = False,
    show_costs: bool = False,
) -> str:
    """Format message with configurable style.

    Args:
        style: Predefined style or "custom" for custom template
        template: Custom Jinja template (required if style="custom")
        variables: Additional variables for template rendering
        show_metadata: Whether to include metadata
        show_costs: Whether to include cost information

    Raises:
        ValueError: If style is "custom" but no template provided
                or if style is invalid
    """
    from jinjarope import Environment
    import yamling

    env = Environment(trim_blocks=True, lstrip_blocks=True)
    env.filters["to_yaml"] = yamling.dump_yaml

    match style:
        case "custom":
            if not template:
                msg = "Custom style requires a template"
                raise ValueError(msg)
            template_str = template
        case _ if style in MESSAGE_TEMPLATES:
            template_str = MESSAGE_TEMPLATES[style]
        case _:
            msg = f"Invalid style: {style}"
            raise ValueError(msg)
    template_obj = env.from_string(template_str)
    vars_ = {
        **(self.__dict__),
        "show_metadata": show_metadata,
        "show_costs": show_costs,
    }

    if variables:
        vars_.update(variables)

    return template_obj.render(**vars_)

forwarded

forwarded(previous_message: ChatMessage[Any]) -> Self

Create new message showing it was forwarded from another message.

Parameters:

Name	Type	Description	Default
previous_message	ChatMessage[Any]	The message that led to this one's creation	required

Returns:

Type	Description
Self	New message with updated chain showing the path through previous message

Source code in src/llmling_agent/messaging/messages.py

def forwarded(self, previous_message: ChatMessage[Any]) -> Self:
    """Create new message showing it was forwarded from another message.

    Args:
        previous_message: The message that led to this one's creation

    Returns:
        New message with updated chain showing the path through previous message
    """
    from_ = [*previous_message.forwarded_from, previous_message.name or "unknown"]
    return replace(self, forwarded_from=from_)

from_pydantic_ai classmethod

from_pydantic_ai(
    content: TContentType,
    message: ModelMessage,
    conversation_id: str | None = None,
    name: str | None = None,
    forwarded_from: list[str] | None = None,
) -> ChatMessage[TContentType]

Convert a Pydantic model to a ChatMessage.

Source code in src/llmling_agent/messaging/messages.py

@classmethod
def from_pydantic_ai[TContentType](
    cls,
    content: TContentType,
    message: ModelMessage,
    conversation_id: str | None = None,
    name: str | None = None,
    forwarded_from: list[str] | None = None,
) -> ChatMessage[TContentType]:
    """Convert a Pydantic model to a ChatMessage."""
    match message:
        case ModelRequest(instructions=_instructions, run_id=run_id):
            return ChatMessage(
                messages=[message],
                content=content,
                role="user" if message.kind == "request" else "assistant",
                message_id=run_id or str(uuid.uuid4()),
                # instructions=instructions,
                forwarded_from=forwarded_from or [],
                name=name,
            )
        case ModelResponse(
            usage=usage,
            model_name=model_name,
            timestamp=timestamp,
            provider_name=provider_name,
            provider_details=provider_details,
            finish_reason=finish_reason,
            provider_response_id=provider_response_id,
            run_id=run_id,
        ):
            return ChatMessage(
                role="user" if message.kind == "request" else "assistant",
                content=content,
                messages=[message],
                usage=usage,
                message_id=run_id or str(uuid.uuid4()),
                conversation_id=conversation_id,
                model_name=model_name,
                timestamp=timestamp,
                provider_name=provider_name,
                provider_details=provider_details or {},
                finish_reason=finish_reason,
                provider_response_id=provider_response_id,
                name=name,
                forwarded_from=forwarded_from or [],
            )
        case _ as unreachable:
            assert_never(unreachable)

from_run_result async classmethod

from_run_result(
    result: AgentRunResult[OutputDataT],
    *,
    agent_name: str | None = None,
    message_id: str | None = None,
    conversation_id: str | None = None,
    response_time: float
) -> ChatMessage[OutputDataT]

Create a ChatMessage from a PydanticAI run result.

Parameters:

Name	Type	Description	Default
result	AgentRunResult[OutputDataT]	The PydanticAI run result	required
agent_name	str | None	Name of the agent that generated this response	None
message_id	str | None	Unique message identifier	None
conversation_id	str | None	Conversation identifier	None
response_time	float	Total time taken for the response	required

Returns:

Type	Description
ChatMessage[OutputDataT]	A ChatMessage with all fields populated from the result

Source code in src/llmling_agent/messaging/messages.py

@classmethod
async def from_run_result[OutputDataT](
    cls,
    result: AgentRunResult[OutputDataT],
    *,
    agent_name: str | None = None,
    message_id: str | None = None,
    conversation_id: str | None = None,
    response_time: float,
) -> ChatMessage[OutputDataT]:
    """Create a ChatMessage from a PydanticAI run result.

    Args:
        result: The PydanticAI run result
        agent_name: Name of the agent that generated this response
        message_id: Unique message identifier
        conversation_id: Conversation identifier
        response_time: Total time taken for the response

    Returns:
        A ChatMessage with all fields populated from the result
    """
    # Calculate costs
    run_usage = result.usage()
    usage = result.response.usage
    cost_info = await TokenCost.from_usage(
        model=result.response.model_name or "", usage=run_usage
    )

    return ChatMessage[OutputDataT](
        content=result.output,
        role="assistant",
        name=agent_name,
        model_name=result.response.model_name,
        finish_reason=result.response.finish_reason,
        messages=result.new_messages(),
        provider_response_id=result.response.provider_response_id,
        usage=usage,
        provider_name=result.response.provider_name,
        message_id=message_id or str(uuid4()),
        conversation_id=conversation_id,
        cost_info=cost_info,
        response_time=response_time,
        provider_details={},
    )

get_tool_calls

get_tool_calls(
    tools: dict[str, Any] | None = None, agent_name: str | None = None
) -> list[ToolCallInfo]

Extract tool call information from all messages lazily.

Parameters:

Name	Type	Description	Default
tools	dict[str, Any] | None	Original Tool set to enrich ToolCallInfos with additional info	None
agent_name	str | None	Name of the caller	None

Source code in src/llmling_agent/messaging/messages.py

def get_tool_calls(
    self,
    tools: dict[str, Any] | None = None,
    agent_name: str | None = None,
) -> list[ToolCallInfo]:
    """Extract tool call information from all messages lazily.

    Args:
        tools: Original Tool set to enrich ToolCallInfos with additional info
        agent_name: Name of the caller
    """
    from uuid import uuid4

    from pydantic_ai import ToolReturnPart

    from llmling_agent.tools import ToolCallInfo

    tools = tools or {}
    parts = [part for message in self.messages for part in message.parts]
    call_parts = {
        part.tool_call_id: part
        for part in parts
        if isinstance(part, ToolCallPart) and part.tool_call_id
    }

    tool_calls = []
    for part in parts:
        if isinstance(part, ToolReturnPart) and part.tool_call_id in call_parts:
            call_part = call_parts[part.tool_call_id]
            tool_info = ToolCallInfo(
                tool_name=call_part.tool_name,
                args=call_part.args_as_dict(),
                agent_name=agent_name or "UNSET",
                result=part.content,
                tool_call_id=call_part.tool_call_id or str(uuid4()),
                timestamp=part.timestamp,
                agent_tool_name=(
                    t.agent_name if (t := tools.get(part.tool_name)) else None
                ),
            )
            tool_calls.append(tool_info)

    return tool_calls

to_pydantic_ai

to_pydantic_ai() -> Sequence[ModelMessage]

Convert this message to a Pydantic model.

Source code in src/llmling_agent/messaging/messages.py

def to_pydantic_ai(self) -> Sequence[ModelMessage]:
    """Convert this message to a Pydantic model."""
    if self.messages:
        return self.messages
    match self.kind:
        case "request":
            return [
                ModelRequest(
                    parts=self.parts,  # type: ignore
                    instructions=None,
                    run_id=self.message_id,
                )
            ]
        case "response":
            return [
                ModelResponse(
                    parts=self.parts,  # type: ignore
                    usage=self.usage,
                    model_name=self.model_name,
                    timestamp=self.timestamp,
                    provider_name=self.provider_name,
                    provider_details=self.provider_details,
                    finish_reason=self.finish_reason,
                    provider_response_id=self.provider_response_id,
                    run_id=self.message_id,
                )
            ]

to_request

to_request() -> Self

Convert this message to a request message.

If the message is already a request (user role), this is a no-op. If it's a response (assistant role), converts response parts to user content.

Returns:

Type	Description
Self	New ChatMessage with role='user' and converted parts

Source code in src/llmling_agent/messaging/messages.py

def to_request(self) -> Self:
    """Convert this message to a request message.

    If the message is already a request (user role), this is a no-op.
    If it's a response (assistant role), converts response parts to user content.

    Returns:
        New ChatMessage with role='user' and converted parts
    """
    if self.role == "user":
        return self  # Already a request, return as-is

    user_content: list[UserContent] = []  # Convert response parts to user content
    for part in self.parts:
        match part:
            case TextPart(content=text_content):
                # Text parts become user content strings
                user_content.append(text_content)
            case FilePart(content=binary_content):
                # File parts (images, etc.) become user content directly
                user_content.append(binary_content)
            case BaseToolReturnPart(
                content=(
                    ImageUrl()
                    | AudioUrl()
                    | DocumentUrl()
                    | VideoUrl()
                    | BinaryContent()
                    | str(),
                ) as content
            ):
                user_content.extend(content)
            case BaseToolReturnPart(
                content=(
                    str()
                    | ImageUrl()
                    | AudioUrl()
                    | DocumentUrl()
                    | VideoUrl()
                    | BinaryContent()
                ) as content
            ):
                user_content.append(content)
            case BaseToolReturnPart():
                # Tool return parts become user content strings
                user_content.append(part.model_response_str())
            case ToolCallPart():
                # Tool return parts become user content strings
                user_content.append(part.args_as_json_str())
            case _:
                pass

    # Create new UserPromptPart with converted content
    if user_content:
        converted_parts = [UserPromptPart(content=user_content)]
    else:
        converted_parts = [UserPromptPart(content=str(self.content))]

    return replace(
        self,
        role="user",
        messages=[ModelRequest(parts=converted_parts)],
        cost_info=None,
        # TODO: what about message_id?
    )

user_prompt classmethod

user_prompt(
    message: TPromptContent,
    conversation_id: str | None = None,
    instructions: str | None = None,
) -> ChatMessage[TPromptContent]

Create a user prompt message.

Source code in src/llmling_agent/messaging/messages.py

@classmethod
def user_prompt[TPromptContent: str | Sequence[UserContent] = str](
    cls,
    message: TPromptContent,
    conversation_id: str | None = None,
    instructions: str | None = None,
) -> ChatMessage[TPromptContent]:
    """Create a user prompt message."""
    part = UserPromptPart(content=message)
    request = ModelRequest(parts=[part], instructions=instructions)
    return ChatMessage(
        messages=[request],
        role="user",
        content=message,
        conversation_id=conversation_id or str(uuid4()),
    )

Show source on GitHub