xai.py

"""xAI model implementation using [xAI SDK](https://github.com/xai-org/xai-sdk-python)."""

import json
from collections import defaultdict
from collections.abc import AsyncIterator, Iterator, Sequence
from contextlib import asynccontextmanager
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Literal, cast

from typing_extensions import assert_never

from .. import _utils
from .._output import OutputObjectDefinition
from .._run_context import RunContext
from ..builtin_tools import CodeExecutionTool, MCPServerTool, WebSearchTool
from ..exceptions import UnexpectedModelBehavior, UserError
from ..messages import (
    AudioUrl,
    BinaryContent,
    BuiltinToolCallPart,
    BuiltinToolReturnPart,
    CachePoint,
    DocumentUrl,
    FilePart,
    FinishReason,
    ImageUrl,
    ModelMessage,
    ModelRequest,
    ModelRequestPart,
    ModelResponse,
    ModelResponsePart,
    ModelResponseStreamEvent,
    RetryPromptPart,
    SystemPromptPart,
    TextPart,
    ThinkingPart,
    ToolCallPart,
    ToolReturnPart,
    UploadedFile,
    UserContent,
    UserPromptPart,
    VideoUrl,
)
from ..models import (
    Model,
    ModelRequestParameters,
    StreamedResponse,
    check_allow_model_requests,
    download_item,
)
from ..profiles import ModelProfileSpec
from ..profiles.grok import GrokModelProfile
from ..providers import Provider, infer_provider
from ..settings import ModelSettings, ThinkingLevel
from ..usage import RequestUsage

XAI_EFFORT_MAP: dict[ThinkingLevel, Literal['low', 'high']] = {
    True: 'high',
    'minimal': 'low',
    'low': 'low',
    'medium': 'high',
    'high': 'high',
    'xhigh': 'high',
}
"""Maps unified thinking values to xAI reasoning_effort. xAI only supports 'low' and 'high'."""

try:
    import xai_sdk.chat as chat_types
    from xai_sdk import AsyncClient
    from xai_sdk.chat import assistant, file, image, system, tool, tool_result, user
    from xai_sdk.proto import chat_pb2, sample_pb2, usage_pb2
    from xai_sdk.tools import code_execution, get_tool_call_type, mcp, web_search  # x_search not yet supported
    from xai_sdk.types.model import ChatModel
except ImportError as _import_error:
    raise ImportError(
        'Please install `xai-sdk` to use the xAI model, '
        'you can use the `xai` optional group — `pip install "pydantic-ai-slim[xai]"`'
    ) from _import_error

XaiModelName = str | ChatModel
"""Possible xAI model names."""

_FINISH_REASON_MAP: dict[str, FinishReason] = {
    'stop': 'stop',
    'length': 'length',
    'content_filter': 'content_filter',
    'max_output_tokens': 'length',
    'cancelled': 'error',
    'failed': 'error',
}

# `GetChatCompletionResponse.outputs[*].finish_reason` uses the proto enum (ints), not the string values returned by
# `Response.finish_reason`.
_FINISH_REASON_PROTO_MAP: dict[int, FinishReason] = {
    sample_pb2.FinishReason.REASON_STOP: 'stop',
    sample_pb2.FinishReason.REASON_MAX_LEN: 'length',
    sample_pb2.FinishReason.REASON_TOOL_CALLS: 'tool_call',
}


class XaiModelSettings(ModelSettings, total=False):
    """Settings specific to xAI models.

    See [xAI SDK documentation](https://docs.x.ai/docs) for more details on these parameters.
    """

    xai_logprobs: bool
    """Whether to return log probabilities of the output tokens or not."""

    xai_top_logprobs: int
    """An integer between 0 and 20 specifying the number of most likely tokens to return at each position."""

    xai_user: str
    """A unique identifier representing your end-user, which can help xAI to monitor and detect abuse."""

    xai_store_messages: bool
    """Whether to store messages on xAI's servers for conversation continuity."""

    xai_previous_response_id: str
    """The ID of the previous response to continue the conversation."""

    xai_include_encrypted_content: bool
    """Whether to include the encrypted content in the response.

    Corresponds to the `use_encrypted_content` value of the model settings in the Responses API.
    """

    xai_include_code_execution_output: bool
    """Whether to include the code execution results in the response.

    Corresponds to the `code_interpreter_call.outputs` value of the `include` parameter in the Responses API.
    """

    xai_include_web_search_output: bool
    """Whether to include the web search results in the response.

    Corresponds to the `web_search_call.action.sources` value of the `include` parameter in the Responses API.
    """

    xai_include_inline_citations: bool
    """Whether to include inline citations in the response.

    Corresponds to the `inline_citations` option in the xAI `include` parameter.
    """

    xai_include_mcp_output: bool
    """Whether to include the MCP results in the response.

    Corresponds to the `mcp_call.outputs` value of the `include` parameter in the Responses API.
    """

    xai_reasoning_effort: Literal['low', 'high']
    """Reasoning effort level for Grok reasoning models.

    See https://docs.x.ai for details.
    """


# Mapping of XaiModelSettings keys to xAI SDK parameter names.
# Most keys are the same, but some differ (e.g., 'stop_sequences' -> 'stop').
_XAI_MODEL_SETTINGS_MAPPING: dict[str, str] = {
    'temperature': 'temperature',
    'top_p': 'top_p',
    'max_tokens': 'max_tokens',
    'stop_sequences': 'stop',
    'parallel_tool_calls': 'parallel_tool_calls',
    'presence_penalty': 'presence_penalty',
    'frequency_penalty': 'frequency_penalty',
    'xai_logprobs': 'logprobs',
    'xai_top_logprobs': 'top_logprobs',
    'xai_user': 'user',
    'xai_store_messages': 'store_messages',
    'xai_previous_response_id': 'previous_response_id',
    'xai_reasoning_effort': 'reasoning_effort',
}


