LLM Clients API Reference

Maticlib provides a unified interface for multiple LLM providers. All clients inherit from BaseLLMClient and support both synchronous and asynchronous completion.

OpenAI Client

Using the modern OpenAI Responses API.

maticlib.llm.openai.client.OpenAIClient

OpenAIClient(
    model="gpt-4o-mini",
    system_instruct=None,
    api_key=None,
    verbose=True,
    return_raw=False,
)

Bases: BaseLLMClient

Client for interacting with OpenAI models via the Responses API.

Inherits from BaseLLMClient and implements OpenAI-specific message formatting and response parsing. Supports all current GPT and o-series models.

Parameters:

Name	Type	Description	Default
model	str	The OpenAI model to use. Defaults to "gpt-4o-mini". Examples: "gpt-4o", "gpt-4.1", "o4-mini", "gpt-5.4".	'gpt-4o-mini'
system_instruct	str | SystemMessage	An optional system / developer prompt prepended to every request.	None
api_key	str	Your OpenAI API key. Falls back to the OPENAI_API_KEY environment variable.	None
verbose	bool	If True, prints HTTP status codes to stdout.	True
return_raw	bool	If True, the complete / async_complete methods return the raw dict instead of an OpenAIResponse model.	False

Source code in maticlib/llm/openai/client.py

def __init__(
    self,
    model: str = "gpt-4o-mini",
    system_instruct: Union[str, SystemMessage, None] = None,
    api_key: Optional[str] = None,
    verbose: bool = True,
    return_raw: bool = False,
) -> None:
    api_key = api_key or os.getenv("OPENAI_API_KEY", "")
    api_key = (api_key or "").strip()
    if not api_key:
        raise ValueError(
            "OpenAI API key is missing. Please provide it via the 'api_key' "
            "argument or set the OPENAI_API_KEY environment variable."
        )
    self.api_key = api_key
    self.model = model
    self.system_instruct = system_instruct
    self.base_url = "https://api.openai.com/v1"
    self.verbose = verbose
    self.return_raw = return_raw
    self.headers = {
        "Authorization": f"Bearer {self.api_key}",
        "Content-Type": "application/json",
    }

async_complete async

async_complete(input, response_model=None, tools=None)

Sends an asynchronous generation request to the OpenAI Responses API.

Parameters:

Name	Type	Description	Default
input	str | list	The user prompt as a plain string, or a conversation history as a list of message objects / dicts.	required
response_model	Type[BaseModel]	A Pydantic model to parse the output into.	None
tools	list	A list of tool functions decorated with @tool.	None

Source code in maticlib/llm/openai/client.py

async def async_complete(
    self,
    input: Union[str, List],
    response_model: Optional[Type[BaseModel]] = None,
    tools: Optional[List[Callable]] = None,
) -> Union[OpenAIResponse, Dict[str, Any]]:
    """
    Sends an asynchronous generation request to the OpenAI Responses API.

    Args:
        input (str | list): The user prompt as a plain string, or a
            conversation history as a list of message objects / dicts.
        response_model (Type[BaseModel], optional): A Pydantic model to
            parse the output into.
        tools (list, optional): A list of tool functions decorated with @tool.
    """
    url = f"{self.base_url}/responses"

    try:
        input = self._inject_runtime_instructions(input, response_model)
        payload = self._build_payload(input, tools=tools)
        async with httpx.AsyncClient() as client:
            response = await client.post(
                url, headers=self.headers, json=payload, timeout=60.0
            )
            response.raise_for_status()

            if self.verbose:
                print(f"Status: {response.status_code}")

            result = self._parse_response(response)
            self._apply_response_model(result, response_model)
            return result

    except httpx.HTTPStatusError as e:
        if self.verbose:
            print(f"HTTP Error: {e.response.status_code}")
            print(f"Response: {e.response.text}")
        raise
    except Exception:
        if self.verbose:
            import traceback

            traceback.print_exc()
        raise

complete

complete(input, response_model=None, tools=None)

Sends a synchronous generation request to the OpenAI Responses API.

Parameters:

Name	Type	Description	Default
input	str | list	The user prompt as a plain string, or a conversation history as a list of message objects / dicts.	required
response_model	Type[BaseModel]	A Pydantic model to parse the output into.	None
tools	list	A list of tool functions decorated with @tool.	None

