xAI
Install
To use XaiModel, you need to either install pydantic-ai, or install pydantic-ai-slim with the xai optional group:
pip install "pydantic-ai-slim[xai]"
uv add "pydantic-ai-slim[xai]"
Configuration
To use xAI models from xAI through their API, go to console.x.ai to create an API key.
docs.x.ai contains a list of available xAI models.
Environment variable
Once you have the API key, you can set it as an environment variable:
export XAI_API_KEY='your-api-key'
You can then use XaiModel by name:
from pydantic_ai import Agent
agent = Agent('xai:grok-4-1-fast-non-reasoning')
...
Or initialise the model directly:
from pydantic_ai import Agent
from pydantic_ai.models.xai import XaiModel
# Uses XAI_API_KEY environment variable
model = XaiModel('grok-4-1-fast-non-reasoning')
agent = Agent(model)
...
You can also customize the XaiModel with a custom provider:
from pydantic_ai import Agent
from pydantic_ai.models.xai import XaiModel
from pydantic_ai.providers.xai import XaiProvider
# Custom API key
provider = XaiProvider(api_key='your-api-key')
model = XaiModel('grok-4-1-fast-non-reasoning', provider=provider)
agent = Agent(model)
...
Or with a custom xai_sdk.AsyncClient:
from xai_sdk import AsyncClient
from pydantic_ai import Agent
from pydantic_ai.models.xai import XaiModel
from pydantic_ai.providers.xai import XaiProvider
xai_client = AsyncClient(api_key='your-api-key')
provider = XaiProvider(xai_client=xai_client)
model = XaiModel('grok-4-1-fast-non-reasoning', provider=provider)
agent = Agent(model)
...
X Search
xAI models support searching X (formerly Twitter) for real-time posts and content. The recommended way to enable it is with the XSearch capability, which configures the underlying x_search native tool and can be passed alongside any other capabilities on the agent. See the xAI X Search documentation for the full list of supported options.
from datetime import datetime
from pydantic_ai import Agent
from pydantic_ai.models.xai import XSearch
agent = Agent(
'xai:grok-4-1-fast',
capabilities=[
XSearch(
allowed_x_handles=['OpenAI', 'AnthropicAI', 'dasfacc'],
from_date=datetime(2024, 1, 1),
to_date=datetime(2024, 12, 31),
enable_image_understanding=True,
enable_video_understanding=True,
include_output=True,
)
],
)
result = agent.run_sync('What have AI companies been posting about?')
print(result.output)
"""
OpenAI announced their latest model updates, while Anthropic shared research on AI safety...
"""
(This example is complete, it can be run "as is")
The XSearch capability accepts:
allowed_x_handles/excluded_x_handles: filter results to (or away from) up to 10 X handles. These are mutually exclusive.from_date/to_date: restrict results to posts created within the given datetime range (naive datetimes are interpreted as UTC).enable_image_understanding(default:False): analyze images attached to posts.enable_video_understanding(default:False): analyze video content attached to posts.include_output(default:False): include the raw X search results on theNativeToolReturnPartavailable viaModelResponse.native_tool_calls. Without this, the model uses the search results internally but only returns its text summary; enabling it gives programmatic access to the searched posts, sources, and metadata.
As an alternative to the capability, you can pass the lower-level XSearchTool directly via capabilities=[NativeTool(XSearchTool(...))] — see the X Search Tool documentation — or enable raw output globally via the XaiModelSettings.xai_include_x_search_output model setting.
Streaming cancellation
Cancellation limitations
The xai-sdk SDK exposes streaming responses only as an async iterator, with no separate handle for cancelling the underlying gRPC call. Because of a Python language rule on async generators, cancel() cannot interrupt an in-flight chunk read while another coroutine is iterating the stream. Pydantic AI marks the response with state='interrupted', but upstream generation may continue until the surrounding async with agent.run_stream(...) block exits.
For reliable cancellation, either pass debounce_by=None to stream_text(), stream_output(), or stream_response() and call cancel() from the same task that's iterating:
from pydantic_ai import Agent
agent = Agent('xai:grok-4-1-fast-non-reasoning')
def should_stop(chunk: str) -> bool:
return len(chunk) > 100
async def main():
async with agent.run_stream('Write a long essay about Python') as result:
async for chunk in result.stream_text(debounce_by=None):
if should_stop(chunk):
await result.cancel()
break
Or, if you need to keep debouncing, wrap the stream with contextlib.aclosing so the iterator is closed before cancel() runs:
from contextlib import aclosing
from pydantic_ai import Agent
agent = Agent('xai:grok-4-1-fast-non-reasoning')
def should_stop(chunk: str) -> bool:
return len(chunk) > 100
async def main():
async with agent.run_stream('Write a long essay about Python') as result:
async with aclosing(result.stream_text()) as stream:
async for chunk in stream:
if should_stop(chunk):
break
await result.cancel()
Calling cancel() from a different task while iteration is in progress is not currently reliable on this provider.