class XaiModel(Model):
    """A model that uses the xAI SDK to interact with xAI models."""

    _model_name: str
    _provider: Provider[AsyncClient]

    def __init__(
        self,
        model_name: XaiModelName,
        *,
        provider: Literal['xai'] | Provider[AsyncClient] = 'xai',
        profile: ModelProfileSpec | None = None,
        settings: ModelSettings | None = None,
    ):
        """Initialize the xAI model.

        Args:
            model_name: The name of the xAI model to use (e.g., "grok-4-1-fast-non-reasoning")
            provider: The provider to use for API calls. Defaults to `'xai'`.
            profile: Optional model profile specification. Defaults to a profile picked by the provider based on the model name.
            settings: Optional model settings.
        """
        self._model_name = model_name

        if isinstance(provider, str):
            provider = infer_provider(provider)
        self._provider = provider
        self.client = provider.client

        super().__init__(settings=settings, profile=profile or provider.model_profile(model_name))

    @property
    def model_name(self) -> str:
        """The model name."""
        return self._model_name

    @property
    def system(self) -> str:
        """The model provider."""
        return 'xai'

    @classmethod
    def supported_builtin_tools(cls) -> frozenset[type]:
        """Return the set of builtin tool types this model can handle."""
        return frozenset({WebSearchTool, CodeExecutionTool, MCPServerTool})

    async def _map_messages(
        self,
        messages: list[ModelMessage],
        model_request_parameters: ModelRequestParameters,
    ) -> list[chat_types.chat_pb2.Message]:
        """Convert pydantic_ai messages to xAI SDK messages."""
        xai_messages: list[chat_types.chat_pb2.Message] = []
        # xAI expects tool results in the same order as tool calls.
        #
        # Pydantic AI doesn't guarantee tool-result part ordering, so we track
        # tool call order as we walk message history and reorder tool results.
        pending_tool_call_ids: list[str] = []

        for message in messages:
            if isinstance(message, ModelRequest):
                mapped_request_parts = await self._map_request_parts(
                    message.parts,
                    pending_tool_call_ids,
                )
                xai_messages.extend(mapped_request_parts)
            elif isinstance(message, ModelResponse):
                xai_messages.extend(self._map_response_parts(message.parts))
                pending_tool_call_ids.extend(
                    part.tool_call_id for part in message.parts if isinstance(part, ToolCallPart) and part.tool_call_id
                )
            else:
                assert_never(message)

        # Insert instructions as a system message after existing system messages if present
        if instructions := self._get_instructions(messages, model_request_parameters):
            system_prompt_count = sum(1 for m in xai_messages if m.role == chat_types.chat_pb2.MessageRole.ROLE_SYSTEM)
            xai_messages.insert(system_prompt_count, system(instructions))

        return xai_messages

    async def _map_request_parts(
        self,
        parts: Sequence[ModelRequestPart],
        pending_tool_call_ids: list[str],
    ) -> list[chat_types.chat_pb2.Message]:
        """Map ModelRequest parts to xAI messages."""
        xai_messages: list[chat_types.chat_pb2.Message] = []
        tool_results: list[ToolReturnPart | RetryPromptPart] = []

        for part in parts:
            if isinstance(part, SystemPromptPart):
                xai_messages.append(system(part.content))
            elif isinstance(part, UserPromptPart):
                if user_msg := await self._map_user_prompt(part):
                    xai_messages.append(user_msg)
            elif isinstance(part, ToolReturnPart):
                tool_results.append(part)
            elif isinstance(part, RetryPromptPart):
                if part.tool_name is None:
                    xai_messages.append(user(part.model_response()))
                else:
                    tool_results.append(part)
            else:
                assert_never(part)

        # Sort tool results by requested order, then emit
        if tool_results:
            order = {id: i for i, id in enumerate(pending_tool_call_ids)}
            tool_results.sort(key=lambda p: order.get(p.tool_call_id, float('inf')))
            file_content: list[UserContent] = []
            for part in tool_results:
                if isinstance(part, ToolReturnPart):
                    text, files = part.model_response_str_and_user_content()
                    xai_messages.append(tool_result(text))
                    file_content.extend(files)
                else:
                    xai_messages.append(tool_result(part.model_response()))
            if file_content and (
                user_msg := await self._map_user_prompt(UserPromptPart(content=file_content))
            ):  # pragma: no branch
                xai_messages.append(user_msg)

        return xai_messages

    def _map_response_parts(self, parts: Sequence[ModelResponsePart]) -> list[chat_types.chat_pb2.Message]:
        """Map ModelResponse parts to xAI assistant messages (one message per part)."""
        messages: list[chat_types.chat_pb2.Message] = []

        # Track builtin tool calls by tool_call_id to update their status with return parts
        builtin_calls: dict[str, chat_types.chat_pb2.ToolCall] = {}

        for item in parts:
            if isinstance(item, TextPart):
                messages.append(assistant(item.content))
            elif isinstance(item, ThinkingPart):
                if (thinking_msg := self._map_thinking_part(item)) is not None:
                    messages.append(thinking_msg)
            elif isinstance(item, ToolCallPart):
                client_side_tool_call = self._map_tool_call(item)
                self._append_tool_call(messages, client_side_tool_call)
            elif isinstance(item, BuiltinToolCallPart):
                builtin_call = self._map_builtin_tool_call_part(item)
                if item.provider_name == self.system and builtin_call:
                    self._append_tool_call(messages, builtin_call)
                    # Track specific tool calls for status updates
                    # Note: tool_call_id is always truthy here since _map_builtin_tool_call_part
                    # returns None when tool_call_id is empty
                    if item.tool_call_id:  # pragma: no branch
                        builtin_calls[item.tool_call_id] = builtin_call
            elif isinstance(item, BuiltinToolReturnPart):
                if (
                    item.provider_name == self.system
                    and item.tool_call_id
                    and (details := item.provider_details) is not None
                    and details.get('status') == 'failed'
                    and (call := builtin_calls.get(item.tool_call_id))
                ):
                    call.status = chat_types.chat_pb2.TOOL_CALL_STATUS_FAILED
                    if error_msg := details.get('error'):
                        call.error_message = str(error_msg)
            elif isinstance(item, FilePart):
                # Files generated by models (e.g., from CodeExecutionTool) are not sent back
                pass
            else:
                assert_never(item)

        return messages

    @staticmethod
    def _append_tool_call(messages: list[chat_types.chat_pb2.Message], tool_call: chat_types.chat_pb2.ToolCall) -> None:
        """Append a tool call to the most recent tool-call assistant message, or create a new one.

        We keep tool calls grouped to avoid generating one assistant message per tool call.
        """
        if messages and messages[-1].tool_calls:
            messages[-1].tool_calls.append(tool_call)
        else:
            msg = assistant('')
            msg.tool_calls.append(tool_call)
            messages.append(msg)

    def _map_thinking_part(self, item: ThinkingPart) -> chat_types.chat_pb2.Message | None:
        """Map a `ThinkingPart` into a single xAI assistant message.

        - Native xAI thinking (with optional signature) is sent via `reasoning_content`/`encrypted_content`
        - Non-xAI (or non-native) thinking is preserved by wrapping in the model profile's thinking tags
        """
        if item.provider_name == self.system and (item.content or item.signature):
            msg = assistant('')
            if item.content:
                msg.reasoning_content = item.content
            if item.signature:
                msg.encrypted_content = item.signature
            return msg
        elif item.content:
            start_tag, end_tag = self.profile.thinking_tags
            return assistant('\n'.join([start_tag, item.content, end_tag]))
        else:
            return None

    def _map_tool_call(self, tool_call_part: ToolCallPart) -> chat_types.chat_pb2.ToolCall:
        """Map a ToolCallPart to an xAI SDK ToolCall."""
        return chat_types.chat_pb2.ToolCall(
            id=tool_call_part.tool_call_id,
            type=chat_types.chat_pb2.TOOL_CALL_TYPE_CLIENT_SIDE_TOOL,
            status=chat_types.chat_pb2.TOOL_CALL_STATUS_COMPLETED,
            function=chat_types.chat_pb2.FunctionCall(
                name=tool_call_part.tool_name,
                arguments=tool_call_part.args_as_json_str(),
            ),
        )

    def _map_builtin_tool_call_part(self, item: BuiltinToolCallPart) -> chat_types.chat_pb2.ToolCall | None:
        """Map a BuiltinToolCallPart to an xAI SDK ToolCall with appropriate type and status."""
        if not item.tool_call_id:
            return None

        if item.tool_name == CodeExecutionTool.kind:
            return chat_types.chat_pb2.ToolCall(
                id=item.tool_call_id,
                type=chat_types.chat_pb2.TOOL_CALL_TYPE_CODE_EXECUTION_TOOL,
                status=chat_types.chat_pb2.TOOL_CALL_STATUS_COMPLETED,
                function=chat_types.chat_pb2.FunctionCall(
                    name=CodeExecutionTool.kind,
                    arguments=item.args_as_json_str(),
                ),
            )
        elif item.tool_name == WebSearchTool.kind:
            return chat_types.chat_pb2.ToolCall(
                id=item.tool_call_id,
                type=chat_types.chat_pb2.TOOL_CALL_TYPE_WEB_SEARCH_TOOL,
                status=chat_types.chat_pb2.TOOL_CALL_STATUS_COMPLETED,
                function=chat_types.chat_pb2.FunctionCall(
                    name=WebSearchTool.kind,
                    arguments=item.args_as_json_str(),
                ),
            )
        elif item.tool_name.startswith(MCPServerTool.kind):
            # Extract server label from tool_name (format: 'mcp_server:server_label')
            server_label = item.tool_name.split(':', 1)[1] if ':' in item.tool_name else item.tool_name
            args_dict = item.args_as_dict() or {}
            # Extract tool_name and tool_args from the structured args (matches OpenAI/Anthropic pattern)
            actual_tool_name = args_dict.get('tool_name', '')
            tool_args = args_dict.get('tool_args', {})
            # Construct the full function name in xAI's format: 'server_label.tool_name'
            function_name = f'{server_label}.{actual_tool_name}' if actual_tool_name else server_label
            return chat_types.chat_pb2.ToolCall(
                id=item.tool_call_id,
                type=chat_types.chat_pb2.TOOL_CALL_TYPE_MCP_TOOL,
                status=chat_types.chat_pb2.TOOL_CALL_STATUS_COMPLETED,
                function=chat_types.chat_pb2.FunctionCall(
                    name=function_name,
                    arguments=json.dumps(tool_args),
                ),
            )
        return None

    async def _upload_file_to_xai(self, data: bytes, filename: str) -> str:
        """Upload a file to xAI files API and return the file ID.

        Args:
            data: The file content as bytes
            filename: The filename to use for the upload

        Returns:
            The file ID from xAI
        """
        uploaded_file = await self._provider.client.files.upload(data, filename=filename)
        return uploaded_file.id

    async def _map_user_prompt(self, part: UserPromptPart) -> chat_types.chat_pb2.Message | None:  # noqa: C901
        """Map a UserPromptPart to an xAI user message."""
        if isinstance(part.content, str):
            return user(part.content)

        # Handle complex content (images, text, etc.)
        content_items: list[chat_types.Content] = []

        for item in part.content:
            if isinstance(item, str):
                content_items.append(item)
            elif isinstance(item, ImageUrl):
                # Get detail from vendor_metadata if available
                detail: chat_types.ImageDetail = 'auto'
                if item.vendor_metadata and 'detail' in item.vendor_metadata:
                    detail = item.vendor_metadata['detail']
                image_url = item.url
                if item.force_download:
                    downloaded = await download_item(item, data_format='base64_uri', type_format='extension')
                    image_url = downloaded['data']
                content_items.append(image(image_url, detail=detail))
            elif isinstance(item, BinaryContent):
                if item.is_image:
                    # Convert binary content to data URI and use image()
                    image_detail: chat_types.ImageDetail = 'auto'
                    if item.vendor_metadata and 'detail' in item.vendor_metadata:
                        image_detail = item.vendor_metadata['detail']
                    content_items.append(image(item.data_uri, detail=image_detail))
                elif item.is_audio:
                    raise NotImplementedError('BinaryContent with audio is not supported in xAI user prompts')
                elif item.is_document:
                    # Upload document to xAI files API and reference it
                    filename = item.identifier or f'document.{item.format}'
                    file_id = await self._upload_file_to_xai(item.data, filename)
                    content_items.append(file(file_id))
                elif item.is_video:
                    raise NotImplementedError('BinaryContent with video is not supported in xAI user prompts')
                else:  # pragma: no cover
                    raise RuntimeError(f'Unsupported binary content type: {item.media_type}')
            elif isinstance(item, AudioUrl):
                raise NotImplementedError('AudioUrl is not supported in xAI user prompts')
            elif isinstance(item, DocumentUrl):
                # Download and upload to xAI files API
                downloaded = await download_item(item, data_format='bytes')
                filename = item.identifier or 'document'
                # Add extension if data_type is available from download
                if 'data_type' in downloaded and downloaded['data_type']:
                    filename = f'{filename}.{downloaded["data_type"]}'

                file_id = await self._upload_file_to_xai(downloaded['data'], filename)
                content_items.append(file(file_id))
            elif isinstance(item, VideoUrl):
                raise NotImplementedError('VideoUrl is not supported in xAI user prompts')
            elif isinstance(item, UploadedFile):
                if item.provider_name != self.system:
                    raise UserError(
                        f'UploadedFile with `provider_name={item.provider_name!r}` cannot be used with XaiModel. '
                        f'Expected `provider_name` to be `{self.system!r}`.'
                    )
                content_items.append(file(item.file_id))
            elif isinstance(item, CachePoint):
                # xAI doesn't support prompt caching via CachePoint, so we filter it out
                pass
            else:
                assert_never(item)

        if content_items:
            return user(*content_items)

        return None

    async def _create_chat(
        self,
        messages: list[ModelMessage],
        model_settings: XaiModelSettings,
        model_request_parameters: ModelRequestParameters,
    ) -> Any:
        """Create an xAI chat instance with common setup for both request and stream.

        Returns:
            The xAI SDK chat object, ready to call .sample() or .stream() on.
        """
        # Convert messages to xAI format
        xai_messages = await self._map_messages(messages, model_request_parameters)

        # Convert tools: combine built-in (server-side) tools and custom (client-side) tools
        tools: list[chat_types.chat_pb2.Tool] = []
        if model_request_parameters.builtin_tools:
            tools.extend(_get_builtin_tools(model_request_parameters))
        if model_request_parameters.tool_defs:
            tools.extend(_map_tools(model_request_parameters))
        tools_param = tools if tools else None

        # Set tool_choice based on whether tools are available and text output is allowed
        profile = GrokModelProfile.from_profile(self.profile)
        if not tools:
            tool_choice: Literal['none', 'required', 'auto'] | None = None
        elif not model_request_parameters.allow_text_output and profile.grok_supports_tool_choice_required:
            tool_choice = 'required'
        else:
            tool_choice = 'auto'

        # Set response_format based on the output_mode
        response_format: chat_pb2.ResponseFormat | None = None
        if model_request_parameters.output_mode == 'native':
            output_object = model_request_parameters.output_object
            assert output_object is not None
            response_format = _map_json_schema(output_object)
        elif (
            model_request_parameters.output_mode == 'prompted' and not tools and profile.supports_json_object_output
        ):  # pragma: no branch
            response_format = _map_json_object()

        # Map model settings to xAI SDK parameters
        xai_settings = _map_model_settings(model_settings)

        # Fall back to unified thinking when xai_reasoning_effort is not set
        if 'reasoning_effort' not in xai_settings and model_request_parameters.thinking is not None:
            thinking = model_request_parameters.thinking
            if thinking is not False:
                xai_settings['reasoning_effort'] = XAI_EFFORT_MAP[thinking]

        # Populate use_encrypted_content and include based on model settings
        include: list[chat_pb2.IncludeOption] = []
        use_encrypted_content = model_settings.get('xai_include_encrypted_content') or False
        if model_settings.get('xai_include_code_execution_output'):
            include.append(chat_pb2.IncludeOption.INCLUDE_OPTION_CODE_EXECUTION_CALL_OUTPUT)
        if model_settings.get('xai_include_web_search_output'):
            include.append(chat_pb2.IncludeOption.INCLUDE_OPTION_WEB_SEARCH_CALL_OUTPUT)
        if model_settings.get('xai_include_inline_citations'):
            include.append(chat_pb2.IncludeOption.INCLUDE_OPTION_INLINE_CITATIONS)
        # x_search not yet supported
        # collections_search not yet supported (could be mapped to file search)
        if model_settings.get('xai_include_mcp_output'):
            include.append(chat_pb2.IncludeOption.INCLUDE_OPTION_MCP_CALL_OUTPUT)

        # Create and return chat instance
        return self._provider.client.chat.create(
            model=self._model_name,
            messages=xai_messages,
            tools=tools_param,
            tool_choice=tool_choice,
            response_format=response_format,
            use_encrypted_content=use_encrypted_content,
            include=include,
            **xai_settings,
        )

    async def request(
        self,
        messages: list[ModelMessage],
        model_settings: ModelSettings | None,
        model_request_parameters: ModelRequestParameters,
    ) -> ModelResponse:
        """Make a request to the xAI model."""
        check_allow_model_requests()
        model_settings, model_request_parameters = self.prepare_request(
            model_settings,
            model_request_parameters,
        )

        chat = await self._create_chat(messages, cast(XaiModelSettings, model_settings or {}), model_request_parameters)
        response = await chat.sample()
        return self._process_response(response)

    @asynccontextmanager
    async def request_stream(
        self,
        messages: list[ModelMessage],
        model_settings: ModelSettings | None,
        model_request_parameters: ModelRequestParameters,
        run_context: RunContext[Any] | None = None,
    ) -> AsyncIterator[StreamedResponse]:
        """Make a streaming request to the xAI model."""
        check_allow_model_requests()
        model_settings, model_request_parameters = self.prepare_request(
            model_settings,
            model_request_parameters,
        )

        chat = await self._create_chat(messages, cast(XaiModelSettings, model_settings or {}), model_request_parameters)
        response_stream = chat.stream()
        yield await self._process_streamed_response(response_stream, model_request_parameters)

    def _process_response(self, response: chat_types.Response) -> ModelResponse:
        """Convert xAI SDK response to pydantic_ai ModelResponse.

        Processes response.proto.outputs to extract (in order):
        - ThinkingPart: For reasoning/thinking content
        - TextPart: For text content
        - ToolCallPart: For client-side tool calls
        - BuiltinToolCallPart + BuiltinToolReturnPart: For server-side (builtin) tool calls
        """
        parts: list[ModelResponsePart] = []
        outputs = response.proto.outputs

        for output in outputs:
            message = output.message

            # Add reasoning/thinking content if present
            if message.reasoning_content or message.encrypted_content:
                signature = message.encrypted_content or None
                parts.append(
                    ThinkingPart(
                        content=message.reasoning_content or '',
                        signature=signature,
                        provider_name=self.system if signature else None,
                    )
                )

            # Add text content from assistant messages
            if message.content and message.role == chat_types.chat_pb2.MessageRole.ROLE_ASSISTANT:
                part_provider_details: dict[str, Any] | None = None
                if output.logprobs and output.logprobs.content:
                    part_provider_details = {'logprobs': _map_logprobs(output.logprobs)}
                parts.append(TextPart(content=message.content, provider_details=part_provider_details))

            # Process tool calls in this output
            for tool_call in message.tool_calls:
                tool_result_content = _get_tool_result_content(message.content)
                _, part = _create_tool_call_part(
                    tool_call,
                    tool_result_content,
                    self.system,
                    message_role=message.role,
                )
                parts.append(part)

        # Convert usage with detailed token information
        usage = _extract_usage(response, self._model_name, self._provider.name, self._provider.base_url)

        # Map finish reason.
        #
        # The xAI SDK exposes `response.finish_reason` as a *string* for the overall response, but in
        # multi-output responses (e.g. server-side tools) it can reflect an intermediate TOOL_CALLS
        # output rather than the final STOP output. We derive the finish reason from the final output
        # when available.
        if outputs:
            last_reason = outputs[-1].finish_reason
            finish_reason = _FINISH_REASON_PROTO_MAP.get(last_reason, 'stop')
        else:  # pragma: no cover
            finish_reason = _FINISH_REASON_MAP.get(response.finish_reason, 'stop')

        return ModelResponse(
            parts=parts,
            usage=usage,
            model_name=self._model_name,
            timestamp=response.created,
            provider_name=self.system,
            provider_url=self._provider.base_url,
            provider_response_id=response.id,
            finish_reason=finish_reason,
        )

    async def _process_streamed_response(
        self,
        response: AsyncIterator[tuple[chat_types.Response, Any]],
        model_request_parameters: ModelRequestParameters,
    ) -> 'XaiStreamedResponse':
        """Process a streamed response, and prepare a streaming response to return."""
        peekable_response = _utils.PeekableAsyncStream(response)
        first_item = await peekable_response.peek()
        if isinstance(first_item, _utils.Unset):
            raise UnexpectedModelBehavior('Streamed response ended without content or tool calls')

        first_response, _ = first_item

        return XaiStreamedResponse(
            model_request_parameters=model_request_parameters,
            _model_name=self._model_name,
            _response=peekable_response,
            _timestamp=first_response.created,
            _provider=self._provider,
        )