Source code in maticlib/llm/openai/client.py

def complete(
    self,
    input: Union[str, List],
    response_model: Optional[Type[BaseModel]] = None,
    tools: Optional[List[Callable]] = None,
) -> Union[OpenAIResponse, Dict[str, Any]]:
    """
    Sends a synchronous generation request to the OpenAI Responses API.

    Args:
        input (str | list): The user prompt as a plain string, or a
            conversation history as a list of message objects / dicts.
        response_model (Type[BaseModel], optional): A Pydantic model to
            parse the output into.
        tools (list, optional): A list of tool functions decorated with @tool.
    """
    url = f"{self.base_url}/responses"

    try:
        input = self._inject_runtime_instructions(input, response_model)
        payload = self._build_payload(input, tools=tools)
        response = httpx.post(url, headers=self.headers, json=payload, timeout=60.0)
        response.raise_for_status()

        if self.verbose:
            print(f"Status: {response.status_code}")

        result = self._parse_response(response)
        self._apply_response_model(result, response_model)
        return result

    except httpx.HTTPStatusError as e:
        if self.verbose:
            print(f"HTTP Error: {e.response.status_code}")
            print(f"Response: {e.response.text}")
        raise
    except Exception:
        if self.verbose:
            import traceback

            traceback.print_exc()
        raise

get_text_response

get_text_response(response)

Extracts the primary text content from an OpenAI response.

This is a convenience helper so callers do not need to traverse the output list manually.

Parameters:

Name	Type	Description	Default
response	OpenAIResponse | dict	The response returned by complete or async_complete.	required

Returns:

Name	Type	Description
str	str	The extracted text string, or an empty string if no text was found.

Source code in maticlib/llm/openai/client.py

def get_text_response(self, response: Union[OpenAIResponse, Dict[str, Any]]) -> str:
    """
    Extracts the primary text content from an OpenAI response.

    This is a convenience helper so callers do not need to traverse
    the ``output`` list manually.

    Args:
        response (OpenAIResponse | dict): The response returned by
            ``complete`` or ``async_complete``.

    Returns:
        str: The extracted text string, or an empty string if no text was found.
    """
    if isinstance(response, OpenAIResponse):
        return response.content or ""

    # Raw dict fallback: walk output items manually
    try:
        for item in response.get("output", []):
            for part in item.get("content", []):
                if part.get("type") == "output_text" and part.get("text"):
                    return part["text"]
    except Exception:
        raise

    return ""

Mistral Client

Using the native Mistral AI Chat Completions API.

maticlib.llm.mistral.client.MistralClient

MistralClient(
    model="mistral-medium-latest",
    system_instruct=None,
    api_key=None,
    verbose=True,
    return_raw=False,
)

Bases: BaseLLMClient

Client for interacting with Mistral AI models.

Inherits from BaseLLMClient and implements Mistral-specific message formatting and response parsing.

Parameters:

Name	Type	Description	Default
model	str	The name of the Mistral model to use. Defaults to "mistral-medium-latest".	'mistral-medium-latest'
system_instruct	str | SystemMessage	Default instructions to prepend to all conversations.	None
api_key	str	Your Mistral AI API key. Defaults to MISTRAL_API_KEY environment variable.	None
verbose	bool	If True, prints status messages to console.	True
return_raw	bool	If True, returns the raw dict response instead of a MistralResponse model.	False

Source code in maticlib/llm/mistral/client.py

def __init__(
    self,
    model: str = "mistral-medium-latest",
    system_instruct: str | SystemMessage | None = None,
    api_key: Optional[str] = None,
    verbose: bool = True,
    return_raw: bool = False,
):
    api_key = api_key or os.getenv("MISTRAL_API_KEY", "")
    api_key = (api_key or "").strip()
    if not api_key:
        raise ValueError(
            "Mistral API key is missing. Please provide it via the 'api_key' "
            "argument or set the MISTRAL_API_KEY environment variable."
        )
    self.api_key = api_key
    self.model = model
    self.system_instruct = system_instruct
    self.base_url = "https://api.mistral.ai/v1"
    self.verbose = verbose
    self.headers = {
        "Authorization": f"Bearer {self.api_key}",
        "Content-Type": "application/json",
        "Accept": "application/json",
    }
    self.return_raw = (
        return_raw  # Option to return raw JSON response or Pydantic model
    )

