ant_ai.core.message

AnyMessage

AnyMessage = Annotated[
    Message | ToolCallMessage | ToolCallResultMessage,
    Field(discriminator="kind"),
]

Type alias for any message type in the system, discriminated by the _kind field.

Message pydantic-model

Bases: BaseModel

Generic message used in a conversation

Show JSON schema:

{
  "description": "Generic message used in a conversation",
  "properties": {
    "kind": {
      "const": "message",
      "default": "message",
      "title": "Kind",
      "type": "string"
    },
    "role": {
      "title": "Role",
      "type": "string"
    },
    "content": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Content"
    },
    "metadata": {
      "additionalProperties": true,
      "title": "Metadata",
      "type": "object"
    }
  },
  "required": [
    "role"
  ],
  "title": "Message",
  "type": "object"
}

Fields:

• kind (Literal['message'])
• role (str)
• content (str | None)
• metadata (dict[str, Any])

Source code in src/ant_ai/core/message.py

class Message(BaseModel):
    """Generic message used in a conversation"""

    kind: Literal["message"] = Field(default="message")
    """Type of the message. Used to reconstruct, not passed to the LLM"""
    role: str
    """Role of the message sender (e.g. "user", "assistant")"""
    content: str | None = Field(default=None)
    """Text content of the message"""
    metadata: dict[str, Any] = Field(default_factory=dict)
    """Additional metadata associated with the message"""

    def to_provider_dict(self) -> dict[str, Any]:
        """The message in the shape chat-completion APIs expect.

        Internal fields (`kind`, and `is_error` on tool results) are left out of
        the OpenAI-style dict; a provider adapter that has a native equivalent
        (Anthropic's `tool_result.is_error`) can map them explicitly.
        """
        return self.model_dump(exclude=set(_PROVIDER_EXCLUDED_FIELDS))

kind pydantic-field

kind: Literal['message'] = 'message'

Type of the message. Used to reconstruct, not passed to the LLM

role pydantic-field

role: str

Role of the message sender (e.g. "user", "assistant")

content pydantic-field

content: str | None = None

Text content of the message

metadata pydantic-field

metadata: dict[str, Any]

Additional metadata associated with the message

to_provider_dict

to_provider_dict() -> dict[str, Any]

The message in the shape chat-completion APIs expect.