@dataclass
class XaiStreamedResponse(StreamedResponse):
    """Implementation of `StreamedResponse` for xAI SDK."""

    _model_name: str
    _response: _utils.PeekableAsyncStream[tuple[chat_types.Response, chat_types.Chunk]]
    _timestamp: datetime
    _provider: Provider[AsyncClient]

    @property
    def system(self) -> str:
        """The model provider system name."""
        return self._provider.name

    @property
    def provider_url(self) -> str:
        """Get the provider base URL."""
        return self._provider.base_url

    def _update_response_state(self, response: chat_types.Response) -> None:
        """Update response state including usage, response ID, and finish reason."""
        # Update usage (SDK Response always provides a usage object)
        self._usage = _extract_usage(response, self._model_name, self._provider.name, self._provider.base_url)

        # Set provider response ID (only set once)
        if response.id and self.provider_response_id is None:
            self.provider_response_id = response.id

        # Handle finish reason (SDK Response always provides a finish_reason)
        self.finish_reason = _FINISH_REASON_MAP.get(response.finish_reason, 'stop')

    def _collect_reasoning_events(
        self,
        *,
        response: chat_types.Response,
        prev_reasoning_content: str,
        prev_encrypted_content: str,
    ) -> tuple[str, str, list[ModelResponseStreamEvent]]:
        """Collect thinking/reasoning events and return updated previous values.

        Note: xAI exposes reasoning via the accumulated Response object (not the per-chunk delta), so we compute
        deltas ourselves to avoid re-emitting the entire accumulated content on every chunk.
        """
        events: list[ModelResponseStreamEvent] = []

        if response.reasoning_content and response.reasoning_content != prev_reasoning_content:
            if response.reasoning_content.startswith(prev_reasoning_content):
                reasoning_delta = response.reasoning_content[len(prev_reasoning_content) :]
            else:
                reasoning_delta = response.reasoning_content
            prev_reasoning_content = response.reasoning_content
            if reasoning_delta:  # pragma: no branch
                events.extend(
                    self._parts_manager.handle_thinking_delta(
                        vendor_part_id='reasoning',
                        content=reasoning_delta,
                        # Only set provider_name when we have an encrypted signature to send back.
                        provider_name=self.system if response.encrypted_content else None,
                    )
                )

        if response.encrypted_content and response.encrypted_content != prev_encrypted_content:
            prev_encrypted_content = response.encrypted_content
            events.extend(
                self._parts_manager.handle_thinking_delta(
                    vendor_part_id='reasoning',
                    signature=response.encrypted_content,
                    provider_name=self.system,
                )
            )

        return prev_reasoning_content, prev_encrypted_content, events

    def _handle_server_side_tool_call(
        self,
        *,
        tool_call: chat_pb2.ToolCall,
        delta: chat_pb2.Delta,
        seen_tool_call_ids: set[str],
        seen_tool_return_ids: set[str],
        last_tool_return_content: dict[str, dict[str, Any] | str | None],
    ) -> Iterator[ModelResponseStreamEvent]:
        """Handle a single server-side tool call delta, yielding stream events."""
        builtin_tool_name = _get_builtin_tool_name(tool_call)

        if delta.role == chat_pb2.MessageRole.ROLE_ASSISTANT:
            # Emit the call part once per tool_call_id.
            if tool_call.id in seen_tool_call_ids:
                return
            seen_tool_call_ids.add(tool_call.id)

            if builtin_tool_name.startswith(MCPServerTool.kind):
                parsed_args = _build_mcp_tool_call_args(tool_call)
            else:
                parsed_args = _parse_tool_args(tool_call.function.arguments)
            call_part = BuiltinToolCallPart(
                tool_name=builtin_tool_name, args=parsed_args, tool_call_id=tool_call.id, provider_name=self.system
            )
            yield self._parts_manager.handle_part(vendor_part_id=tool_call.id, part=call_part)
            return

        if delta.role == chat_pb2.MessageRole.ROLE_TOOL:
            # Emit the return part once per tool_call_id.
            return_vendor_id = f'{tool_call.id}_return'
            tool_result_content = _get_tool_result_content(delta.content)
            if return_vendor_id in seen_tool_return_ids and tool_result_content == last_tool_return_content.get(
                return_vendor_id
            ):
                return
            seen_tool_return_ids.add(return_vendor_id)
            last_tool_return_content[return_vendor_id] = tool_result_content
            return_part = BuiltinToolReturnPart(
                tool_name=builtin_tool_name,
                content=tool_result_content,
                tool_call_id=tool_call.id,
                provider_name=self.system,
            )
            yield self._parts_manager.handle_part(vendor_part_id=return_vendor_id, part=return_part)

    async def _get_event_iterator(self) -> AsyncIterator[ModelResponseStreamEvent]:
        """Iterate over streaming events from xAI SDK."""
        # Local state to avoid re-emmiting duplicate events.
        prev_reasoning_content = ''
        prev_encrypted_content = ''
        seen_tool_call_ids: set[str] = set()
        seen_tool_return_ids: set[str] = set()
        last_tool_return_content: dict[str, dict[str, Any] | str | None] = {}
        # Track previous tool call args to compute deltas (like we do for reasoning content).
        prev_tool_call_args: dict[str, str] = {}

        async for response, chunk in self._response:
            self._update_response_state(response)

            prev_reasoning_content, prev_encrypted_content, reasoning_events = self._collect_reasoning_events(
                response=response,
                prev_reasoning_content=prev_reasoning_content,
                prev_encrypted_content=prev_encrypted_content,
            )
            for event in reasoning_events:
                yield event

            # Handle text content (property filters for ROLE_ASSISTANT)
            if chunk.content:
                for event in self._parts_manager.handle_text_delta(
                    vendor_part_id='content',
                    content=chunk.content,
                ):
                    yield event

            # Handle tool calls/tool results from *this chunk*.
            #
            # Important: xAI SDK `Response` is an accumulated view; `response.tool_calls` includes tool calls from
            # previous chunks. Iterating over it would re-emit tool calls repeatedly. Instead, we read tool calls
            # from the chunk's deltas which represent what changed in this frame.
            for output_chunk in chunk.proto.outputs:
                delta = output_chunk.delta
                if not delta.tool_calls:
                    continue
                for tool_call in delta.tool_calls:
                    if not tool_call.function.name:
                        continue

                    if tool_call.type != chat_pb2.ToolCallType.TOOL_CALL_TYPE_CLIENT_SIDE_TOOL:
                        for event in self._handle_server_side_tool_call(
                            tool_call=tool_call,
                            delta=delta,
                            seen_tool_call_ids=seen_tool_call_ids,
                            seen_tool_return_ids=seen_tool_return_ids,
                            last_tool_return_content=last_tool_return_content,
                        ):
                            yield event
                    else:
                        # Client-side tools: emit args as deltas so UI adapters receive PartDeltaEvents
                        # (not repeated PartStartEvents). Use accumulated args from response.tool_calls
                        # and compute the delta like we do for reasoning content.
                        accumulated = next((tc for tc in response.tool_calls if tc.id == tool_call.id), None)
                        accumulated_args = (
                            accumulated.function.arguments
                            if accumulated is not None and accumulated.function.arguments
                            else tool_call.function.arguments
                        )
                        prev_args = prev_tool_call_args.get(tool_call.id, '')
                        is_new_tool_call = tool_call.id not in prev_tool_call_args
                        args_changed = accumulated_args != prev_args

                        if is_new_tool_call or args_changed:
                            # Compute delta: if accumulated starts with prev, extract the new portion.
                            if accumulated_args.startswith(prev_args):
                                args_delta = accumulated_args[len(prev_args) :] or None
                            else:
                                args_delta = accumulated_args or None
                            prev_tool_call_args[tool_call.id] = accumulated_args
                            maybe_event = self._parts_manager.handle_tool_call_delta(
                                vendor_part_id=tool_call.id,
                                # Only pass tool_name on the first call; it would be appended otherwise.
                                tool_name=tool_call.function.name if is_new_tool_call else None,
                                args=args_delta,
                                tool_call_id=tool_call.id,
                            )
                            if maybe_event is not None:  # pragma: no branch
                                yield maybe_event

    @property
    def model_name(self) -> str:
        """Get the model name of the response."""
        return self._model_name

    @property
    def provider_name(self) -> str:
        """The model provider."""
        return self.system

    @property
    def timestamp(self) -> datetime:
        """Get the timestamp of the response."""
        return self._timestamp