async_complete async

async_complete(input, response_model=None, tools=None)

Sends an asynchronous chat completion request to Mistral.

Parameters:

Name	Type	Description	Default
input	str | list	The user prompt or conversation history.	required
response_model	Type[BaseModel]	A Pydantic model to parse the output into.	None
tools	list	A list of tool functions decorated with @tool.	None

Source code in maticlib/llm/mistral/client.py

async def async_complete(
    self,
    input: Union[str, List],
    response_model: Optional[Type[BaseModel]] = None,
    tools: Optional[List[Callable]] = None,
) -> Union[MistralResponse, Dict[str, Any]]:
    """
    Sends an asynchronous chat completion request to Mistral.

    Args:
        input (str | list): The user prompt or conversation history.
        response_model (Type[BaseModel], optional): A Pydantic model to
            parse the output into.
        tools (list, optional): A list of tool functions decorated with @tool.
    """
    url = f"{self.base_url}/chat/completions"

    try:
        # Inject structure instructions if requested
        input = self._inject_runtime_instructions(input, response_model)

        # Format messages
        formatted_messages = self._format_messages(input)

        payload = {"model": self.model, "messages": formatted_messages}

        # Handle tools
        if tools:
            payload["tools"] = self._format_tools(tools)

        # Make async request
        async with httpx.AsyncClient() as client:
            response = await client.post(
                url, headers=self.headers, json=payload, timeout=30.0
            )
            response.raise_for_status()

            if self.verbose:
                print(f"Status: {response.status_code}")

            # Parse and return response
            result = self._parse_response(response)
            self._apply_response_model(result, response_model)
            return result

    except httpx.HTTPStatusError as e:
        if self.verbose:
            print(f"HTTP Error: {e.response.status_code}")
            print(f"Response: {e.response.text}")
        raise
    except Exception as e:
        if self.verbose:
            import traceback

            traceback.print_exc()
        raise

complete

complete(input, response_model=None, tools=None)

Sends a synchronous chat completion request to Mistral.

Parameters:

Name	Type	Description	Default
input	str | list	The user prompt or conversation history.	required
response_model	Type[BaseModel]	A Pydantic model to parse the output into.	None
tools	list	A list of tool functions decorated with @tool.	None

Source code in maticlib/llm/mistral/client.py

def complete(
    self,
    input: Union[str, List],
    response_model: Optional[Type[BaseModel]] = None,
    tools: Optional[List[Callable]] = None,
) -> Union[MistralResponse, Dict[str, Any]]:
    """
    Sends a synchronous chat completion request to Mistral.

    Args:
        input (str | list): The user prompt or conversation history.
        response_model (Type[BaseModel], optional): A Pydantic model to
            parse the output into.
        tools (list, optional): A list of tool functions decorated with @tool.
    """
    url = f"{self.base_url}/chat/completions"

    try:
        # Inject structure instructions if requested
        input = self._inject_runtime_instructions(input, response_model)

        # Format messages
        formatted_messages = self._format_messages(input)

        payload = {"model": self.model, "messages": formatted_messages}

        # Handle tools
        if tools:
            payload["tools"] = self._format_tools(tools)

        # Make request
        response = httpx.post(url, headers=self.headers, json=payload, timeout=30.0)
        response.raise_for_status()

        if self.verbose:
            print(f"Status: {response.status_code}")

        # Parse and return response
        result = self._parse_response(response)
        self._apply_response_model(result, response_model)
        return result

    except httpx.HTTPStatusError as e:
        if self.verbose:
            print(f"HTTP Error: {e.response.status_code}")
            print(f"Response: {e.response.text}")
        raise
    except Exception as e:
        if self.verbose:
            import traceback

            traceback.print_exc()
        raise

get_text_response

get_text_response(response)

Extracts the primary text content from a Mistral response.

Parameters:

Name	Type	Description	Default
response	MistralResponse | dict	The response to extract from.	required

Returns:

