pydantic_ai.external_events
Wait inside an agent run until something outside it resolves a token the run handed out.
A run waiting on input from outside (an approval, an answer, a review) hands a resume token to an
ask function. ask either answers right away, as a terminal prompt or an automated policy does, or
delivers the token to whoever will answer, by posting a Slack message, sending an email or writing a
database row, and the run waits until code elsewhere resolves the token with the answer.
RequestInput is built on this.
By default the wait happens in the current process: the token is a random ID, and
resolve_external_event or
fail_external_event, called from anywhere in the
same process (such as your approval endpoint), ends it. The wait only lives as long as the process.
Inside a durable run, the durable engine supplies the wait instead, so it survives restarts and
holds no compute while it lasts. Engines whose API hands out tokens map onto it directly: AWS Lambda
durable functions call it a callback, Restate an awakeable, Trigger.dev a waitpoint token, Vercel
Workflow a hook. An engine installs an ExternalEventWaiter
for each durable agent run with external_event_scope.
In the durable runs of an engine that installs none, such as Temporal, DBOS or Prefect so far, the wait
is refused: a wait in process would not be recorded, so a replay would ask again, could get a different
answer than the one the run already acted on, and diverge from the engine’s history.
Engines whose external events are addressed by name rather than by token (Temporal signals, Azure
Durable Task external events, DBOS messages) can implement the same contract by encoding the
execution’s ID and an event key in the token they hand out.
Bases: Exception
The external event a run waited for will not arrive.
Raised by wait_for_external_event when the
wait’s timeout elapsed, or the token was resolved with a failure instead of a payload. The message
is the reason.
Bases: ABC
What a durable engine implements so a run can wait for an external event durably.
An engine installs one for each durable agent run with
external_event_scope. See
Waiting for external events.
@abstractmethod
@async
def wait(
name: str,
*,
ask: Callable[[str], Awaitable[JsonValue]],
timeout: timedelta | None,
) -> JsonValue
Create a resume token, pass it to ask, and unless ask answered, suspend the run until the token is resolved.
The engine must journal every part, so that a replay of the run returns the recorded payload
without creating a second token or calling ask again.
JsonValue — The payload ask returned, or else the JSON payload the token was resolved with.
name : str
A name for the wait in the engine’s history, stable across replays.
Called once with the token, inside a durable unit whose result the engine records.
It returns the payload when it answered right away, and the wait returns that
payload without suspending. It returns None when it handed the token to whoever
will resolve it, and the wait suspends until they do.
timeout : timedelta | None
How long to wait before giving up, or None to wait as long as the engine allows.
ExternalEventError— The timeout elapsed, or the token was resolved with a failure.
@async
def wait_for_external_event(
name: str,
*,
ask: Callable[[str], Awaitable[object | None] | object | None],
timeout: timedelta | None = None,
) -> JsonValue
Hand a resume token to ask, and return the answer ask gives or else the payload the token is resolved with.
A resume token is created and ask is called with it once. ask either answers right away, as a
terminal prompt or an automated policy does, by returning the payload, or hands the token to whoever
will answer, by posting to Slack, sending an email or writing a database row, and returns None. Then
code you write elsewhere delivers the answer by resolving the token. A payload of None can’t be
returned from ask; resolve the token with it instead.
Outside a durable run, the wait happens in this process, and
resolve_external_event or
fail_external_event ends it. It only lives
as long as the process does.
Inside a durable run on an engine that supports it, currently AWS Lambda durable functions through
Pydantic AI Harness’s AWSLambdaDurability, the engine records the token, the answer ask returns
and the payload, and suspends the run while it waits, so the wait survives restarts and a replay
doesn’t ask again. The token is then resolved through the engine’s API. Inside an agent run on a
durable engine that doesn’t support it, the engine refuses the wait with a UserError: a wait in
process would not be recorded, so a replay would ask again and could get a different answer than
the one the run already acted on, and a restart would lose the wait.
JsonValue — The JSON payload ask returned, or the token was resolved with.
name : str
A name for the wait in a durable engine’s history. Keep it stable across deployments.
Sync or async function called with the token, returning the answer or None. A sync
function runs in a worker thread, like a sync tool. An answer is converted to its JSON form,
as it is on its way through a durable engine. In a durable run, an exception it raises fails
the durable unit it runs in, which the engine retries according to its policy, so it must
return quickly and be safe to call again.
timeout : timedelta | None Default: None
How long to wait before giving up, or None to wait without a limit (or, in a
durable run, as long as the engine allows). In process, it also bounds ask: an async
ask still running when it elapses is cancelled.
ExternalEventError— The timeout elapsed, or the token was resolved with a failure.UserError— The wait was requested from inside a durable unit, such as a tool call, or inside a durable agent run on an engine that does not support waiting for external events; oraskreturned a value that can’t be converted to JSON.
def resolve_external_event(token: str, payload: object) -> None
End an in-process wait_for_external_event with payload.
Call it from anywhere in the process the run is waiting in, such as the endpoint your approval UI
posts to, from any thread. A token from a durable engine is resolved through that engine’s API
instead, such as AWS Lambda’s SendDurableExecutionCallbackSuccess.
payload is converted to its JSON form, as it would be on its way through a durable engine, so the
wait gets the same value in process as on an engine.
UserError—payloadcan’t be converted to JSON, or no in-process wait has this token: it is wrong, from another process, or its wait already ended because it was resolved, timed out or was cancelled.
def fail_external_event(token: str, reason: str) -> None
End an in-process wait_for_external_event without a payload.
The wait raises ExternalEventError with reason,
as it does when it times out. RequestInput then denies
the pending calls with reason in the message to the model.
UserError— No in-process wait has this token: it is wrong, from another process, or its wait already ended because it was resolved, timed out or was cancelled.
def external_event_scope(waiter: ExternalEventWaiter) -> Generator[ExternalEventWaiter]
Install waiter for the code inside, typically one durable agent run.
wait_for_external_event called inside
the scope waits through waiter instead of in the current process.
Generator[ExternalEventWaiter]