def _map_json_schema(o: OutputObjectDefinition) -> chat_pb2.ResponseFormat:
    """Convert OutputObjectDefinition to xAI ResponseFormat protobuf object."""
    return chat_pb2.ResponseFormat(
        format_type=chat_pb2.FORMAT_TYPE_JSON_SCHEMA,
        schema=json.dumps(o.json_schema),
    )


def _map_json_object() -> chat_pb2.ResponseFormat:
    """Create a ResponseFormat for JSON object mode (prompted output)."""
    return chat_pb2.ResponseFormat(format_type=chat_pb2.FORMAT_TYPE_JSON_OBJECT)


def _map_model_settings(model_settings: XaiModelSettings) -> dict[str, Any]:
    """Map pydantic_ai ModelSettings to xAI SDK parameters."""
    return {
        _XAI_MODEL_SETTINGS_MAPPING[key]: value
        for key, value in model_settings.items()
        if key in _XAI_MODEL_SETTINGS_MAPPING
    }


def _map_tools(model_request_parameters: ModelRequestParameters) -> list[chat_types.chat_pb2.Tool]:
    """Convert pydantic_ai tool definitions to xAI SDK tools."""
    return [
        tool(
            name=tool_def.name,
            description=tool_def.description or '',
            parameters=tool_def.parameters_json_schema,
        )
        for tool_def in model_request_parameters.tool_defs.values()
    ]