Name	Type	Description
str	str	The extracted text string.

Source code in maticlib/llm/mistral/client.py

def get_text_response(
    self, response: Union[MistralResponse, Dict[str, Any]]
) -> str:
    """
    Extracts the primary text content from a Mistral response.

    Args:
        response (MistralResponse | dict): The response to extract from.

    Returns:
        str: The extracted text string.
    """
    if isinstance(response, MistralResponse):
        return response.content or ""

    # Handle raw dict response
    try:
        choices = response.get("choices", [])
        if choices:
            message = choices[0].get("message", {})
            content = message.get("content", "")
            return content
    except Exception:
        raise

Google GenAI Client

Using the Google Gemini Developer API.

maticlib.llm.google_genai.client.GoogleGenAIClient

GoogleGenAIClient(
    model="gemini-2.5-flash-lite",
    system_instruct=None,
    api_key=None,
    thinking_budget=0,
    verbose=True,
    return_raw=False,
)

Bases: BaseLLMClient

Client for interacting with Google's Generative AI (Gemini) models.

Inherits from BaseLLMClient and implements Gemini-specific message formatting and response parsing.

Parameters:

Name	Type	Description	Default
model	str	The name of the Gemini model to use. Defaults to "gemini-2.5-flash".	'gemini-2.5-flash-lite'
system_instruct	str | SystemMessage	Default instructions to prepend to all conversations.	None
api_key	str	Your Google AI API key.	None
thinking_budget	int	Optional token budget for model reasoning/thinking.	0
verbose	bool	If True, prints status messages to console.	True
return_raw	bool	If True, returns the raw dict response instead of a GeminiResponse model.	False

Source code in maticlib/llm/google_genai/client.py

def __init__(
    self,
    model: str = "gemini-2.5-flash-lite",
    system_instruct: str | SystemMessage | None = None,
    api_key: Optional[str] = None,
    thinking_budget: int = 0,
    verbose: bool = True,
    return_raw: bool = False,
):
    api_key = (
        api_key or os.getenv("GOOGLE_API_KEY") or os.getenv("GEMINI_API_KEY") or ""
    )
    api_key = (api_key or "").strip()
    if not api_key:
        raise ValueError(
            "Google Gemini API key is missing. Please provide it via the 'api_key' "
            "argument or set the GOOGLE_API_KEY environment variable."
        )
    self.api_key = api_key
    self.model = model
    self.system_instruct = system_instruct
    self.base_url = "https://generativelanguage.googleapis.com/v1beta"
    self.verbose = verbose
    self.headers = {
        "x-goog-api-key": self.api_key,
        "Content-Type": "application/json",
    }
    self.thinking_budget = thinking_budget
    self.return_raw = (
        return_raw  # Option to return raw JSON response or Pydantic model
    )

async_complete async

async_complete(input, response_model=None, tools=None)

Sends an asynchronous generation request to Gemini.

Parameters:

Name	Type	Description	Default
input	str	The text input to send to the model.	required
response_model	Type[BaseModel]	A Pydantic model to parse the output into.	None
tools	list	A list of tool functions decorated with @tool.	None

Source code in maticlib/llm/google_genai/client.py

async def async_complete(
    self,
    input: str,
    response_model: Optional[Type[BaseModel]] = None,
    tools: Optional[List[Callable]] = None,
) -> Union[GeminiResponse, Dict[str, Any]]:
    """
    Sends an asynchronous generation request to Gemini.

    Args:
        input (str): The text input to send to the model.
        response_model (Type[BaseModel], optional): A Pydantic model to
            parse the output into.
        tools (list, optional): A list of tool functions decorated with @tool.
    """
    url = f"{self.base_url}/models/{self.model}:generateContent"

    # Inject structure instructions if requested
    input = self._inject_runtime_instructions(input, response_model)

    formatted_messages = self._format_messages(input=input)

    payload = {}

    # Handle tools
    if tools:
        payload["tools"] = self._format_tools(tools)

    if self.system_instruct:
        self.system_instruct = self._format__system_instruction()

        payload["system_instruction"] = {"parts": [{"text": self.system_instruct}]}

    payload["contents"] = formatted_messages

    if self.thinking_budget > 0:
        payload["generationConfig"] = {"thinkingBudget": self.thinking_budget}

    try:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                url, headers=self.headers, json=payload, timeout=60.0
            )
            response.raise_for_status()
        if self.verbose:
            print(f"Status: {response.status_code}")

        result = self._parse_response(response=response)
        self._apply_response_model(result, response_model)
        return result
    except Exception as e:
        import traceback

        traceback.print_exc()
        raise
    finally:
        await client.aclose()

