Prefect
Prefect is a workflow orchestration framework for building resilient data pipelines in Python, natively integrated with Pydantic AI.
Prefect 3.0 brings transactional semantics to your Python workflows, allowing you to group tasks into atomic units and define failure modes. If any part of a transaction fails, the entire transaction can be rolled back to a clean state.
- Flows are the top-level entry points for your workflow. They can contain tasks and other flows.
- Tasks are individual units of work that can be retried, cached, and monitored independently.
Prefect 3.0’s approach to transactional orchestration makes your workflows automatically idempotent: rerunnable without duplication or inconsistency across any environment. Every task is executed within a transaction that governs when and where the task’s result record is persisted. If the task runs again under an identical context, it will not re-execute but instead load its previous result.
The diagram below shows the overall architecture of an agentic application with Prefect. Prefect uses client-side task orchestration by default, with optional server connectivity for advanced features like scheduling and monitoring.
+---------------------+
| Prefect Server | (Monitoring,
| or Cloud | scheduling, UI,
+---------------------+ orchestration)
^
|
Flow state, | Schedule flows,
metadata, | track execution
logs |
|
+------------------------------------------------------+
| Application Process |
| +----------------------------------------------+ |
| | Flow (Agent.run) | |
| +----------------------------------------------+ |
| | | | |
| v v v |
| +-----------+ +------------+ +-------------+ |
| | Task | | Task | | Task | |
| | (Tool) | | (MCP Tool) | | (Model API) | |
| +-----------+ +------------+ +-------------+ |
| | | | |
| Cache & Cache & Cache & |
| persist persist persist |
| to to to |
| v v v |
| +----------------------------------------------+ |
| | Result Storage (Local FS, S3, etc.) | |
| +----------------------------------------------+ |
+------------------------------------------------------+
| | |
v v v
[External APIs, services, databases, etc.]
See the Prefect documentation for more information.
Add durable execution to any Agent by attaching the PrefectDurability capability. When the agent runs inside a Prefect flow, the capability routes model requests, tool calls, and MCP communication through Prefect tasks. To make a run durable, call agent.run() inside a @flow.
The agent stays a normal Agent everywhere — outside a Prefect flow the capability is transparent, and the original agent, model, and MCP server can still be used as normal.
See Streaming for event handling inside tasks and flow code.
Here is a simple but complete example of attaching durable execution to an agent. All it requires is to install Pydantic AI with Prefect:
pip install pydantic-ai[prefect]
uv add pydantic-ai[prefect]
Or if you’re using the slim package, you can install it with the prefect optional group:
pip install pydantic-ai-slim[prefect]
uv add pydantic-ai-slim[prefect]
from prefect import flow
from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectDurability
agent = Agent(
'openai:gpt-5.6-sol',
instructions="You're an expert in geography.",
name='geography', # (1)
capabilities=[PrefectDurability()], # (2)
)
@flow # (3)
async def answer(question: str) -> str:
result = await agent.run(question)
return result.output
async def main():
answer_text = await answer('What is the capital of Mexico?')
print(answer_text)
#> Mexico City (Ciudad de México, CDMX)The agent's name is used to uniquely identify its flows and tasks.
Attach durability via capabilities=[...]. The capability routes model requests, tool calls, and MCP communication through Prefect tasks when the agent runs inside a flow.
Wrap agent.run() in your own @flow to make the run durable.
(This example is complete, it can be run “as is” — you’ll need to add asyncio.run(main()) to run main)
Because the same agent works inside and outside a Prefect flow, PrefectDurability composes with all other capabilities without each needing a Prefect-specific wrapper variant.
For more information on how to use Prefect in Python applications, see their Python documentation.
PrefectAgent remains supported as the wrapper-agent alternative to PrefectDurability.
Any agent can be wrapped in a PrefectAgent to get a durable agent variant that routes model requests, tool calls, and MCP communication through Prefect tasks:
from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectAgent
agent = Agent('openai:gpt-5.6-sol', name='geography')
prefect_agent = PrefectAgent(agent) # Use `prefect_agent` in place of `agent`.
Migrating to the capability means attaching PrefectDurability and adding the flow decorator that PrefectAgent used to apply for you:
-prefect_agent = PrefectAgent(agent)
-result = await prefect_agent.run(prompt)
+agent = Agent(..., capabilities=[PrefectDurability()])
+
+@flow
+async def answer(prompt: str) -> str:
+ result = await agent.run(prompt)
+ return result.output
When using Prefect with Pydantic AI agents, there are a few important considerations to ensure workflows behave correctly.
Each agent instance must have a unique name so Prefect can correctly identify and track its flows and tasks.
Toolsets that implement their own tool listing and calling (i.e. FunctionToolset and MCPToolset) must have a unique id set, which is used to identify their tasks within the flow.
Agent.run(model=...) supports both model strings (like 'openai:gpt-5.6-sol') and model instances. A model instance can’t be serialized across the task boundary, so it’s sent as its model_id string and rebuilt inside the task. That faithfully reproduces model-name strings and models with standard providers, but not an instance whose exact behavior depends on a custom provider, client, or settings — pre-register those by passing a models dict to PrefectDurability and reference them by key (or pass the registered instance). The agent’s own model, set at construction, is always available as the default.
To customize how a model string is built — a custom provider, or per-user credentials carried on the run’s deps — add a ResolveModelId capability before PrefectDurability: it gets first crack at every string, and the resolver runs again inside the task with the run’s actual deps, so it must be deterministic for a given (model_id, deps) and must not perform external I/O.
Agent tools are automatically wrapped as Prefect tasks, which means they benefit from:
- Retry logic: Failed tool calls can be retried automatically
- Caching: Tool results are cached based on their inputs
- Observability: Tool execution is tracked in the Prefect UI
A default TaskConfig for all tools can be passed as tool_task_config to the PrefectDurability constructor. Per-tool config lives on the tool’s metadata field — PrefectDurability looks for a 'prefect' key. You can set the metadata directly on the tool definition, or apply it across a selection of tools via the SetToolMetadata capability. See the capabilities documentation for the full selector vocabulary.
from pydantic_ai import Agent
from pydantic_ai.capabilities import SetToolMetadata
from pydantic_ai.durable_exec.prefect import PrefectDurability, TaskConfig
from pydantic_ai.toolsets import FunctionToolset
toolset = FunctionToolset(id='research')
@toolset.tool(metadata={'prefect': TaskConfig(timeout_seconds=10.0)}) # (1)
def fetch_data(url: str) -> str: ...
@toolset.tool(metadata={'prefect': False}) # (2)
def simple_tool() -> str: ...
agent = Agent(
'openai:gpt-5.6-sol',
name='research',
toolsets=[toolset],
capabilities=[
SetToolMetadata( # (3)
tools=['fetch_data', 'fetch_dataset'],
prefect=TaskConfig(timeout_seconds=10.0),
),
PrefectDurability(tool_task_config=TaskConfig(retries=3)), # (4)
],
)Inline: declare the task config alongside the tool definition. Per-tool config merges on top of the base tool_task_config.
Set 'prefect': False to skip task wrapping entirely for that tool.
Selector-based: SetToolMetadata applies the same metadata across a selection of tools ('all', a name list, a dict, or a callable).
tool_task_config sets the default config for every tool.
Agent.run_stream(), Agent.run_stream_events(), and Agent.iter() work inside a Prefect flow, but their events are buffered rather than delivered in real time. The model stream runs inside the durable task, and its events are replayed to the flow after the task completes.
For handlers with I/O side effects, pass event_stream_handler= to PrefectDurability. Model events are delivered live inside each model-request task, while each tool event is delivered in its own event-handler task. Configure those per-event tasks with event_stream_handler_task_config=. As with any Prefect task, a handler may run more than once if a task retries, so keep its side effects idempotent.
Alternatively, register ProcessEventStream. Its handler runs in flow code and must be deterministic because it re-runs on flow replay. Tool and final-output events arrive live, while the real captured model events are replayed after each model request completes. For examples, see the streaming docs.
A durability event_stream_handler= and a separately registered ProcessEventStream are two distinct handlers, and each fires once. The durability handler receives live events inside the durable task, while ProcessEventStream sees the buffered replay in flow code.
A per-run handler passed to Agent.run(event_stream_handler=...) also runs flow-side against replayed model events.
Because the model stream is consumed inside the task, cancelling it from the flow side (e.g. with AgentStream.cancel()) is not available across the durable boundary.
When a provider pauses a model turn mid-flight (Anthropic pause_turn) or runs it as a server-side job that’s polled until it’s ready (OpenAI background mode), each segment runs in a separate model request task. The suspended ModelResponse and background job ID are checkpointed between segments, while the final response is merged and usage is recorded once. A message_history ending in a suspended response is passed to the first task. Size timeout_seconds in Task Configuration for one provider round trip. If an error abandons a suspended job, its provider teardown runs in a dedicated cancellation task.
Additional toolsets can be passed per run via agent.run(toolsets=...), but only toolsets that don’t need durable wrapping are supported: non-executing toolsets like ExternalToolset, whose tools are executed outside the agent run, and FunctionToolsets whose tools all opt out of task wrapping with metadata={'prefect': False}. Other executing toolsets (FunctionToolset and MCPToolset) and dynamic toolsets must be set when constructing the agent so their tasks are registered before the flow runs; passing them at runtime raises a UserError.
You can customize Prefect task behavior, such as retries and timeouts, by passing TaskConfig objects to the PrefectDurability constructor:
mcp_task_config: Configuration for MCP server communication tasksmodel_task_config: Configuration for model request tasksevent_stream_handler_task_config: Configuration for event stream handler taskstool_task_config: Default configuration for all tool calls (per-tool overrides go on the tool’s'prefect'metadata — see Tool Wrapping above)
Available TaskConfig options:
retries: Maximum number of retries for the task (default:0)retry_delay_seconds: Delay between retries in seconds (can be a single value or list for exponential backoff, default:1.0)timeout_seconds: Maximum time in seconds for the task to completecache_policy: Custom Prefect cache policy for the taskpersist_result: Whether to persist the task resultresult_storage: Prefect result storage for the task (e.g.,'s3-bucket/my-storage'or aWritableFileSystemblock)log_prints: Whether to log print statements from the task (default:False)
Example:
from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectDurability, TaskConfig
agent = Agent(
'openai:gpt-5.6-sol',
instructions="You're an expert in geography.",
name='geography',
capabilities=[
PrefectDurability(
model_task_config=TaskConfig(
retries=3,
retry_delay_seconds=[1.0, 2.0, 4.0], # Exponential backoff
timeout_seconds=30.0,
),
),
],
)
async def main():
result = await agent.run('What is the capital of France?')
print(result.output)
#> Paris
(This example is complete, it can be run “as is” — you’ll need to add asyncio.run(main()) to run main)
Pydantic AI and provider API clients have their own retry logic. When using Prefect, you may want to:
- Disable HTTP Request Retries in Pydantic AI
- Turn off your provider API client’s retry logic (e.g.,
max_retries=0on a custom OpenAI client) - Rely on Prefect’s task-level retry configuration for consistency
This prevents requests from being retried multiple times at different layers.
Prefect 3.0 provides built-in caching and transactional semantics. Tasks with identical inputs will not re-execute if their results are already cached, making workflows naturally idempotent and resilient to failures.
- Task inputs: Messages, settings, parameters, tool arguments, and serializable dependencies
Note: For user dependencies to be included in cache keys, they must be serializable (e.g., Pydantic models or basic Python types). Non-serializable dependencies are automatically excluded from cache computation.
Prefect provides a built-in UI for monitoring flow runs, task executions, and failures. You can:
- View real-time flow run status
- Debug failures with full stack traces
- Set up alerts and notifications
To access the Prefect UI, you can either:
- Use Prefect Cloud (managed service)
- Run a local Prefect server with
prefect server start
You can also use Pydantic Logfire for detailed observability. When using both Prefect and Logfire, you’ll get complementary views:
- Prefect: Workflow-level orchestration, task status, and retry history
- Logfire: Fine-grained tracing of agent runs, model requests, and tool invocations
When using Logfire with Prefect, you can enable distributed tracing to see spans for your Prefect runs included with your agent runs, model requests, and tool invocations.
For more information about Prefect monitoring, see the Prefect documentation.
To deploy and schedule a Prefect-durable agent, wrap it in a Prefect flow and use the flow’s serve() or deploy() methods:
from prefect import flow
from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectDurability
@flow
async def daily_report_flow(user_prompt: str):
"""Generate a daily report using the agent."""
agent = Agent( # (1)
'openai:gpt-5.6-sol',
name='daily_report_agent',
instructions='Generate a daily summary report.',
capabilities=[PrefectDurability()],
)
result = await agent.run(user_prompt)
return result.output
# Serve the flow with a daily schedule
if __name__ == '__main__':
daily_report_flow.serve(
name='daily-report-deployment',
cron='0 9 * * *', # Run daily at 9am
parameters={'user_prompt': "Generate today's report"},
tags=['production', 'reports'],
)Each flow run executes in an isolated process, and all inputs and dependencies must be serializable. Because Agent instances cannot be serialized, instantiate the agent inside the flow rather than at the module level.
The serve() method accepts scheduling options:
cron: Cron schedule string (e.g.,'0 9 * * *'for daily at 9am)interval: Schedule interval in seconds or as a timedeltarrule: iCalendar RRule schedule string
For production deployments with Docker, Kubernetes, or other infrastructure, use the flow’s deploy() method. See the Prefect deployment documentation for more information.