pydantic_ai.direct
Methods for making imperative requests to language models with minimal abstraction.
These methods allow you to make requests to LLMs where the only abstraction is input and output schema translation so you can use all models with the same API.
These methods are thin wrappers around Model implementations.
Synchronous wrapper for an async streaming response, running the whole stream on the caller’s event loop.
The stream uses the internal SyncStreamBridge to keep context-manager
and iterator lifecycles in stable tasks. Exiting the with block cancels the underlying request promptly
and closes the connection instead of waiting for the whole response to arrive.
This class must be used as a context manager with the with statement. The synchronous stream is created
when the with block is entered and must be used and closed on that thread.
Get the current state of the response.
Type: messages.ModelResponse
Get the usage of the response so far.
Type: RequestUsage
Get the model name of the response.
Type: str
Get the timestamp of the response.
Type: datetime
def __iter__() -> Iterator[messages.ModelResponseStreamEvent]
Stream the response as an iterable of ModelResponseStreamEvents.
Iterator[messages.ModelResponseStreamEvent]
@async
def model_request(
model: models.Model | models.KnownModelName | str,
messages: Sequence[messages.ModelMessage],
*,
model_settings: settings.ModelSettings | None = None,
model_request_parameters: models.ModelRequestParameters | None = None,
instrument: instrumented_models.InstrumentationSettings | bool | None = None,
) -> messages.ModelResponse
Make a non-streamed request to a model.
from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request
async def main():
model_response = await model_request(
'anthropic:claude-haiku-4-5',
[ModelRequest.user_text_prompt('What is the capital of France?')] # (1)
)
print(model_response)
'''
ModelResponse(
parts=[TextPart(content='The capital of France is Paris.')],
usage=RequestUsage(input_tokens=56, output_tokens=7),
model_name='claude-haiku-4-5',
timestamp=datetime.datetime(...),
)
'''See ModelRequest.user_text_prompt for details.
messages.ModelResponse — The model response and token usage associated with the request.
model : models.Model | models.KnownModelName | str
The model to make a request to. We allow str here since the actual list of allowed models changes frequently.
messages : Sequence[messages.ModelMessage]
Messages to send to the model
model_settings : settings.ModelSettings | None Default: None
optional model settings
model_request_parameters : models.ModelRequestParameters | None Default: None
optional model request parameters
Whether to instrument the request with OpenTelemetry/Logfire, if None the value from
logfire.instrument_pydantic_ai is used.
@async
def model_request_batch(
model: models.Model | models.KnownModelName | str,
requests: Sequence[batch_models.BatchRequest],
*,
max_requests: int | None = None,
max_resubmissions: int = DEFAULT_MAX_RESUBMISSIONS,
poll_interval: timedelta = DEFAULT_POLL_INTERVAL,
max_poll_interval: timedelta = DEFAULT_MAX_POLL_INTERVAL,
) -> list[messages.ModelResponse | ModelAPIError]
Send requests through the model’s batch API, and wait for their responses.
Batch APIs cost less, typically half, but can take up to a day to finish. The requests are submitted in as few
batches as the provider accepts, or max_requests allows, and each batch is checked on with a
growing interval until it ends. Cancelling the call, for example with asyncio.timeout, cancels the batches that
haven’t ended. A request’s request_timeout setting raises a UserError,
since one request in a batch can’t be timed out separately.
Requests that expired before the provider ran them, that a batch ended without a result for, or that failed with
a rate limit, timeout or server error are submitted again in new batches, up to max_resubmissions times, as a
provider SDK retries a live request. A cancelled request isn’t.
A request’s own error is returned in its place. An error creating a batch, checking on one or downloading its
results is raised from the call instead, after cancelling the batches that haven’t ended. A check that fails the way
a live request is retried on, like a dropped connection or a server error, is first tried again at the next poll,
up to 3 failures in a row, as with BatchModelRequests.
from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_batch
from pydantic_ai.models.batch import BatchRequest
async def main():
questions = {'france': 'What is the capital of France?', 'italy': 'What is the capital of Italy?'}
responses = await model_request_batch(
'anthropic:claude-haiku-5-5',
[
BatchRequest(custom_id=key, messages=[ModelRequest.user_text_prompt(question)])
for key, question in questions.items()
],
)
for key, response in zip(questions, responses):
print(key, response if isinstance(response, Exception) else response.text)
#> france The capital of France is Paris.
#> italy The capital of Italy is Rome.
To submit requests now and collect their results in another process, use the batch_* methods of
Model directly; see the low-level Model API.
Unlike model_request, the requests aren’t instrumented with
OpenTelemetry/Logfire.
list[messages.ModelResponse | ModelAPIError] — One item per request, in the same order: the request’s response, or the
ModelAPIError it failed with.
model : models.Model | models.KnownModelName | str
The model to send the requests to. We allow str here since the actual list of allowed models changes frequently.
requests : Sequence[batch_models.BatchRequest]
The requests to send, each with its own messages, settings and parameters. Their custom_ids must be unique.
The most requests per batch. Defaults to the provider’s limit, which also caps it.
max_resubmissions : int Default: DEFAULT_MAX_RESUBMISSIONS
How many times a request that expired, or that failed with a transient error, is submitted
again in a new batch. 0 returns the error instead.
How long to wait before first checking a batch. The interval grows by half after every check,
up to max_poll_interval.
The longest wait between checks of a batch.
def model_request_sync(
model: models.Model | models.KnownModelName | str,
messages: Sequence[messages.ModelMessage],
*,
model_settings: settings.ModelSettings | None = None,
model_request_parameters: models.ModelRequestParameters | None = None,
instrument: instrumented_models.InstrumentationSettings | bool | None = None,
) -> messages.ModelResponse
Make a Synchronous, non-streamed request to a model.
This is a convenience method that wraps model_request with
loop.run_until_complete(...). You therefore can’t use this method inside async code or if there’s an active event loop.
from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_sync
model_response = model_request_sync(
'anthropic:claude-haiku-4-5',
[ModelRequest.user_text_prompt('What is the capital of France?')] # (1)
)
print(model_response)
'''
ModelResponse(
parts=[TextPart(content='The capital of France is Paris.')],
usage=RequestUsage(input_tokens=56, output_tokens=7),
model_name='claude-haiku-4-5',
timestamp=datetime.datetime(...),
)
'''See ModelRequest.user_text_prompt for details.
messages.ModelResponse — The model response and token usage associated with the request.
model : models.Model | models.KnownModelName | str
The model to make a request to. We allow str here since the actual list of allowed models changes frequently.
messages : Sequence[messages.ModelMessage]
Messages to send to the model
model_settings : settings.ModelSettings | None Default: None
optional model settings
model_request_parameters : models.ModelRequestParameters | None Default: None
optional model request parameters
Whether to instrument the request with OpenTelemetry/Logfire, if None the value from
logfire.instrument_pydantic_ai is used.
def model_request_stream(
model: models.Model | models.KnownModelName | str,
messages: Sequence[messages.ModelMessage],
*,
model_settings: settings.ModelSettings | None = None,
model_request_parameters: models.ModelRequestParameters | None = None,
instrument: instrumented_models.InstrumentationSettings | bool | None = None,
) -> AbstractAsyncContextManager[models.StreamedResponse]
Make a streamed async request to a model.
from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_stream
async def main():
messages = [ModelRequest.user_text_prompt('Who was Albert Einstein?')] # (1)
async with model_request_stream('openai:gpt-5-mini', messages) as stream:
chunks = []
async for chunk in stream:
chunks.append(chunk)
print(chunks)
'''
[
PartStartEvent(index=0, part=TextPart(content='Albert Einstein was ')),
FinalResultEvent(tool_name=None, tool_call_id=None),
PartDeltaEvent(
index=0, delta=TextPartDelta(content_delta='a German-born theoretical ')
),
PartDeltaEvent(index=0, delta=TextPartDelta(content_delta='physicist.')),
PartEndEvent(
index=0,
part=TextPart(
content='Albert Einstein was a German-born theoretical physicist.'
),
),
]
'''See ModelRequest.user_text_prompt for details.
AbstractAsyncContextManager[models.StreamedResponse] — A stream response async context manager.
model : models.Model | models.KnownModelName | str
The model to make a request to. We allow str here since the actual list of allowed models changes frequently.
messages : Sequence[messages.ModelMessage]
Messages to send to the model
model_settings : settings.ModelSettings | None Default: None
optional model settings
model_request_parameters : models.ModelRequestParameters | None Default: None
optional model request parameters
Whether to instrument the request with OpenTelemetry/Logfire, if None the value from
logfire.instrument_pydantic_ai is used.
def model_request_stream_sync(
model: models.Model | models.KnownModelName | str,
messages: Sequence[messages.ModelMessage],
*,
model_settings: settings.ModelSettings | None = None,
model_request_parameters: models.ModelRequestParameters | None = None,
instrument: instrumented_models.InstrumentationSettings | bool | None = None,
) -> StreamedResponseSync
Make a streamed synchronous request to a model.
This is the synchronous version of model_request_stream.
It drives the asynchronous stream on the caller’s event loop while providing a synchronous iterator interface.
The returned context manager must be used and closed on the thread where the synchronous stream is created.
from pydantic_ai import ModelRequest
from pydantic_ai.direct import model_request_stream_sync
messages = [ModelRequest.user_text_prompt('Who was Albert Einstein?')]
with model_request_stream_sync('openai:gpt-5-mini', messages) as stream:
chunks = []
for chunk in stream:
chunks.append(chunk)
print(chunks)
'''
[
PartStartEvent(index=0, part=TextPart(content='Albert Einstein was ')),
FinalResultEvent(tool_name=None, tool_call_id=None),
PartDeltaEvent(
index=0, delta=TextPartDelta(content_delta='a German-born theoretical ')
),
PartDeltaEvent(index=0, delta=TextPartDelta(content_delta='physicist.')),
PartEndEvent(
index=0,
part=TextPart(
content='Albert Einstein was a German-born theoretical physicist.'
),
),
]
'''
StreamedResponseSync — A sync stream response context manager.
model : models.Model | models.KnownModelName | str
The model to make a request to. We allow str here since the actual list of allowed models changes frequently.
messages : Sequence[messages.ModelMessage]
Messages to send to the model
model_settings : settings.ModelSettings | None Default: None
optional model settings
model_request_parameters : models.ModelRequestParameters | None Default: None
optional model request parameters
Whether to instrument the request with OpenTelemetry/Logfire, if None the value from
logfire.instrument_pydantic_ai is used.