def _get_builtin_tools(model_request_parameters: ModelRequestParameters) -> list[chat_types.chat_pb2.Tool]:
    """Convert pydantic_ai built-in tools to xAI SDK server-side tools."""
    tools: list[chat_types.chat_pb2.Tool] = []
    for builtin_tool in model_request_parameters.builtin_tools:
        if isinstance(builtin_tool, WebSearchTool):
            # xAI web_search supports:
            # - excluded_domains (from blocked_domains)
            # - allowed_domains
            # Note: user_location and search_context_size are not supported by xAI SDK
            tools.append(
                web_search(
                    excluded_domains=builtin_tool.blocked_domains,
                    allowed_domains=builtin_tool.allowed_domains,
                    enable_image_understanding=False,  # Not supported by PydanticAI
                )
            )
        elif isinstance(builtin_tool, CodeExecutionTool):
            # xAI code_execution takes no parameters
            tools.append(code_execution())
        elif isinstance(builtin_tool, MCPServerTool):
            # xAI mcp supports:
            # - server_url, server_label, server_description
            # - allowed_tool_names, authorization, extra_headers
            tools.append(
                mcp(
                    server_url=builtin_tool.url,
                    server_label=builtin_tool.id,
                    server_description=builtin_tool.description,
                    allowed_tool_names=builtin_tool.allowed_tools,
                    authorization=builtin_tool.authorization_token,
                    extra_headers=builtin_tool.headers,
                )
            )
        else:  # pragma: no cover
            # Defensive fallback - validation in models/__init__.py catches unsupported tools earlier
            raise UserError(
                f'`{builtin_tool.__class__.__name__}` is not supported by `XaiModel`. '
                f'Supported built-in tools: WebSearchTool, CodeExecutionTool, MCPServerTool. '
                f'If XSearchTool should be supported, please file an issue.'
            )
    return tools


