Build Durable Agents With Pydantic AI And Temporal

Blessed Josiah — Sat, 18 Jul 2026 11:00:00 +0000

Agentic applications can fail mid-run: a network call drops, a server process crashes, an LLM request times out. Without durability built in, these failures mean lost progress and repeated (wasted) LLM calls when the process has to start over.

Pydantic AI provides a native integration with Temporal, a durable execution engine. Temporal records each step of a workflow's execution as an event history, so if a process crashes, it can resume from where it left off instead of starting over.

Pydantic AI exposes this through the TemporalAgent class, which wraps an existing agent and converts its tool calls, MCP calls, and model requests into Temporal Activities.

from pydantic_ai import Agent
from pydantic_ai.durable_exec.temporal import TemporalAgent

planning_agent = Agent(
    model='anthropic:claude-fable-5',
)
temporal_agent = TemporalAgent(
    wrapped=planning_agent
)

You call .run(...) on temporal_agent the same way you would on the underlying agent. The difference is under the hood: each tool call, MCP call, or model request now runs as a Temporal Activity.

Three Core Concepts

Activity — a function that performs a unit of work: an LLM call, an API request, a file write. If it fails, Temporal can retry it based on a configured retry policy.

from temporalio import activity

@activity.defn
async def create_sandbox(user_id: str) -> str:
    sandbox = await AsyncSandbox.create(template="node-expo-builder")
    return sandbox.sandbox_id

Workflow — orchestrates Activities, defining what runs and in what order. Workflow code must be deterministic (no direct network calls, no random values, no reading the system clock), since Temporal replays it to reconstruct state.

from temporalio import workflow
from datetime import timedelta

@workflow.defn
class ExampleWorkflow:
    @workflow.run
    async def run(self, prompt: str) -> str:
        sandbox_id = await workflow.execute_activity(
            create_sandbox,
            args=[self.user_id],
            start_to_close_timeout=timedelta(minutes=2),
        )
        return sandbox_id

Worker — polls a task queue for scheduled work and executes it. A Workflow or Activity must be registered with a Worker before it can run.

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker

async def main():
    client = await Client.connect("localhost:7233")

    worker = Worker(
        client,
        task_queue="coding-agent",
        workflows=[ExampleWorkflow],
        activities=[create_sandbox],
    )
    await worker.run()

if __name__ == "__main__":
    asyncio.run(main())

The task queue name is the only link between the code that starts a Workflow and the Worker that executes it — they never call each other directly.

Workers run Activities and Workflows differently: Activities run their function body when scheduled, and may be retried on failure per their retry policy. Workflows are replayed against their event history each time they need to advance, which is what lets a Workflow resume correctly on a new Worker after a crash.

Signal and Query

Once a Workflow is running, Signal and Query let you interact with it from outside.

Signal sends data into a running Workflow, asynchronously, with no return value:

@workflow.signal
async def approve_plan(self) -> None:
    self._plan_approved = True

@workflow.signal
async def reject_plan(self, feedback: str = "") -> None:
    self._plan_approved = False
    self._plan_feedback = feedback

Signals only work while a Workflow is still running — once it's completed, failed, or been terminated, it can no longer receive one.

Query reads a Workflow's current state, synchronously and without side effects:

@workflow.query
def state(self) -> WorkflowState:
    return WorkflowState(
        status=self._status,
        project_name=self._project_name,
        preview_url=self._preview_url,
    )

Queries can also run against a completed Workflow, within your namespace's retention period. This doesn't extend to terminated Workflows, which aren't safely queryable.

Human In The Loop

Because Temporal keeps a full event history, a Workflow can pause and resume without losing state — a pattern suited to human approval steps.

from temporalio import workflow

@workflow.defn
class ExampleWorkflow:
    @workflow.run
    async def run(self):
        while True:
            self._plan_approved = None
            await workflow.wait_condition(lambda: self._plan_approved is not None)

            if self._plan_approved:
                break

workflow.wait_condition pauses execution until the given condition is true. Here, the Workflow waits until a Signal (like approve_plan above) sets self._plan_approved. If the process running this Workflow crashes while it's paused, a new Worker can pick it up and it resumes waiting at the same point.

This example has no timeout, so it waits indefinitely. In production, you'd typically pass a timeout to wait_condition and handle the case where no decision ever comes.

Resources

Watch Part 4 of this series: https://youtu.be/J0_GeI8Srzc
Full code for this example: https://github.com/bjoxiah/pydantic-ai-series/tree/agent-workflow

This is part of an ongoing series on my channel, Mastering Pydantic AI.

Building Modular AI Agent Features with Pydantic AI Capabilities

Blessed Josiah — Sun, 14 Jun 2026 08:30:00 +0000

If you're building AI Agents with Pydantic AI, understanding Capabilities is invaluable - it's the recommended way to add modular, reusable features to your agents.

This tutorial is part of my ongoing Pydantic AI series on YouTube, where I build a full no-code AI agent platform from scratch.

What is a Capability?

A capability in Pydantic AI is a modular unit of behavior that can be passed to an AI agent.

A capability can give your agent:

Custom toolsets
Instructions
Model settings
Lifecycle hooks

Think of it as a plug-and-play feature module - build it once, attach it to any agent.

How to Create a Capability

Capabilities are created using the Capability or AbstractCapability class:

from pydantic_ai.capabilities import AbstractCapability

Using AbstractCapability gives you full control over instructions, tools, and behavior. It's Pydantic AI's recommended pattern if you're building a library or platform on top of the framework.

Example: A Research Capability

In this tutorial, I build a Research Capability powered by the Tavily Search API, and an Email Capability powered by Resend.

Here's the research capability:

from pydantic_ai import FunctionToolset
from pydantic_ai.capabilities import AbstractCapability
from dataclasses import dataclass
from pydantic_ai.common_tools.tavily import tavily_search_tool
from settings import settings


@dataclass
class ResearchCapability(AbstractCapability):
    def get_instructions(self):
        return "You can use the Tavily search tool for research"

    def get_toolset(self):
        toolset = FunctionToolset()
        toolset.add_tool(tavily_search_tool(api_key=settings.tavily_api_key))
        return toolset

That's it - get_instructions() tells the agent what it can do, and get_toolset() gives it the tools to do it. Attach this to any agent, and it instantly gains web research abilities.

Going Further: RAG + GraphRAG

I also built a Company Knowledge Capability that combines:

pgvector for semantic search over your own data
Neo4j + Graphiti for knowledge graph retrieval (GraphRAG)

This lets an agent answer questions from your company's documents, website content, or any knowledge base with both vector search and relationship-aware graph queries.

Watch the Full Tutorial

This post covers the concept — the full video walks through building both capabilities live, plus the RAG/GraphRAG ingestion pipeline, observability with Logfire, and wiring it all into a working agent.

🎥 Watch on YouTube →
💻 Full source code →

While you're there, subscribe for more software and AI related content!

DEV Community: Blessed Josiah