complete

complete(input, response_model=None, tools=None)

Sends a synchronous generation request to Gemini.

Parameters:

Name	Type	Description	Default
input	str | list	The user prompt or conversation history.	required
response_model	Type[BaseModel]	A Pydantic model to parse the output into.	None
tools	list	A list of tool functions decorated with @tool.	None

Source code in maticlib/llm/google_genai/client.py

def complete(
    self,
    input: Union[str, List],
    response_model: Optional[Type[BaseModel]] = None,
    tools: Optional[List[Callable]] = None,
) -> Union[GeminiResponse, Dict[str, Any]]:
    """
    Sends a synchronous generation request to Gemini.

    Args:
        input (str | list): The user prompt or conversation history.
        response_model (Type[BaseModel], optional): A Pydantic model to
            parse the output into.
        tools (list, optional): A list of tool functions decorated with @tool.
    """
    url = f"{self.base_url}/models/{self.model}:generateContent"

    try:
        # Inject structure instructions if requested
        input = self._inject_runtime_instructions(input, response_model)

        # Format messages
        formatted_messages = self._format_messages(input)

        payload = {}

        # Handle tools
        if tools:
            payload["tools"] = self._format_tools(tools)

        if self.system_instruct:
            self.system_instruct = self._format__system_instruction()

            payload["system_instruction"] = {
                "parts": [{"text": self.system_instruct}]
            }

        payload["contents"] = formatted_messages

        # Add thinking budget if configured
        if self.thinking_budget > 0:
            payload["generationConfig"] = {"thinkingBudget": self.thinking_budget}

        # Make request
        response = httpx.post(url, headers=self.headers, json=payload, timeout=60.0)
        response.raise_for_status()

        if self.verbose:
            print(f"Status: {response.status_code}")

        result = self._parse_response(response)
        self._apply_response_model(result, response_model)
        return result

    except httpx.HTTPStatusError as e:
        if self.verbose:
            print(f"HTTP Error: {e.response.status_code}")
            print(f"Response: {e.response.text}")
        raise
    except Exception as e:
        if self.verbose:
            import traceback

            traceback.print_exc()
        raise

get_text_response

get_text_response(response)

Extracts the primary text content from a Gemini response.

Parameters:

Name	Type	Description	Default
response	GeminiResponse | dict	The response to extract from.	required

Returns:

Name	Type	Description
str	str	The extracted text string.

Source code in maticlib/llm/google_genai/client.py

def get_text_response(self, response: Union[GeminiResponse, Dict[str, Any]]) -> str:
    """
    Extracts the primary text content from a Gemini response.

    Args:
        response (GeminiResponse | dict): The response to extract from.

    Returns:
        str: The extracted text string.
    """
    if isinstance(response, GeminiResponse):
        return response.content or ""

    # Handle raw dict response
    try:
        candidates = response.get("candidates", [])
        if candidates:
            parts = candidates[0].get("content", {}).get("parts", [])
            texts = [part.get("text", "") for part in parts if "text" in part]
            return " ".join(texts)
    except Exception:
        raise

---

Response Models

Standardized models used to ensure consistency across providers.

OpenAI Response (OpenAIResponse)

All OpenAI clients return an OpenAIResponse containing both general and provider-specific fields.

Common (Inherited) Fields

Field	Type	Source Mapping
content	str	Concatenated text from all output_text parts.
content_parts	List[ContentPart]	Exactly one ContentPart per output_text chunk.
prompt_tokens	int	Mapped from usage.input_tokens.
completion_tokens	int	Mapped from usage.output_tokens.
total_tokens	int	Mapped from usage.total_tokens.
finish_reason	str	Mapped from the first output item's status.
response_id	str	Mapped from top-level id.
raw_response	Dict[str, Any]	The original, full JSON response dictionary.