Internal fields (kind, and is_error on tool results) are left out of the OpenAI-style dict; a provider adapter that has a native equivalent (Anthropic's tool_result.is_error) can map them explicitly.

Source code in src/ant_ai/core/message.py

def to_provider_dict(self) -> dict[str, Any]:
    """The message in the shape chat-completion APIs expect.

    Internal fields (`kind`, and `is_error` on tool results) are left out of
    the OpenAI-style dict; a provider adapter that has a native equivalent
    (Anthropic's `tool_result.is_error`) can map them explicitly.
    """
    return self.model_dump(exclude=set(_PROVIDER_EXCLUDED_FIELDS))

MessageChunk pydantic-model

Bases: BaseModel

Represents partial output from an LLM during streaming. delta is the content's only the newly streamed text fragment, not the whole message so far.

Show JSON schema:

{
  "description": "Represents *partial output* from an LLM during streaming.\n`delta` is the content's only the newly streamed text fragment,\nnot the whole message so far.",
  "properties": {
    "role": {
      "title": "Role",
      "type": "string"
    },
    "delta": {
      "title": "Delta",
      "type": "string"
    },
    "metadata": {
      "additionalProperties": true,
      "title": "Metadata",
      "type": "object"
    }
  },
  "required": [
    "role",
    "delta"
  ],
  "title": "MessageChunk",
  "type": "object"
}

Fields:

• role (str)
• delta (str)
• metadata (dict)

Source code in src/ant_ai/core/message.py

class MessageChunk(BaseModel):
    """
    Represents *partial output* from an LLM during streaming.
    `delta` is the content's only the newly streamed text fragment,
    not the whole message so far.
    """

    role: str
    delta: str
    metadata: dict = Field(default_factory=dict)

    def merge(self, other: MessageChunk) -> MessageChunk:
        """Combines two MessageChunk into one

        Args:
            other (MessageChunk): The other message chunk to combine with

        Returns:
            MessageChunk: A combined chunk where this is before the other chunk
        """
        if self.role != other.role:
            raise ValueError("roles must match to merge")
        return MessageChunk(
            role=self.role,
            delta=self.delta + other.delta,
            metadata={**self.metadata, **other.metadata},
        )

    def to_message(self) -> Message:
        """Convert final chunk to a full message once streaming ends."""
        return Message(role=self.role, content=self.delta, metadata=self.metadata)

merge

merge(other: MessageChunk) -> MessageChunk

Combines two MessageChunk into one

Parameters:

Name	Type	Description	Default
other	MessageChunk	The other message chunk to combine with	required

Returns:

Name	Type	Description
MessageChunk	MessageChunk	A combined chunk where this is before the other chunk

Source code in src/ant_ai/core/message.py

def merge(self, other: MessageChunk) -> MessageChunk:
    """Combines two MessageChunk into one

    Args:
        other (MessageChunk): The other message chunk to combine with

    Returns:
        MessageChunk: A combined chunk where this is before the other chunk
    """
    if self.role != other.role:
        raise ValueError("roles must match to merge")
    return MessageChunk(
        role=self.role,
        delta=self.delta + other.delta,
        metadata={**self.metadata, **other.metadata},
    )

to_message

to_message() -> Message

Convert final chunk to a full message once streaming ends.

Source code in src/ant_ai/core/message.py

def to_message(self) -> Message:
    """Convert final chunk to a full message once streaming ends."""
    return Message(role=self.role, content=self.delta, metadata=self.metadata)

ToolCallMessage pydantic-model

Bases: Message

Message representing a tool call in a conversation

Show JSON schema:

{
  "$defs": {
    "ToolCall": {
      "description": "Single tool call object inside assistant.tool_calls (OpenAI schema).",
      "properties": {
        "id": {
          "title": "Id",
          "type": "string"
        },
        "type": {
          "default": "function",
          "title": "Type",
          "type": "string"
        },
        "function": {
          "$ref": "#/$defs/ToolFunction"
        }
      },
      "required": [
        "id",
        "function"
      ],
      "title": "ToolCall",
      "type": "object"
    },
    "ToolFunction": {
      "description": "Inner function payload for a tool call (OpenAI schema).",
      "properties": {
        "name": {
          "title": "Name",
          "type": "string"
        },
        "arguments": {
          "title": "Arguments",
          "type": "string"
        }
      },
      "required": [
        "name",
        "arguments"
      ],
      "title": "ToolFunction",
      "type": "object"
    }
  },
  "description": "Message representing a tool call in a conversation",
  "properties": {
    "kind": {
      "const": "tool_call",
      "default": "tool_call",
      "title": "Kind",
      "type": "string"
    },
    "role": {
      "default": "assistant",
      "title": "Role",
      "type": "string"
    },
    "content": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Content"
    },
    "metadata": {
      "additionalProperties": true,
      "title": "Metadata",
      "type": "object"
    },
    "tool_calls": {
      "items": {
        "$ref": "#/$defs/ToolCall"
      },
      "title": "Tool Calls",
      "type": "array"
    }
  },
  "required": [
    "tool_calls"
  ],
  "title": "ToolCallMessage",
  "type": "object"
}

Fields:

• content (str | None)
• metadata (dict[str, Any])
• kind (Literal['tool_call'])
• role (str)
• tool_calls (list[ToolCall])

Source code in src/ant_ai/core/message.py

class ToolCallMessage(Message):
    """Message representing a tool call in a conversation"""

    kind: Literal["tool_call"] = Field(default="tool_call")
    role: str = "assistant"
    tool_calls: list[ToolCall]

ToolCallResultMessage pydantic-model

Bases: Message

Message representing the result of a tool call in a conversation

Show JSON schema:

{
  "description": "Message representing the result of a tool call in a conversation",
  "properties": {
    "kind": {
      "const": "tool_call_result",
      "default": "tool_call_result",
      "title": "Kind",
      "type": "string"
    },
    "role": {
      "default": "tool",
      "title": "Role",
      "type": "string"
    },
    "content": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "title": "Content"
    },
    "metadata": {
      "additionalProperties": true,
      "title": "Metadata",
      "type": "object"
    },
    "tool_call_id": {
      "title": "Tool Call Id",
      "type": "string"
    },
    "name": {
      "title": "Name",
      "type": "string"
    },
    "is_error": {
      "default": false,
      "description": "Whether the tool call failed. Set by `ToolStep` when the tool raised; mirrors `is_error` on Anthropic tool results and `isError` on MCP.",
      "title": "Is Error",
      "type": "boolean"
    }
  },
  "required": [
    "tool_call_id",
    "name"
  ],
  "title": "ToolCallResultMessage",
  "type": "object"
}

Fields:

• content (str | None)
• metadata (dict[str, Any])
• kind (Literal['tool_call_result'])
• role (str)
• tool_call_id (str)
• name (str)
• is_error (bool)