def _get_builtin_tool_name(tool_call: chat_types.chat_pb2.ToolCall) -> str:
    """Get the PydanticAI tool name for an xAI builtin tool call.

    Maps xAI SDK tool call types to PydanticAI builtin tool names.

    Args:
        tool_call: The xAI SDK tool call.

    Returns:
        The PydanticAI tool name (e.g., 'web_search', 'code_execution').
    """
    tool_type = get_tool_call_type(tool_call)

    if tool_type == 'web_search_tool':
        return WebSearchTool.kind
    elif tool_type == 'code_execution_tool':
        return CodeExecutionTool.kind
    elif tool_type == 'mcp_tool':
        # Extract server label from "server_label.tool_name" format
        function_name = tool_call.function.name
        server_label = function_name.split('.', 1)[0] if '.' in function_name else function_name
        return f'{MCPServerTool.kind}:{server_label}'
    else:
        # x_search, collections_search, or unknown - use function name
        return tool_call.function.name


def _map_server_side_tools_used_to_name(server_side_tool: usage_pb2.ServerSideTool) -> str:
    """Map xAI SDK ServerSideTool enum from usage.server_side_tools_used to a tool name.

    Args:
        server_side_tool: The ServerSideTool enum value from usage.server_side_tools_used.

    Returns:
        The tool name (e.g., 'web_search', 'code_execution').
    """
    mapping = {
        usage_pb2.SERVER_SIDE_TOOL_WEB_SEARCH: WebSearchTool.kind,
        usage_pb2.SERVER_SIDE_TOOL_CODE_EXECUTION: CodeExecutionTool.kind,
        usage_pb2.SERVER_SIDE_TOOL_MCP: MCPServerTool.kind,
        usage_pb2.SERVER_SIDE_TOOL_X_SEARCH: 'x_search',
        usage_pb2.SERVER_SIDE_TOOL_COLLECTIONS_SEARCH: 'collections_search',
        usage_pb2.SERVER_SIDE_TOOL_VIEW_IMAGE: 'view_image',
        usage_pb2.SERVER_SIDE_TOOL_VIEW_X_VIDEO: 'view_x_video',
    }
    return mapping.get(server_side_tool, 'unknown')