OpenAI-Specific Fields

Field	Type	Description
id	str	Response ID (prefixed with resp_).
object	str	Always "response".
created_at	int	Unix timestamp of creation.
status	str	Response-level status (e.g. completed, failed).
output	List[OpenAIOutputMessage]	Ordered list of output items returned by the API.
usage	OpenAIUsage	Detailed token-usage breakdown.
model_version	str	Model version string echoed back by OpenAI.

OpenAI-Specific Properties

Property	Type	Description
cached_tokens	int	Input tokens served from the prompt cache. A non-zero value means the model reused previously computed KV-cache entries.
reasoning_tokens	int \| None	Tokens used for internal model reasoning (o-series models only). Returns None for standard models.
timestamp	datetime	Converts the created_at Unix timestamp into a datetime object.

maticlib.llm.openai.openai_classes.OpenAIResponse

OpenAIResponse(**data)

Bases: LLMResponseBase

OpenAI Responses API response (/v1/responses).

Maps the raw JSON payload onto the shared LLMResponseBase interface so callers can use response.content and response.content_parts the same way they would with MistralResponse or GeminiResponse.

Attributes:

Name	Type	Description
content	str	Concatenated text from all output_text parts.
content_parts	List[ContentPart]	One ContentPart per output_text chunk.
prompt_tokens	int	Mapped from usage.input_tokens.
completion_tokens	int	Mapped from usage.output_tokens.
total_tokens	int	Mapped from usage.total_tokens.
finish_reason	str	Mapped from first output item's status.
response_id	str	Mapped from top-level id.
raw_response	Dict[str, Any]	Original JSON dict.
id	str	Response ID (resp_...).
object	str	Always "response".
created_at	int	Unix timestamp of creation.
status	str	Response-level status (completed, failed, ...).
output	List[OpenAIOutputMessage]	Ordered list of output items.
usage	OpenAIUsage	Detailed token-usage breakdown.
model_version	str	Model string echoed back by OpenAI.
cached_tokens	Optional[int]	Input tokens served from the prompt cache.
reasoning_tokens	Optional[int]	Tokens used for internal model reasoning (o-series models only).
timestamp	datetime	Converts the created_at Unix timestamp into a datetime object.

Source code in maticlib/llm/openai/openai_classes.py

def __init__(self, **data: Any) -> None:
    # ------------------------------------------------------------------
    # 1. Walk every output item and extract text into content / content_parts
    # ------------------------------------------------------------------
    text_parts: List[str] = []
    content_parts: List[ContentPart] = []

    for item in data.get("output", []):
        if not isinstance(item, dict):
            continue
        for part in item.get("content", []):
            if not isinstance(part, dict):
                continue
            if part.get("type") == "output_text" and part.get("text"):
                text_parts.append(part["text"])
                content_parts.append(
                    ContentPart(type=ModalityType.TEXT, text=part["text"])
                )

    if text_parts:
        data["content"] = " ".join(text_parts)
    if content_parts:
        data["content_parts"] = content_parts

    # ------------------------------------------------------------------
    # 2. Map usage onto the shared LLMResponseBase token fields
    # ------------------------------------------------------------------
    usage_raw = data.get("usage") or {}
    if isinstance(usage_raw, dict):
        data["prompt_tokens"] = usage_raw.get("input_tokens")
        data["completion_tokens"] = usage_raw.get("output_tokens")
        data["total_tokens"] = usage_raw.get("total_tokens")

    # ------------------------------------------------------------------
    # 3. finish_reason -- use the status of the first output item
    # ------------------------------------------------------------------
    output_list = data.get("output") or []
    if output_list and isinstance(output_list[0], dict):
        data["finish_reason"] = output_list[0].get("status")

    # ------------------------------------------------------------------
    # 5. Extract tool calls from output items
    # ------------------------------------------------------------------
    tool_calls: List[Dict[str, Any]] = []
    for item in data.get("output", []):
        if not isinstance(item, dict):
            continue
        if item.get("type") == "call_tool":
            tool_calls.append(
                {
                    "id": item.get("id"),
                    "type": "function",
                    "function": {
                        "name": item.get("call_tool", {}).get("name"),
                        "arguments": item.get("call_tool", {}).get("arguments"),
                    },
                }
            )
    if tool_calls:
        data["tool_calls"] = tool_calls

    # Standardise response identifiers
    data["response_id"] = data.get("id")
    data["model_version"] = data.get("model")

    # Preserve raw JSON before super().__init__ may alter data
    data["raw_response"] = data.copy()

    super().__init__(**data)