Source code in src/ant_ai/core/message.py

class ToolCallResultMessage(Message):
    """Message representing the result of a tool call in a conversation"""

    kind: Literal["tool_call_result"] = Field(default="tool_call_result")
    role: str = "tool"
    tool_call_id: str
    name: str
    is_error: bool = Field(
        default=False,
        description=(
            "Whether the tool call failed. Set by `ToolStep` when the tool raised; "
            "mirrors `is_error` on Anthropic tool results and `isError` on MCP."
        ),
    )

is_error pydantic-field

is_error: bool = False

Whether the tool call failed. Set by ToolStep when the tool raised; mirrors is_error on Anthropic tool results and isError on MCP.

ToolFunction pydantic-model

Bases: BaseModel

Inner function payload for a tool call (OpenAI schema).

Show JSON schema:

{
  "description": "Inner function payload for a tool call (OpenAI schema).",
  "properties": {
    "name": {
      "title": "Name",
      "type": "string"
    },
    "arguments": {
      "title": "Arguments",
      "type": "string"
    }
  },
  "required": [
    "name",
    "arguments"
  ],
  "title": "ToolFunction",
  "type": "object"
}

Fields:

• name (str)
• arguments (str)

Source code in src/ant_ai/core/message.py

class ToolFunction(BaseModel):
    """Inner function payload for a tool call (OpenAI schema)."""

    name: str
    """Name of the tool called"""
    # OpenAI expects a JSON string, not a dict.
    arguments: str
    """Arguments for the tool"""

name pydantic-field

name: str

Name of the tool called

arguments pydantic-field

arguments: str

Arguments for the tool

ToolCall pydantic-model

Bases: BaseModel

Single tool call object inside assistant.tool_calls (OpenAI schema).

Show JSON schema:

{
  "$defs": {
    "ToolFunction": {
      "description": "Inner function payload for a tool call (OpenAI schema).",
      "properties": {
        "name": {
          "title": "Name",
          "type": "string"
        },
        "arguments": {
          "title": "Arguments",
          "type": "string"
        }
      },
      "required": [
        "name",
        "arguments"
      ],
      "title": "ToolFunction",
      "type": "object"
    }
  },
  "description": "Single tool call object inside assistant.tool_calls (OpenAI schema).",
  "properties": {
    "id": {
      "title": "Id",
      "type": "string"
    },
    "type": {
      "default": "function",
      "title": "Type",
      "type": "string"
    },
    "function": {
      "$ref": "#/$defs/ToolFunction"
    }
  },
  "required": [
    "id",
    "function"
  ],
  "title": "ToolCall",
  "type": "object"
}

Fields:

• id (str)
• type (str)
• function (ToolFunction)

Source code in src/ant_ai/core/message.py

class ToolCall(BaseModel):
    """Single tool call object inside assistant.tool_calls (OpenAI schema)."""

    id: str
    """Unique ID for the tool call, is generated by the OpenAI compatible endpoint"""
    type: str = "function"
    """Type of the tool call"""
    function: ToolFunction
    """Tool called"""

id pydantic-field

id: str

Unique ID for the tool call, is generated by the OpenAI compatible endpoint

type pydantic-field

type: str = 'function'

Type of the tool call

function pydantic-field

function: ToolFunction

Tool called

sanitize_messages

sanitize_messages(messages: list[Message]) -> list[Message]

Return a copy of messages with every incomplete tool-call group removed.

An incomplete group is a ToolCallMessage whose immediately-following role="tool" messages are fewer than the number of tool_calls it declares. Such groups cause OpenAI-compatible APIs to reject the request with a 400 error.

Complete groups and all non-tool-call messages are preserved in order.

Source code in src/ant_ai/core/message.py

def sanitize_messages(messages: list[Message]) -> list[Message]:
    """Return a copy of *messages* with every incomplete tool-call group removed.

    An incomplete group is a ``ToolCallMessage`` whose immediately-following
    ``role="tool"`` messages are fewer than the number of ``tool_calls`` it
    declares.  Such groups cause OpenAI-compatible APIs to reject the request
    with a 400 error.

    Complete groups and all non-tool-call messages are preserved in order.
    """
    result: list[Message] = []
    i = 0
    while i < len(messages):
        msg = messages[i]
        if isinstance(msg, ToolCallMessage):
            n_calls = len(msg.tool_calls)
            j = i + 1
            while j < len(messages) and messages[j].role == "tool":
                j += 1
            if j - (i + 1) >= n_calls:
                result.extend(messages[i:j])
            i = j
        else:
            result.append(msg)
            i += 1
    return result