def _extract_usage(
    response: chat_types.Response,
    model: str,
    provider: str,
    provider_url: str,
) -> RequestUsage:
    """Extract usage information from xAI SDK response.

    Extracts token counts and additional usage details including:
    - reasoning_tokens: Tokens used for model reasoning/thinking
    - cache_read_tokens: Tokens read from prompt cache
    - server_side_tools_used: Count of server-side (built-in) tools executed
    """
    usage_obj = response.usage

    # Build usage data dict with all integer fields for genai-prices extraction
    usage_data: dict[str, int] = {
        'prompt_tokens': usage_obj.prompt_tokens or 0,
        'completion_tokens': usage_obj.completion_tokens or 0,
    }

    # Add reasoning tokens if available (optional attribute)
    if usage_obj.reasoning_tokens:
        usage_data['reasoning_tokens'] = usage_obj.reasoning_tokens

    # Add cached prompt tokens if available (optional attribute)
    if usage_obj.cached_prompt_text_tokens:
        usage_data['cached_prompt_text_tokens'] = usage_obj.cached_prompt_text_tokens

    # Aggregate server-side tools used by PydanticAI builtin tool name
    if usage_obj.server_side_tools_used:
        tool_counts: dict[str, int] = defaultdict(int)
        for server_side_tool in usage_obj.server_side_tools_used:
            tool_name = _map_server_side_tools_used_to_name(server_side_tool)
            tool_counts[tool_name] += 1
        # Add each tool as a separate details entry (server_side_tools must be flattened to comply with details being dict[str, int])
        for tool_name, count in tool_counts.items():
            usage_data[f'server_side_tools_{tool_name}'] = count

    # Build details from non-standard fields
    details = {
        k: v
        for k, v in usage_data.items()
        if k not in {'prompt_tokens', 'completion_tokens', 'cached_prompt_text_tokens'}
    }

    extracted = RequestUsage.extract(
        dict(model=model, usage=usage_data),
        provider=provider,
        provider_url=provider_url,
        provider_fallback='x-ai',  # genai-prices provider ID is 'x-ai'
        details=details or None,
    )

    # Ensure token counts are set even if genai-prices extraction failed
    if extracted.input_tokens == 0 and usage_data['prompt_tokens']:
        extracted.input_tokens = usage_data['prompt_tokens']
    if extracted.output_tokens == 0 and usage_data['completion_tokens']:
        extracted.output_tokens = usage_data['completion_tokens']

    return extracted


