Temporal¶
Run an agent as a durable Temporal workflow: a failing tool is retried, and a worker that dies mid-run is replaced, without asking the model again for anything it already answered.
Normally an agent run lives in one process: if the process dies, or a tool fails halfway through, the run is lost and you pay for the model calls again. Temporal is a workflow engine that records every step of a run on a server. With one capability on the agent, each model request and each tool call becomes a recorded step. If a tool fails, Temporal retries just that step under a policy you set, and the model never sees the failure. If the worker dies, another picks the run up and replays the record, so steps that already finished are not repeated. The Temporal server runs here as a Docker service.
Use it when
- A run is long enough, or touches enough flaky systems (a carrier's API, a payment provider, a slow database), that a crash or an outage partway through would mean starting over and paying for the model calls again.
- You want to start a run, leave, and come back to it.
Look elsewhere when
- Runs are short and starting again is cheap: a plain agent is enough.
- You would rather not operate a Temporal server. This example uses a development server in Docker; production needs a real deployment.
your code ── start workflow ─▶ Temporal server ◀── polls ── worker
(its own container) runs the agent loop (workflow),
keeps the history every model request and every tool
call (activities)
What it shows
- Durability by one capability.
TemporalDurability()in the agent'scapabilitiesturns every model request and tool call into a Temporal activity and the agent loop into a workflow. The agent, its tools and its output type are written exactly as in a plain agent - A failing tool is retried, and the model never knows. The shipping carrier is down the first
time it is asked about each shipment. Temporal retries the call under
RETRY_POLICY(up to 5 attempts, backing off); the model sees only the answer. The tests check the server's own record: the model-request activity ran once per request, and the carrier call's recorded attempt is 2 - A worker that dies is replaced. Start a run, kill the worker mid-tool-call (the tests do both
a tidy shutdown and a
SIGKILLof a separate worker process), start another, and the run finishes. The model request that had already finished is not run again: the new worker replays the history - A run that can't hang. A bug in workflow code does not fail the run: Temporal keeps retrying
the failing step, and the caller waits (one mistyped attribute did this while building the example).
run_order_desksets an execution timeout (RUN_TIMEOUT) so a run that never finishes raisesRunTimedOutinstead; the tests check it with a model call that never returns - A server that isn't there:
TemporalUnavailablesays where it looked and how to start it - The same
RunResult. The workflow returns anAgentRunResult(its return annotation is what makes the client receive a real one), so.output,.usageand the messages are as for any agent - Tested against the real server. There is no offline Temporal: its test server downloads a
binary at run time. So the tests need the Docker service and are skipped without it; the release
check starts the service, runs them (with the model replaced by
TestModel) and then the live tests
See it run: sample_run.md is a recorded run against a real model: what each agent was asked, which tools it called, and what it returned.
It is the model's view, in which the carrier never failed: the retry is what the tests and the server's history show.
Running it¶
docker compose -f examples/temporal/service/docker-compose.yml up -d --wait
export TEMPORAL_ADDRESS=$(docker compose -f examples/temporal/service/docker-compose.yml port temporal 7233)
uv run --with "temporalio>=1.34,<2" python -m examples.temporal.agent
docker compose -f examples/temporal/service/docker-compose.yml down
Start it with python -m, not by file path: Temporal's sandbox imports the workflow's module by
name, so a workflow cannot live in __main__.
To use it in your project, add_agent.py copies the agent into agent/agents/, the service into
services/<name>/, and installs temporalio:
Putting the worker in its own process¶
run_order_desk runs a worker inside the calling process, which keeps the example in one file. In
production the worker is a separate, long-running process, so that a crash of your web server does not
stop the work and a crash of a worker does not stop the caller. That process is three lines:
client = await connect(os.environ["TEMPORAL_ADDRESS"])
await make_worker(client).run() # polls the task queue until you stop it
The tests run exactly that in a subprocess and SIGKILL it. Note that the server only learns a worker
is gone when the activity's timeout (ACTIVITY_CONFIG, 30 s) passes, so a hard kill delays the retry by
about that long; a tidy shutdown does not.
Things worth knowing¶
- Tools and deps run in activities, so they are serialized.
DeskDepscrosses into every activity, and a tool's arguments and results cross as JSON. State kept in module-level objects (like the carrier's call ledger here) is visible only because the demo's worker is in the same process - Spans:
LogfirePluginsends Temporal's own spans (the workflow and each activity) to Logfire. Pydantic AI'sinvoke_agentspan is not among them, so the span-based checks used by the other examples cannot see this agent run; the workflow's history on the server is the record - One workflow per agent. Two workflows that list the same agent register the same activities and the worker refuses to start
To adapt it, replace the tools and Carrier with your own, give them the retry behaviour you want in
RETRY_POLICY (a tool whose failure retrying cannot fix should raise a Temporal ApplicationError
with non_retryable=True, which is tried once and fails the run), and rewrite the instructions.
Source¶
All of it is in examples/temporal/.
"""Temporal: a durable agent whose run survives failing tools and crashing workers.
Use this pattern when:
- A run takes long enough, or touches enough flaky systems, that a crash or an outage partway through
would otherwise mean starting over (and paying for the model calls again)
- Tools call things that fail transiently (a carrier's API, a payment provider, a slow database)
- You want a run you can start, leave, and come back to: it is resumed by any worker, not lost
How it works: with the `TemporalDurability` capability, every model request and every tool call runs
as a Temporal *activity*, and the agent loop itself runs as a Temporal *workflow* whose history is
stored on the Temporal server. Two things follow, and the tests prove both:
a tool raises Temporal retries that activity (the policy is `RETRY_POLICY`); the model
is never asked again, and never sees the failure
the worker dies another worker picks the workflow up and replays its history: the model
requests and tool calls that already finished are not run again
The server is a service (`service/`, started by Docker Compose). Its address is a dependency
(`DeskDeps.temporal_address`, read from TEMPORAL_ADDRESS). `run_order_desk` starts a worker *in this
process* and runs one workflow, which keeps the example in one file. In a real system the worker is
its own long-running process; see the README for the three lines that split it out.
The stock and prices are invented. The shipping carrier fails the first time it is asked about each
shipment (`CARRIER.fail_first`) so the retry is visible; replace it with your own tools.
"""
import asyncio
import os
from dataclasses import dataclass, field
from datetime import timedelta
from typing import Annotated, Literal
from uuid import uuid4
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.agent import AgentRunResult
from pydantic_ai.capabilities import RaiseContentFilterError
from pydantic_ai.durable_exec.temporal import (
LogfirePlugin,
PydanticAIPlugin,
PydanticAIWorkflow,
TemporalDurability,
)
from pydantic_ai.usage import UsageLimits
from temporalio import workflow
from temporalio.client import Client, WorkflowFailureError
from temporalio.common import RetryPolicy
from temporalio.exceptions import TimeoutError as WorkflowTimeoutError
from temporalio.worker import Worker
from temporalio.workflow import ActivityConfig
# Temporal runs a workflow in a sandbox that re-executes this module, and the sandbox forbids what
# the template's own modules do at import time (reading .env, for one). Importing them as
# "passed through" makes the sandbox reuse the ones already loaded instead of running them again.
with workflow.unsafe.imports_passed_through():
from agent.config import settings
from agent.logging import agent_label, configure_logging, get_logger
from agent.prompts.templates import load_prompt
from agent.runs import RunResult, Step
logger = get_logger(__name__)
LABEL = agent_label(__name__) # names this agent's run spans in Logfire traces
USAGE_LIMITS = UsageLimits(
request_limit=8, total_tokens_limit=100_000, cost_limit=settings.cost_limit
)
# --- Temporal settings ---
# Temporal's own port, for a server you start yourself on it. The Docker service publishes on a random
# free host port instead, so set TEMPORAL_ADDRESS to the address `docker compose port` shows (see service/).
DEFAULT_ADDRESS = "127.0.0.1:7233"
TASK_QUEUE = "order-desk" # the queue the worker polls and the workflow is started on
RUN_TIMEOUT = timedelta(
minutes=2
) # a whole run; a workflow that never finishes is failed, not hung
# Every activity (a model request or a tool call) gets this timeout and this retry policy.
RETRY_POLICY = RetryPolicy(
initial_interval=timedelta(milliseconds=200), backoff_coefficient=1.5, maximum_attempts=5
)
ACTIVITY_CONFIG = ActivityConfig(
start_to_close_timeout=timedelta(seconds=30), retry_policy=RETRY_POLICY
)
def temporal_address_from_env() -> str:
return os.environ.get("TEMPORAL_ADDRESS", DEFAULT_ADDRESS)
# --- The data ---
Sku = Literal["widget", "gadget", "gizmo"]
Country = Literal["Norway", "Germany", "Canada"]
STOCK = {"widget": 120, "gadget": 0, "gizmo": 8}
BASE_SHIPPING_USD = {"Norway": 14.00, "Germany": 9.50, "Canada": 18.00}
PER_ITEM_USD = {"widget": 0.40, "gadget": 1.25, "gizmo": 2.10}
def quote_for(sku: str, quantity: int, country: str) -> float:
"""What the carrier charges: a base rate for the country plus a rate per item."""
return round(BASE_SHIPPING_USD[country] + PER_ITEM_USD[sku] * quantity, 2)
class CarrierUnavailable(ConnectionError):
"""The carrier's API did not answer. An ordinary exception: Temporal retries the activity."""
@dataclass
class Carrier:
"""A shipping carrier's API that is down the first `fail_first` times for each shipment.
`attempts` records every call as (shipment, succeeded), so a test can see the retries that
really happened. Replace this with your own client; the agent only sees the tool.
"""
fail_first: int = 1
attempts: list[tuple[str, bool]] = field(default_factory=list)
def reset(self) -> None:
self.attempts.clear()
async def quote(self, sku: str, quantity: int, country: str) -> float:
shipment = f"{quantity}x{sku}->{country}"
earlier = sum(1 for key, _ in self.attempts if key == shipment)
if earlier < self.fail_first:
self.attempts.append((shipment, False))
raise CarrierUnavailable(f"carrier API timed out quoting {shipment}")
self.attempts.append((shipment, True))
return quote_for(sku, quantity, country)
CARRIER = Carrier()
# --- Dependencies ---
@dataclass
class DeskDeps:
"""Runtime dependencies for the order desk. They are serialized into every activity."""
temporal_address: str = field(default_factory=temporal_address_from_env)
class TemporalUnavailable(Exception):
"""The Temporal server could not be reached."""
class RunTimedOut(Exception):
"""The workflow did not finish within its time limit."""
# --- Output type ---
class Answer(BaseModel):
# `result` is the conventional output field in these examples; the generated
# eval starter reads it when present (see evals/helpers.py).
result: str
shipping_usd: float | None = None
# --- Agent ---
# The capability makes the agent durable. The agent is built here, at module level, because the
# worker must know its activities before a workflow runs; model requests and tool calls are routed
# through activities, so they run on the worker, not inside the workflow.
order_desk_agent: Agent[DeskDeps, Answer] = Agent(
settings.model,
name=LABEL,
output_type=Answer,
deps_type=DeskDeps,
capabilities=[
RaiseContentFilterError(),
TemporalDurability(activity_config=ACTIVITY_CONFIG),
],
instructions=load_prompt("temporal"),
)
@order_desk_agent.tool_plain
async def check_stock(sku: Sku) -> int:
"""How many units of an item are in stock."""
return STOCK[sku]
@order_desk_agent.tool_plain
async def shipping_quote(
sku: Sku, quantity: Annotated[int, Field(ge=1, le=100)], country: Country
) -> float:
"""The carrier's shipping cost in US dollars for `quantity` units of an item to a country."""
return await CARRIER.quote(sku, quantity, country)
# --- Workflow ---
@workflow.defn
class OrderDeskWorkflow(PydanticAIWorkflow):
"""One agent run as a durable workflow. Its history is kept by the Temporal server."""
__pydantic_ai_agents__ = [order_desk_agent]
@workflow.run
async def run(self, user_input: str, deps: DeskDeps) -> AgentRunResult[Answer]:
# The return annotation matters: it is what makes the client receive a real AgentRunResult.
return await order_desk_agent.run(user_input, deps=deps, usage_limits=USAGE_LIMITS)
# --- Running it ---
async def connect(address: str) -> Client:
"""A client for the Temporal server at `address`, able to carry Pydantic AI's types.
`LogfirePlugin` sends Temporal's own spans (the workflow, and each model and tool activity) to
Logfire; without it a trace of a run shows none of them. It also keeps replays from emitting
spans a second time. Pydantic AI's `invoke_agent` span is not among them: the run happens in
a workflow and an activity, not in the caller, so the server's history is the record of it.
"""
try:
return await Client.connect(address, plugins=[PydanticAIPlugin(), LogfirePlugin()])
except RuntimeError as exc: # temporalio reports a refused connection as a RuntimeError
raise TemporalUnavailable(
f"No Temporal server at {address}. Start the service (`docker compose up -d --wait` "
"in the directory with its docker-compose.yml) and set TEMPORAL_ADDRESS to the "
"address Docker published, or point TEMPORAL_ADDRESS at your own Temporal server."
) from exc
def make_worker(client: Client, **options: object) -> Worker:
"""A worker that runs the order desk's workflow and the activities of its agent."""
return Worker(client, task_queue=TASK_QUEUE, workflows=[OrderDeskWorkflow], **options) # type: ignore[arg-type]
async def run_order_desk(
user_input: str,
deps: DeskDeps | None = None,
*,
workflow_id: str | None = None,
timeout: timedelta = RUN_TIMEOUT,
) -> RunResult[Answer]:
"""Answer an order question as a Temporal workflow, with a worker running in this process.
Returns:
A RunResult: `.output` is the `Answer`; its one step's result is the workflow's own result.
Raises:
TemporalUnavailable: When the server can't be reached.
RunTimedOut: When the workflow does not finish within `timeout`.
"""
if deps is None:
deps = DeskDeps()
logger.info(
"Running order desk", extra={"user_input": user_input, "server": deps.temporal_address}
)
client = await connect(deps.temporal_address)
async with make_worker(client):
try:
result: AgentRunResult[Answer] = await client.execute_workflow(
OrderDeskWorkflow.run,
args=[user_input, deps],
id=workflow_id or f"order-desk-{uuid4()}",
task_queue=TASK_QUEUE,
# Temporal retries a workflow task that fails forever, so a bug in workflow code
# would hang the caller. The execution timeout turns that into a failure.
execution_timeout=timeout,
)
except WorkflowFailureError as exc:
if isinstance(exc.cause, WorkflowTimeoutError):
raise RunTimedOut(f"the workflow did not finish within {timeout}") from exc
raise
return RunResult(
output=result.output, steps=[Step(agent=LABEL, result=result)], usage=result.usage
)
def main() -> None:
configure_logging()
question = "I'd like to order 3 gizmos for Norway. Are they in stock, and what is the shipping?"
run = asyncio.run(run_order_desk(question))
print(run.output)
print(f"{len(CARRIER.attempts)} carrier calls: {CARRIER.attempts}")
if __name__ == "__main__":
# Temporal cannot run a workflow that is defined in `__main__` (its sandbox imports the workflow's
# module by name), and the agent's name would differ too. So run this module again under its real
# name. That is why it is started with `python -m examples.temporal.agent`, not by file path.
import importlib
importlib.import_module(__spec__.name).main()
You answer customers' questions about an order: whether an item is in stock and what shipping costs.
- Use `check_stock` for stock and `shipping_quote` for shipping. Never guess a number; every figure
you give must come from a tool.
- Ask for a shipping quote only for items that are in stock in the quantity wanted.
- Fill `shipping_usd` with the quoted shipping cost in US dollars, or leave it empty when no quote
was asked for or the item cannot be shipped.
title = "Temporal"
pattern = "temporal"
summary = "Run an agent as a durable Temporal workflow: failing tools are retried and a crashed worker is replaced, without repeating the model calls that already finished."
smoke_input = "I'd like to order 3 gizmos for Norway. Are they in stock, and what is the shipping?"
# A real model must use both tools rather than guess (checked by the release check).
expected_tools = ["check_stock", "shipping_quote"]
# The one thing the agent needs beyond the template: the Temporal Python SDK. (Pydantic AI's
# `temporal` extra is only this package.)
dependencies = ["temporalio>=1.34,<2"]
# The server runs as a docker-compose service (service/). The release check starts it, finds the
# port Docker chose, and passes the address to the example in TEMPORAL_ADDRESS.
services = ["temporal"]
[service.temporal]
port = 7233
env = "TEMPORAL_ADDRESS"
url = "{address}"
[entrypoint]
deps = "DeskDeps"
run = "run_order_desk"
"""The durable order desk: its tools and carrier, and what Temporal really does when things fail.
The tests that run the workflow need a Temporal server: the release check starts the example's Docker
service and passes its address in TEMPORAL_ADDRESS (see example.toml); without it they are skipped.
(Temporal's own test server downloads a binary at run time, which a hermetic suite must not do.)
They run against the real server with the model replaced by `TestModel`, and check what the *server*
recorded: the workflow's event history says which activities ran, how often, and on which attempt.
Needs temporalio (declared in example.toml), so the module is skipped without it.
"""
import asyncio
import os
import subprocess
import sys
import uuid
from dataclasses import dataclass, field
from datetime import timedelta
from pathlib import Path
import pytest
pytest.importorskip("temporalio", reason="needs temporalio")
from pydantic_ai.messages import RetryPromptPart, ToolCallPart, ToolReturnPart # noqa: E402
from pydantic_ai.models.test import TestModel # noqa: E402
from temporalio.client import WorkflowFailureError # noqa: E402
from temporalio.exceptions import ApplicationError # noqa: E402
from agent.runs import RunResult # noqa: E402
from examples.temporal import agent as module # noqa: E402
from examples.temporal.agent import ( # noqa: E402
BASE_SHIPPING_USD,
PER_ITEM_USD,
STOCK,
Answer,
Carrier,
CarrierUnavailable,
DeskDeps,
RunTimedOut,
TemporalUnavailable,
check_stock,
connect,
make_worker,
quote_for,
run_order_desk,
shipping_quote,
temporal_address_from_env,
)
@pytest.fixture(autouse=True)
def fresh_carrier(monkeypatch):
"""Each test gets its own flaky carrier (one failure per shipment), so tests can't leak."""
carrier = Carrier(fail_first=1)
monkeypatch.setattr(module, "CARRIER", carrier)
return carrier
@pytest.fixture
def address() -> str:
server = os.environ.get("TEMPORAL_ADDRESS")
if not server:
pytest.skip("needs the Temporal service running (TEMPORAL_ADDRESS)")
return server
@pytest.fixture
def deps(address) -> DeskDeps:
return DeskDeps(temporal_address=address)
# --- The data and the tools, without Temporal ---
def test_a_quote_is_the_countrys_base_rate_plus_a_rate_per_item():
assert quote_for("gizmo", 3, "Norway") == 20.30 # 14.00 + 3 * 2.10
assert quote_for("widget", 10, "Germany") == 13.50 # 9.50 + 10 * 0.40
assert quote_for("gadget", 1, "Canada") == 19.25
def test_every_item_and_country_has_a_price():
assert set(STOCK) == set(PER_ITEM_USD)
for country in BASE_SHIPPING_USD:
for sku in PER_ITEM_USD:
assert quote_for(sku, 1, country) > BASE_SHIPPING_USD[country]
async def test_the_stock_tool_reports_the_catalogue():
assert await check_stock("widget") == 120
assert await check_stock("gadget") == 0
async def test_the_carrier_is_down_for_the_first_try_at_each_shipment_only(fresh_carrier):
with pytest.raises(CarrierUnavailable):
await shipping_quote("gizmo", 3, "Norway")
assert await shipping_quote("gizmo", 3, "Norway") == 20.30 # the retry succeeds
assert await shipping_quote("gizmo", 3, "Norway") == 20.30 # and keeps succeeding
with pytest.raises(CarrierUnavailable): # a different shipment starts out failing again
await shipping_quote("gizmo", 4, "Norway")
assert fresh_carrier.attempts == [
("3xgizmo->Norway", False),
("3xgizmo->Norway", True),
("3xgizmo->Norway", True),
("4xgizmo->Norway", False),
]
async def test_a_carrier_that_never_fails_makes_one_call(fresh_carrier):
fresh_carrier.fail_first = 0
assert await fresh_carrier.quote("widget", 2, "Canada") == 18.80
assert fresh_carrier.attempts == [("2xwidget->Canada", True)]
fresh_carrier.reset()
assert fresh_carrier.attempts == []
def test_a_carrier_outage_is_an_ordinary_connection_error():
"""Not a ModelRetry: the model must never see it. Temporal retries the activity instead."""
assert issubclass(CarrierUnavailable, ConnectionError)
def test_the_address_comes_from_the_environment(monkeypatch):
monkeypatch.setenv("TEMPORAL_ADDRESS", "temporal.internal:7233")
assert temporal_address_from_env() == "temporal.internal:7233"
assert DeskDeps().temporal_address == "temporal.internal:7233"
monkeypatch.delenv("TEMPORAL_ADDRESS")
assert temporal_address_from_env() == module.DEFAULT_ADDRESS
async def test_an_unreachable_server_is_reported_with_how_to_start_it():
with pytest.raises(TemporalUnavailable, match="docker compose"):
await run_order_desk("hello", DeskDeps(temporal_address="127.0.0.1:1"))
# --- Reading the server's record of a run ---
@dataclass
class Activity:
name: str
attempts: list[int] = field(default_factory=list) # the attempt number of each recorded start
completed: int = 0
failed: int = 0
async def activities(client, workflow_id: str) -> dict[str, Activity]:
"""What the server recorded: every activity the workflow scheduled, by type, with its attempts."""
scheduled: dict[int, Activity] = {}
by_name: dict[str, Activity] = {}
async for event in client.get_workflow_handle(workflow_id).fetch_history_events():
kind = event.WhichOneof("attributes")
if kind == "activity_task_scheduled_event_attributes":
name = event.activity_task_scheduled_event_attributes.activity_type.name
scheduled[event.event_id] = by_name.setdefault(name, Activity(name))
elif kind == "activity_task_started_event_attributes":
attrs = event.activity_task_started_event_attributes
scheduled[attrs.scheduled_event_id].attempts.append(attrs.attempt)
elif kind == "activity_task_completed_event_attributes":
scheduled[
event.activity_task_completed_event_attributes.scheduled_event_id
].completed += 1
elif kind == "activity_task_failed_event_attributes":
scheduled[event.activity_task_failed_event_attributes.scheduled_event_id].failed += 1
return by_name
def tool_activity(by_name: dict[str, Activity]) -> Activity:
(found,) = [a for name, a in by_name.items() if name.endswith("__call_tool")]
return found
def model_requests(by_name: dict[str, Activity]) -> Activity:
(found,) = [a for name, a in by_name.items() if name.endswith("__model_request")]
return found
def tool_calls(result: RunResult) -> list[ToolCallPart]:
return [
part
for message in result.all_messages()
for part in message.parts
if isinstance(part, ToolCallPart) and part.tool_name != "final_result"
]
QUESTION = "Order 3 gizmos for Norway: in stock, and what is the shipping?"
# --- A run through Temporal ---
async def test_a_run_goes_through_the_real_server_and_returns_a_normal_result(deps, fresh_carrier):
result = await run_order_desk(QUESTION, deps)
assert isinstance(result, RunResult)
assert isinstance(result.output, Answer)
assert [step.agent for step in result.steps] == ["temporal"]
assert result.usage.requests >= 2 # the tool calls, then the answer
assert {call.tool_name for call in tool_calls(result)} == {"check_stock", "shipping_quote"}
async def test_a_failing_tool_is_retried_and_the_model_never_sees_the_failure(deps, fresh_carrier):
result = await run_order_desk(QUESTION, deps)
# The carrier really failed once and then answered, for the shipment the model asked about.
outcomes = [ok for _, ok in fresh_carrier.attempts]
assert outcomes == [False, True]
# The tool result in the history is the carrier's real answer, for the arguments the model used.
[call] = [c for c in tool_calls(result) if c.tool_name == "shipping_quote"]
returned = [
part.content
for message in result.all_messages()
for part in message.parts
if isinstance(part, ToolReturnPart) and part.tool_name == "shipping_quote"
]
assert returned == [quote_for(**call.args_as_dict())]
# Nothing in the conversation asks the model to retry or mentions the outage.
assert not [
part
for message in result.all_messages()
for part in message.parts
if isinstance(part, RetryPromptPart)
]
async def test_the_server_recorded_the_retry_and_no_repeated_model_request(deps, fresh_carrier):
workflow_id = f"test-retry-{uuid.uuid4()}"
result = await run_order_desk(QUESTION, deps, workflow_id=workflow_id)
by_name = await activities(await connect(deps.temporal_address), workflow_id)
# Two tool calls, two activities. The server records the attempt that produced each result: the
# stock check on its first try, the carrier call on its second (the failed first try is not an
# event of its own; the carrier's own ledger, checked above, shows it).
tools = tool_activity(by_name)
assert sorted(tools.attempts) == [1, 2]
assert tools.completed == 2
assert tools.failed == 0
model = model_requests(by_name)
# Every model request ran exactly once, even though a tool failed in between.
assert model.attempts == [1] * result.usage.requests
assert model.completed == result.usage.requests
assert model.failed == 0
async def test_a_tool_that_keeps_failing_is_retried_to_the_limit_then_the_run_fails(
deps, fresh_carrier
):
fresh_carrier.fail_first = 99
workflow_id = f"test-exhausted-{uuid.uuid4()}"
with pytest.raises(WorkflowFailureError):
await run_order_desk(QUESTION, deps, workflow_id=workflow_id)
assert [ok for _, ok in fresh_carrier.attempts] == [
False
] * module.RETRY_POLICY.maximum_attempts
by_name = await activities(await connect(deps.temporal_address), workflow_id)
# The model was asked once for the tool calls and never again: the failure did not loop back to it.
assert model_requests(by_name).attempts == [1]
async def test_a_tool_that_raises_a_non_retryable_error_is_not_retried(
deps, fresh_carrier, monkeypatch
):
"""For a failure that retrying cannot fix (a rejected address, a refused payment)."""
class Rejects(Carrier):
async def quote(self, sku, quantity, country):
self.attempts.append((f"{quantity}x{sku}->{country}", False))
raise ApplicationError("the carrier does not ship there", non_retryable=True)
rejecting = Rejects()
monkeypatch.setattr(module, "CARRIER", rejecting)
with pytest.raises(WorkflowFailureError):
await run_order_desk(QUESTION, deps)
assert len(rejecting.attempts) == 1 # one try, not RETRY_POLICY's five
async def test_a_run_that_does_not_finish_in_time_is_failed_not_left_hanging(
deps, fresh_carrier, monkeypatch
):
async def stalled(self, *args, **kwargs):
await asyncio.sleep(30)
monkeypatch.setattr(TestModel, "request", stalled)
with pytest.raises(RunTimedOut, match="did not finish"):
await run_order_desk(QUESTION, deps, timeout=timedelta(seconds=2))
# --- A worker that dies mid-run ---
@dataclass
class HeldCarrier(Carrier):
"""A carrier whose first call hangs until released, so a test can kill the worker mid-call."""
in_flight: asyncio.Event = field(default_factory=asyncio.Event)
release: asyncio.Event = field(default_factory=asyncio.Event)
async def quote(self, sku: str, quantity: int, country: str) -> float:
if not self.release.is_set():
self.in_flight.set()
await self.release.wait()
return await super().quote(sku, quantity, country)
async def test_a_second_worker_finishes_a_run_whose_first_worker_died(deps, monkeypatch):
held = HeldCarrier(fail_first=0)
monkeypatch.setattr(module, "CARRIER", held)
asked: list[object] = []
real_request = TestModel.request
async def counting_request(self, *args, **kwargs):
asked.append(args[0] if args else None)
return await real_request(self, *args, **kwargs)
monkeypatch.setattr(TestModel, "request", counting_request)
client = await connect(deps.temporal_address)
workflow_id = f"test-crash-{uuid.uuid4()}"
async with make_worker(client, graceful_shutdown_timeout=timedelta(0)):
handle = await client.start_workflow(
module.OrderDeskWorkflow.run,
args=[QUESTION, deps],
id=workflow_id,
task_queue=module.TASK_QUEUE,
execution_timeout=timedelta(seconds=90),
)
await asyncio.wait_for(
held.in_flight.wait(), 30
) # the model has answered; a tool is mid-call
asked_before_crash = len(asked)
# The first worker is gone and the call it was making was cancelled. The workflow is still open.
assert not held.release.is_set()
assert asked_before_crash == 1
held.release.set()
async with make_worker(client):
result = await asyncio.wait_for(handle.result(), 60)
assert isinstance(result.output, Answer)
# The second worker did not ask the model what to do again: the first answer was in the history.
assert len(asked) == result.usage.requests == 2
by_name = await activities(client, workflow_id)
assert model_requests(by_name).attempts == [1, 1]
assert max(tool_activity(by_name).attempts) >= 2 # the interrupted tool call was run again
# The worker as its own process (what the README tells you to run in production), killed outright.
WORKER_PROCESS = """
import asyncio, os, pathlib
from examples.temporal import agent as m
class Hangs(m.Carrier):
async def quote(self, sku, quantity, country):
pathlib.Path(os.environ["IN_FLIGHT"]).write_text("the carrier call has started")
await asyncio.sleep(3600)
m.CARRIER = Hangs()
async def main():
client = await m.connect(os.environ["TEMPORAL_ADDRESS"])
await m.make_worker(client).run()
asyncio.run(main())
"""
REPO = Path(__file__).resolve().parents[2]
async def test_a_run_survives_its_worker_process_being_killed_outright(
deps, fresh_carrier, tmp_path
):
"""SIGKILL, not a tidy shutdown: nothing is reported to the server, which only notices when the
activity's timeout passes (so this takes about as long as ACTIVITY_CONFIG's)."""
fresh_carrier.fail_first = 0
in_flight = tmp_path / "in_flight"
worker = subprocess.Popen(
[sys.executable, "-c", WORKER_PROCESS],
cwd=REPO,
env={
**os.environ,
"AGENT_MODEL": "test",
"TEMPORAL_ADDRESS": deps.temporal_address,
"IN_FLIGHT": str(in_flight),
"PYTHONPATH": str(REPO),
},
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
)
client = await connect(deps.temporal_address)
workflow_id = f"test-sigkill-{uuid.uuid4()}"
try:
handle = await client.start_workflow(
module.OrderDeskWorkflow.run,
args=[QUESTION, deps],
id=workflow_id,
task_queue=module.TASK_QUEUE,
execution_timeout=timedelta(minutes=3),
)
for _ in range(600): # up to 60 s for the worker to start and reach the carrier call
if in_flight.exists() or worker.poll() is not None:
break
await asyncio.sleep(0.1)
assert in_flight.exists(), (
f"the worker never reached the tool call: {worker.stderr.read()[-1500:]}"
)
finally:
worker.kill()
worker.wait(timeout=10)
worker.stderr.close()
async with make_worker(client): # a new worker, in this process, with a working carrier
result = await asyncio.wait_for(handle.result(), 120)
assert isinstance(result.output, Answer)
by_name = await activities(client, workflow_id)
# The model request that finished before the kill ran once; it was not repeated by the new worker.
assert model_requests(by_name).attempts == [1] * result.usage.requests == [1, 1]
# The tool call that was in flight when the worker died was started again, by the new one.
assert max(tool_activity(by_name).attempts) >= 2
"""Live check: a real model runs as a durable Temporal workflow against the real server. `-m eval`.
The release check (scripts/release_check.py) builds and starts the Temporal service, then runs this
with its address in TEMPORAL_ADDRESS. To run it by hand: start the service, then set the variable:
docker compose -f examples/temporal/service/docker-compose.yml up -d --wait
export TEMPORAL_ADDRESS=$(docker compose -f examples/temporal/service/docker-compose.yml port temporal 7233)
"""
import os
import uuid
import pytest
pytest.importorskip("temporalio", reason="needs temporalio")
from pydantic_ai.messages import RetryPromptPart, ToolCallPart, ToolReturnPart # noqa: E402
from evals.trace import traced_run # noqa: E402
from examples.live_support import assert_every_agent_ran, run_as_script # noqa: E402
from examples.temporal import agent as module # noqa: E402
from examples.temporal.agent import ( # noqa: E402
CARRIER,
DeskDeps,
connect,
quote_for,
run_order_desk,
)
from examples.temporal.test_example import ( # noqa: E402
activities,
model_requests,
tool_activity,
)
pytestmark = pytest.mark.eval
if not os.environ.get("TEMPORAL_ADDRESS"):
pytest.skip("needs the Temporal service running (TEMPORAL_ADDRESS)", allow_module_level=True)
@pytest.fixture(autouse=True)
def flaky_carrier():
"""The carrier is down on the first try of each shipment, as it is in the demo."""
CARRIER.reset()
CARRIER.fail_first = 1
yield
CARRIER.reset()
async def ask(question: str):
"""Run one question; returns the traced run, its workflow id, and the carrier calls it made."""
workflow_id = f"live-{uuid.uuid4()}"
async def run(text: str):
return await run_order_desk(text, workflow_id=workflow_id)
traced = await traced_run(run, question)
return traced, workflow_id
def tool_names(result) -> set[str]:
return {
part.tool_name
for message in result.all_messages()
for part in message.parts
if isinstance(part, ToolReturnPart) and part.tool_name != "final_result"
}
async def test_the_quote_is_right_even_though_the_carrier_failed_first():
traced, _ = await ask(
"I'd like to order 3 gizmos for Norway. Are they in stock, and what is the shipping?"
)
output = traced.result.output
expected = quote_for("gizmo", 3, "Norway") # 20.30, worked out independently of the model
assert output.shipping_usd == pytest.approx(expected)
assert f"{expected:.2f}" in output.result
assert {"check_stock", "shipping_quote"} <= tool_names(traced.result)
# The carrier really did fail on the first call and succeed on a later one.
outcomes = [ok for _, ok in CARRIER.attempts]
assert outcomes[0] is False and outcomes[-1] is True
async def test_the_model_never_saw_the_failure_and_was_never_asked_twice_for_one_answer():
traced, workflow_id = await ask("What does it cost to ship 2 widgets to Germany?")
result = traced.result
assert not [
part
for message in result.all_messages()
for part in message.parts
if isinstance(part, RetryPromptPart)
]
by_name = await activities(await connect(DeskDeps().temporal_address), workflow_id)
model = model_requests(by_name)
# The server ran each model request exactly once, and as many as the run's usage reports.
assert model.attempts == [1] * result.usage.requests
assert model.failed == 0
# …while the tool call behind the carrier was retried: the server's record of it says attempt 2.
assert max(tool_activity(by_name).attempts) >= 2
async def test_an_item_that_is_out_of_stock_is_not_quoted():
traced, _ = await ask("Can I order 5 gadgets to Canada? Tell me about stock and shipping.")
output = traced.result.output
assert "check_stock" in tool_names(traced.result)
# Gadgets are out of stock: whatever the model says, it must not invent a price for them.
assert output.shipping_usd is None or output.shipping_usd == pytest.approx(
quote_for("gadget", 5, "Canada")
)
assert any(
isinstance(part, ToolCallPart) and part.tool_name == "check_stock"
for message in traced.result.all_messages()
for part in message.parts
)
async def test_the_agent_ran():
"""The span-based check can't see a run inside a workflow, so use the server's own record: the
activities of an agent are named after it, and its model requests ran and completed."""
traced, workflow_id = await ask("Is the widget in stock?")
by_name = await activities(await connect(DeskDeps().temporal_address), workflow_id)
ran = {name.split("__")[1] for name in by_name if name.startswith("agent__")}
assert model_requests(by_name).completed >= 1
assert_every_agent_ran(module, ran | {step.agent for step in traced.result.steps})
async def test_the_demo_script_runs():
out = await run_as_script("examples.temporal.agent")
assert "shipping_usd=" in out
assert "carrier calls" in out
# A Temporal development server, as a service. Start it by hand with:
# docker compose up -d --wait
# scripts/release_check.py starts and stops it for you when it checks the example. (Its web UI is
# on port 8233 inside the container: publish it too if you want to watch the workflows run.)
services:
temporal:
# The CLI's own version tag, pinned so the release check runs against the same server every time.
image: temporalio/temporal:1.8.0
command: ["server", "start-dev", "--ip", "0.0.0.0", "--headless"]
ports:
# Publish on 127.0.0.1 only, on a free host port (the first field is left empty so Docker
# picks one); `docker compose port temporal 7233` says which.
- "127.0.0.1::7233"
healthcheck:
# Healthy once the server answers a cluster health check, so `up --wait` returns when it is ready.
test: ["CMD", "temporal", "operator", "cluster", "health", "--address", "127.0.0.1:7233"]
interval: 1s
timeout: 3s
retries: 40
Recorded run · gemini-3.1-flash-lite · 1 step · $0.0005
Recorded 2026-10-07 with google:gemini-3.1-flash-lite · 1 step · 1,413 tokens · $0.0005 · 4.4 s.
Model output varies between runs. Regenerate with uv run python scripts/record_example.py temporal.
Input
I'd like to order 3 gizmos for Norway. Are they in stock, and what is the shipping?
Steps
1. temporal
1,413 tokens · $0.0005
Prompt
I'd like to order 3 gizmos for Norway. Are they in stock, and what is the shipping?
What happened
- called check_stock({"sku": "gizmo"})
- check_stock returned: 8
- called shipping_quote({"quantity": 3, "country": "Norway", "sku": "gizmo"})
- shipping_quote returned: 20.3
Output
{
"result": "Yes, 3 gizmos are in stock, and the shipping cost to Norway is $20.30.",
"shipping_usd": 20.3
}
Result
run_order_desk(...).output