"""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,
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
from ..usage import RequestUsage
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.
"""
# 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',
}
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')))
for part in tool_results:
text = part.model_response_str() if isinstance(part, ToolReturnPart) else part.model_response()
xai_messages.append(tool_result(text))
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('AudioUrl/BinaryContent with audio is not supported by xAI SDK')
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))
else:
raise RuntimeError(f'Unsupported binary content type: {item.media_type}')
elif isinstance(item, AudioUrl):
raise NotImplementedError('AudioUrl is not supported by xAI SDK')
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 by xAI SDK')
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)
# 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['cache_read_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'}}
extracted = RequestUsage.extract(
dict(model=model, usage=usage_data),
provider=provider,
provider_url=provider_url,
provider_fallback='x_ai', # Pricing file is defined as x_ai.yml
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
]
}