def _get_tool_result_content(content: str) -> dict[str, Any] | str | None:
    """Extract tool result content from a content string.

    Args:
        content: The content string (may be JSON or plain text)

    Returns:
        Tool result content as dict (if JSON), string, or None if no content
    """
    if content:
        try:
            return json.loads(content)
        except (json.JSONDecodeError, TypeError):
            return content
    return None


def _parse_tool_args(arguments: str) -> dict[str, Any]:
    """Parse tool call arguments from JSON string to dict.

    Args:
        arguments: JSON string of tool arguments

    Returns:
        Parsed arguments as dict, or empty dict if parsing fails
    """
    try:
        return json.loads(arguments)
    except (json.JSONDecodeError, TypeError):
        return {}


def _build_mcp_tool_call_args(tool_call: chat_pb2.ToolCall) -> dict[str, Any]:
    """Build args dict for MCP server tool calls.

    Follows the same pattern as OpenAI and Anthropic for MCP tool calls,
    including the actual tool name in the args dict.

    Args:
        tool_call: The xAI SDK tool call.

    Returns:
        Args dict with action, tool_name, and tool_args.
    """
    # xAI provides function name in "server_label.tool_name" format
    function_name = tool_call.function.name
    actual_tool_name = function_name.split('.', 1)[1] if '.' in function_name else function_name
    return {
        'action': 'call_tool',
        'tool_name': actual_tool_name,
        'tool_args': _parse_tool_args(tool_call.function.arguments),
    }


def _create_tool_call_part(
    tool_call: chat_pb2.ToolCall,
    tool_result_content: dict[str, Any] | str | None,
    provider_name: str,
    message_role: chat_pb2.MessageRole | None = None,
) -> tuple[str, ModelResponsePart]:
    """Create a part for a tool call, returning (vendor_part_id, part).

    Handles both server-side (builtin) and client-side tool calls.

    Args:
        tool_call: The tool call from the xAI response.
        tool_result_content: The content for the tool result (extracted by caller).
        provider_name: The provider name for builtin tools.
        message_role: The role of the message containing the tool call, if available. For server-side tools in
            non-streamed responses, xAI emits tool *calls* on `ROLE_ASSISTANT` messages and tool *results* on
            `ROLE_TOOL` messages; in those cases, role should take precedence over tool status.

    Returns:
        Tuple of (vendor_part_id, part).
    """
    is_server_side_tool = tool_call.type != chat_pb2.ToolCallType.TOOL_CALL_TYPE_CLIENT_SIDE_TOOL

    if is_server_side_tool:
        builtin_tool_name = _get_builtin_tool_name(tool_call)
        status = tool_call.status
        provider_details: dict[str, Any] | None = None

        if status == chat_pb2.ToolCallStatus.TOOL_CALL_STATUS_FAILED:
            provider_details = {
                'status': 'failed',
                'error': tool_call.error_message,
            }

        # If we know the message role (non-streamed responses), use it to decide call vs return.
        # Note: FAILED is always a return part (it may be emitted on an assistant message without a separate ROLE_TOOL
        # message).
        if status == chat_pb2.ToolCallStatus.TOOL_CALL_STATUS_FAILED or message_role == chat_pb2.MessageRole.ROLE_TOOL:
            return (
                f'{tool_call.id}_return',
                BuiltinToolReturnPart(
                    tool_name=builtin_tool_name,
                    content=tool_result_content,
                    tool_call_id=tool_call.id,
                    provider_name=provider_name,
                    provider_details=provider_details,
                ),
            )
        else:
            if builtin_tool_name.startswith(MCPServerTool.kind):
                args = _build_mcp_tool_call_args(tool_call)
            else:
                args = _parse_tool_args(tool_call.function.arguments)
            return (
                tool_call.id,
                BuiltinToolCallPart(
                    tool_name=builtin_tool_name,
                    args=args,
                    tool_call_id=tool_call.id,
                    provider_name=provider_name,
                ),
            )
    else:
        # Client-side tool call
        return (
            tool_call.id,
            ToolCallPart(
                tool_name=tool_call.function.name,
                args=tool_call.function.arguments,
                tool_call_id=tool_call.id,
            ),
        )


def _map_logprobs(logprobs: chat_types.chat_pb2.LogProbs) -> dict[str, Any]:
    """Map xAI logprobs to a dictionary format."""
    return {
        'content': [
            {
                'token': lp.token,
                'bytes': list(lp.bytes) if lp.bytes else None,
                'logprob': lp.logprob,
                'top_logprobs': [
                    {
                        'token': tlp.token,
                        'bytes': list(tlp.bytes) if tlp.bytes else None,
                        'logprob': tlp.logprob,
                    }
                    for tlp in lp.top_logprobs
                ],
            }
            for lp in logprobs.content
        ]
    }