cached_tokens property

cached_tokens

Input tokens served from the prompt cache.

A non-zero value means the model reused previously computed KV-cache entries, which are billed at a reduced rate.

reasoning_tokens property

reasoning_tokens

Tokens used for internal model reasoning (o-series models only).

Returns None for standard GPT models that do not expose reasoning-token counts.

timestamp property

timestamp

Converts the created_at Unix timestamp into a datetime object.

Mistral Response (MistralResponse)

maticlib.llm.mistral.mistral_classes.MistralResponse

Bases: LLMResponseBase

Mistral-specific response structure. Supports both text-only and multimodal (Pixtral) models. Inherits from LLMResponseBase and adds Mistral-specific fields.

Mistral-Specific Properties

timestamp

Convert Unix timestamp to datetime.

maticlib.llm.mistral.mistral_classes.MistralResponse

MistralResponse(**data)

Bases: LLMResponseBase

Mistral-specific response structure.

Supports both text-only and multimodal (Pixtral) models. Inherits from LLMResponseBase and adds Mistral-specific fields.

Attributes:

Name	Type	Description
id	str	Unique identifier for the Mistral response.
created	int	Unix timestamp of creation.
object	str	Object type (e.g., 'chat.completion').
choices	List[MistralChoice]	List of completion choices.
timestamp	datetime	Convert Unix timestamp to datetime object.

Source code in maticlib/llm/mistral/mistral_classes.py

def __init__(self, **data):
    # Extract common fields from Mistral structure
    if "choices" in data and len(data["choices"]) > 0:
        first_choice = data["choices"][0]
        message = first_choice.get("message", {})
        content = message.get("content")

        # Handle multimodal content (list of parts) or text-only (string)
        if isinstance(content, list):
            # Multimodal response with parts
            content_parts = []
            text_parts = []
            for part in content:
                if isinstance(part, dict):
                    content_part = ContentPart(
                        type=ModalityType(part.get("type", "text")),
                        text=part.get("text"),
                        image_url=part.get("image_url"),
                    )
                    content_parts.append(content_part)
                    if part.get("text"):
                        text_parts.append(part["text"])

            data["content_parts"] = content_parts
            data["content"] = " ".join(text_parts) if text_parts else None
        else:
            # Simple text response
            data["content"] = content
            if content:
                data["content_parts"] = [
                    ContentPart(type=ModalityType.TEXT, text=content)
                ]

        data["finish_reason"] = first_choice.get("finish_reason")

        # Extract tool calls
        tool_calls = message.get("tool_calls")
        if tool_calls:
            data["tool_calls"] = tool_calls

    # Extract token usage
    if "usage" in data:
        usage = data["usage"]
        data["prompt_tokens"] = usage.get("prompt_tokens")
        data["completion_tokens"] = usage.get("completion_tokens")
        data["total_tokens"] = usage.get("total_tokens")
        data["image_tokens"] = usage.get("image_tokens")
        data["audio_tokens"] = usage.get("audio_tokens")
        data["video_tokens"] = usage.get("video_tokens")

    # Set response_id and model
    data["response_id"] = data.get("id")

    # Store raw response
    data["raw_response"] = data.copy()

    super().__init__(**data)

timestamp property

timestamp

Convert Unix timestamp to datetime

Gemini Response (GeminiResponse)

maticlib.llm.google_genai.gemini_classes.GeminiResponse

Bases: LLMResponseBase

Gemini-specific response structure. Supports multimodal inputs (text, image, audio, video) and outputs. Inherits from LLMResponseBase and adds Gemini-specific fields.

Gemini-Specific Properties

cached_token_count

Get cached content token count (Gemini context caching).

thoughts_token_count

Get the thoughts token count if available (Gemini-specific).

maticlib.llm.google_genai.gemini_classes.GeminiResponse

