Add support for durable execution with Prefect

[desertaxle]

Summary

This PR introduces an integration with Prefect to enable durable, fault-tolerant execution of Pydantic AI agents.

Examples

Run a pydantic-ai agent as a Prefect flow

from pydantic_ai import Agent
from pydantic_ai.durable_exec.prefect import PrefectAgent

# Existing agent
agent = Agent('gpt-4o', name='my_agent')

# Enable durable execution
prefect_agent = PrefectAgent(agent)

# Use as normal - now with automatic persistence and recovery
result = await prefect_agent.run('What is the capital of Mexico?')

print(result.output) # The capital of Mexico is Mexico City

The PrefectAgent class wraps an Agent and instruments it so that .run calls are executed as a Prefect flow, and all tool and model calls are executed as Prefect tasks.

Customize underlying task behavior

prefect_agent = PrefectAgent(
    agent,
    model_task_config=TaskConfig(
        retries=3,
        retry_delay_seconds=[1.0, 2.0, 4.0],
        timeout_seconds=30.0,
    ),
)

Create an agent service to enable scheduling and remote execution

# Deploy agent with scheduled execution
prefect_agent.serve(
    name='daily-report-deployment',
    cron='0 9 * * *',  # Run daily at 9am
    parameters={'user_prompt': 'Generate daily report'},
    tags=['production'],
)

Implementation Details

The integration is implemented through:

• PrefectAgent: Wrapper class that extends WrapperAgent to intercept and wrap agent operations
• PrefectModel: Model wrapper that executes requests as Prefect tasks with event streaming support
• PrefectFunctionToolset: Wraps tool calls as tasks while preserving type safety
• A custom Prefect cache policy to ensure resumability of agent runs

Concerns

Complex Serialization Requirements

The .serve() method requires the Agent to be picklable because each run is executed in a separate process. I've added __getstate__ and __setstate__ implementations for PrefectAgent, but they are complex and may rely on too much internal pydantic-ai functionality.

[zzstoatzz]

a couple small thoughts

[zzstoatzz]

maybe we could mention the escape hatches you have, ie impl a serializer on the model or field type

[zzstoatzz]

maybe we don't need it, but looks like a special case of visit_collection

[zzstoatzz]

prefect.types.TaskRetryDelaySeconds