GeminiResponse(**data)

Bases: LLMResponseBase

Gemini-specific response structure.

Supports multimodal inputs (text, image, audio, video) and outputs. Inherits from LLMResponseBase and adds Gemini-specific fields.

Attributes:

Name	Type	Description
responseId	str	Unique identifier for the Gemini response.
modelVersion	str	Gemini model version.
candidates	List[GeminiCandidate]	List of candidate responses.
usageMetadata	GeminiUsageMetadata	Token usage metadata.
cached_token_count	Optional[int]	Get cached content token count (Gemini context caching).
thoughts_token_count	Optional[int]	Get the thoughts token count if available (Gemini-specific).

Source code in maticlib/llm/google_genai/gemini_classes.py

def __init__(self, **data):
    # Extract common fields from Gemini structure
    if "candidates" in data and len(data["candidates"]) > 0:
        first_candidate = data["candidates"][0]
        parts = first_candidate.get("content", {}).get("parts", [])

        if parts:
            content_parts = []
            text_parts = []

            for part in parts:
                if isinstance(part, dict):
                    # Determine modality type
                    modality = ModalityType.TEXT
                    content_part = ContentPart(type=modality)

                    # Extract text
                    if part.get("text"):
                        text_parts.append(part["text"])
                        content_part.text = part["text"]

                    # Extract inline data (images, audio, etc.)
                    if part.get("inline_data"):
                        inline = part["inline_data"]
                        mime_type = inline.get("mime_type", "")
                        content_part.inline_data = inline

                        if "image" in mime_type:
                            modality = ModalityType.IMAGE
                        elif "audio" in mime_type:
                            modality = ModalityType.AUDIO
                        elif "video" in mime_type:
                            modality = ModalityType.VIDEO

                    # Extract file data
                    if part.get("file_data"):
                        file_data = part["file_data"]
                        mime_type = file_data.get("mime_type", "")

                        if "image" in mime_type:
                            modality = ModalityType.IMAGE
                            content_part.image_url = file_data.get("file_uri")
                        elif "audio" in mime_type:
                            modality = ModalityType.AUDIO
                            content_part.audio_url = file_data.get("file_uri")
                        elif "video" in mime_type:
                            modality = ModalityType.VIDEO
                            content_part.video_url = file_data.get("file_uri")

                    # Extract function calls (tools)
                    if part.get("functionCall"):
                        call = part["functionCall"]
                        if "tool_calls" not in data:
                            data["tool_calls"] = []
                        data["tool_calls"].append(
                            {
                                "id": None,  # Gemini doesn't always provide a call ID in the same way
                                "type": "function",
                                "function": {
                                    "name": call.get("name"),
                                    "arguments": call.get(
                                        "args"
                                    ),  # Gemini returns args as dict, not JSON string
                                },
                            }
                        )

                    content_part.type = modality
                    content_parts.append(content_part)

            data["content_parts"] = content_parts
            data["content"] = " ".join(text_parts) if text_parts else None

        data["finish_reason"] = first_candidate.get("finishReason")

    # Extract token usage with multimodal support
    if "usageMetadata" in data:
        usage = data["usageMetadata"]
        data["prompt_tokens"] = usage.get("promptTokenCount")
        data["completion_tokens"] = usage.get("candidatesTokenCount")
        data["total_tokens"] = usage.get("totalTokenCount")

        # Parse modality-specific tokens from promptTokensDetails
        if usage.get("promptTokensDetails"):
            for detail in usage["promptTokensDetails"]:
                modality = detail.get("modality", "").lower()
                token_count = detail.get("tokenCount", 0)

                if "image" in modality:
                    data["image_tokens"] = token_count
                elif "audio" in modality:
                    data["audio_tokens"] = token_count
                elif "video" in modality:
                    data["video_tokens"] = token_count

    # Set response_id and model
    data["response_id"] = data.get("responseId")
    data["model"] = data.get("modelVersion", "gemini")

    # Store raw response
    data["raw_response"] = data.copy()

    super().__init__(**data)

cached_token_count property

cached_token_count

Get cached content token count (Gemini context caching)

thoughts_token_count property

thoughts_token_count

Get the thoughts token count if available (Gemini-specific)