A Comprehensive Guide on Agent Development with Pydantic AI: Beginner to Advance

Photo by Gerard Siderius on Unsplash

I ’ve always been on the lookout for a solid framework to build AI agents — especially as a frequent hackathon participant. Stability is crucial for me since I can’t afford my setup to break right before a demo. In the past, I built my own agent framework from scratch without relying on LangChain or other libraries.

But then I discovered Pydantic AI — a powerful, structured, and reliable framework for building AI agents. It provides a seamless way to manage agent behaviors, enforce data validation, and scale multi-agent systems without the typical headaches.

What’s in This Guide?

This guide is structured into six key chapters, taking you from the basics to advanced multi-agent workflows:

• Chapter 1: Introduction to Pydantic AI
• Chapter 2: Building Basic Agents with Pydantic AI
• Chapter 3: Structured Responses and Data Validation with Pydantic AI
• Chapter 4: Extending Agents with Tools in Pydantic AI
• Chapter 5: Advancing Multi-Agent Systems with Pydantic AI
• Chapter 6: Multi-Agent Workflows with Pydantic Graph and Pydantic AI

Chapter 1: Introduction to Pydantic AI

I’m excited to share this journey with you, whether you’re a beginner exploring agent-based systems or an advanced developer looking to optimize multi-agent workflows. If you find this guide helpful and would like to support my work, feel free to check out my Ko-fi page:

☕ Buy me a coffee on Ko-fi

Chapter 1: Introduction to Pydantic AI

1. Introduction to Pydantic AI

Pydantic AI is a Python framework designed to simplify the development of production-grade applications utilizing Generative AI. Built by the team behind Pydantic, it offers a model-agnostic approach, supporting various AI models such as OpenAI, Anthropic, Gemini, Deepseek, Ollama, Groq, Cohere, and Mistral.

Pydantic AI emphasizes:

• Type safety
• Structured responses
• Seamless integration with tools like Pydantic Logfire for real-time debugging and performance monitoring

To begin using Pydantic AI, ensure you have Python 3.9 or higher installed.

pip install pydantic-ai

In case of any lack of library, these are the necessary library in the `requirements.txt` that I am using to teach this tutorial in Pydantic AI.

annotated-types==0.7.0
anthropic==0.48.0
anyio==4.8.0
asttokens==3.0.0
cachetools==5.5.2
certifi==2025.1.31
charset-normalizer==3.4.1
click==8.1.8
cohere==5.13.12
colorama==0.4.6
comm==0.2.2
debugpy==1.8.12
decorator==5.2.1
distro==1.9.0
duckduckgo_search==7.5.0
eval_type_backport==0.2.2
executing==2.2.0
fastavro==1.10.0
filelock==3.17.0
fsspec==2025.2.0
google-auth==2.38.0
griffe==1.5.7
groq==0.18.0
h11==0.14.0
httpcore==1.0.7
httpx==0.28.1
httpx-sse==0.4.0
huggingface-hub==0.29.1
idna==3.10
ipykernel==6.29.5
ipython==8.32.0
jedi==0.19.2
jiter==0.8.2
joblib==1.4.2
jsonpath-python==1.0.6
jupyter_client==8.6.3
jupyter_core==5.7.2
llm-sandbox==0.2.3
logfire-api==3.6.4
lxml==5.3.1
markdown-it-py==3.0.0
matplotlib-inline==0.1.7
mdurl==0.1.2
mistralai==1.5.0
mypy-extensions==1.0.0
nest-asyncio==1.6.0
nltk==3.9.1
openai==1.65.1
packaging==24.2
parso==0.8.4
pexpect==4.9.0
platformdirs==4.3.6
primp==0.14.0
prompt_toolkit==3.0.50
psutil==7.0.0
ptyprocess==0.7.0
pure_eval==0.2.3
pyasn1==0.6.1
pyasn1_modules==0.4.1
pydantic==2.10.6
pydantic-ai==0.0.29
pydantic-ai-slim==0.0.29
pydantic-graph==0.0.29
pydantic_core==2.27.2
Pygments==2.19.1
python-dateutil==2.9.0.post0
python-dotenv==1.0.1
PyYAML==6.0.2
pyzmq==26.2.1
regex==2024.11.6
requests==2.32.3
rich==13.9.4
rsa==4.9
six==1.17.0
sniffio==1.3.1
stack-data==0.6.3
tokenizers==0.21.0
tornado==6.4.2
tqdm==4.67.1
traitlets==5.14.3
types-requests==2.32.0.20241016
typing-inspect==0.9.0
typing_extensions==4.12.2
urllib3==2.3.0
wcwidth==0.2.13

If you are using Jupyter Notebook, you need to import nest_asyncio and apply it to enable the use of synchronous operations within Jupyter Notebook

You can add your API KEY to your `.env` file

GEMINI_API_KEY=AIzaSyB0-0000000000000000000000000000000
OPENAI_API_KEY=sk-proj-00000000000000000000000000000000
ANTHROPIC_API_KEY=sk-ant-00000000000000000000000000000000

If you are using Jupyter Notebook, you need to import nest_asyncio and apply it to enable the use of synchronous operations within Jupyter Notebook

Run this to import the library of Pydantic AI

### Import the Agent class and set the Google API key
from pydantic_ai import Agent
import os
import dotenv
dotenv.load_dotenv()

# Set your Google API key
os.environ["GOOGLE_API_KEY"] = os.getenv("GEMINI_API_KEY")

3. Creating a “Hello World” Agent

Let’s create simple agents that respond to basic queries using Gemini model.

# Initialize the agent with Gemini model
gemini_agent = Agent(
    'google-gla:gemini-2.0-flash',  # Using Gemini 1.5 Flash model
    system_prompt='You are a helpful assistant specialized in Python programming.',
)

# Run the agent with a user query
result = gemini_agent.run_sync('What are the key features of Pydantic AI in a short response.')
print("Gemini Response:")
print(result.data)

4. Async Usage

Pydantic AI also supports asynchronous operations, which is useful for web applications or when making multiple requests.

async def ask_gemini():
    # Initialize the agent with Gemini model
    agent = Agent(
        'google-gla:gemini-2.0-flash',
        system_prompt='You are a helpful assistant specialized in Python programming.',
    )
    
    # Run the agent asynchronously
    result = await agent.run('Explain how to use structured outputs in Pydantic AI in a short response')
    print("Gemini Response:")
    print(result.data)

# Run the async function
asyncio.run(ask_gemini())

5. Streaming Usage

# Storyteller Agent
story_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt="You are an AI storyteller. Generate engaging, real-time sci-fi adventures."
)

# Stream the story
async def stream_story():
    user_prompt = "Tell me a sci-fi story about a lost spaceship in a short response."
    async with story_agent.run_stream(user_prompt) as response:
        async for part in response.stream_text():
            print(part, end='', flush=True)

# Run the streaming story generator
asyncio.run(stream_story())

Conclusion

In this chapter, we’ve introduced Pydantic AI and demonstrated how to create simple agents using Gemini model.

In the upcoming chapters, we’ll delve deeper into building more complex agents, incorporating tools, handling structured responses, and exploring advanced features of Pydantic AI.

Chapter 2: Building Basic Agents with Pydantic AI

In this chapter, we’ll delve into constructing basic agents using Pydantic AI with the Gemini model. We’ll cover the essential components of an agent, differentiate between stateless and stateful agents, and explore techniques to enhance agent responses.

1. Understanding Agents Components

An agent in Pydantic AI comprises several key elements:

• System Prompt: Provides context and guidance to the model, shaping its responses.​
• User Input: The query or prompt from the end-user.​
• Response: The output generated by the agent based on the system prompt and user input.​

These components work in unison to facilitate meaningful interactions between the user and the AI model.

2. Creating Stateless vs. Stateful Agents

Stateless Agents

Stateless agents treat each interaction independently, without retaining any context from previous exchanges.

# Initialize a stateless agent with the Gemini model
stateless_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt='You are an assistant that provides general information.',
)

# User query
user_query = 'What is the capital of France?'

# Run the agent synchronously
response = stateless_agent.run_sync(user_query)
print(response.data)
# Output: 'The capital of France is Paris.'

In this example, the agent responds to the user’s query without any knowledge of prior interactions.

Stateful Agents

Stateful agents maintain context across multiple interactions, allowing for more coherent and context-aware conversations.

# Initialize a stateful agent with the Gemini model
stateful_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt='You are a conversational assistant.',
)

# Initial user query
initial_query = 'Tell me about the Eiffel Tower in a short response.'

# Run the agent synchronously
initial_response = stateful_agent.run_sync(initial_query)
print(initial_response.data)
# Output: 'The Eiffel Tower is a wrought-iron lattice tower in Paris, France.'

# Follow-up query
follow_up_query = 'How tall is it?'

# Run the agent with the follow-up query
follow_up_response = stateful_agent.run_sync(follow_up_query, message_history=initial_response.new_messages())
print(follow_up_response.data)
# Output: 'The Eiffel Tower is approximately 300 meters tall.'

Here, the agent retains context from the initial query, enabling it to understand and respond appropriately to the follow-up question.

3. Enhancing Agent Responses

To improve the quality and relevance of agent responses, consider the following techniques:

Crafting Effective System Prompts

A well-crafted system prompt can significantly enhance the agent’s performance.

• Clarity and Focus: Ensure the prompt is clear and concise, specifying the agent’s role and expected behavior.
• Contextual Information: Include relevant background information or guidelines that help the model understand the user’s request.
• Avoid Ambiguity: Use precise language and avoid vague terms that could lead to misinterpretation.
• Update Regularly: As the agent evolves, update the system prompt to reflect its current capabilities and objectives.

# System prompt emphasizing concise responses
concise_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt='You are a concise assistant. Provide brief and to-the-point answers.',
)

# User query
query = 'Explain the theory of relativity.'

# Run the agent
concise_response = concise_agent.run_sync(query)
print(concise_response.data)

By specifying the desired response style in the system prompt, the agent tailors its output accordingly.

Managing User Inputs

Preprocessing user inputs can improve the agent’s understanding and response quality.

• Normalization: Convert inputs to a consistent format (e.g., lowercase, standardized units).
• Validation: Check for valid inputs and provide appropriate error messages.

# Initialize the agent
input_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt='You are an assistant that answers questions.',
)

# Raw user input
raw_input = '   What is the speed of light?   '

# Preprocess input by stripping leading/trailing whitespace
processed_input = raw_input.strip()

# Run the agent
input_response = input_agent.run_sync(processed_input)
print(input_response.data)
# Output: 'The speed of light in a vacuum is approximately 299,792 kilometers per second.'

By cleaning the user input before passing it to the agent, we ensure that extraneous characters do not affect the agent’s performance.

Conclusion

In this chapter, we’ve explored the foundational aspects of building basic agents with Pydantic AI using the Gemini model. Understanding the components of an agent, distinguishing between stateless and stateful interactions, and employing techniques to enhance responses are crucial steps in developing effective AI agents.​

In the next chapter, we’ll explore how to enforce structured outputs and validate data, ensuring that the AI’s outputs adhere to a predefined format, enhancing reliability and predictability.

Chapter 3: Structured Responses and Data Validation with Pydantic AI

In this chapter, we’ll explore how to enforce structured outputs and validate data using Pydantic AI with the Gemini model. Structured responses ensure that the AI’s outputs adhere to a predefined format, enhancing reliability and predictability.

1. Introduction to Structured Responses

Structured responses involve defining a schema that the AI’s responses must follow. This approach is particularly useful when integrating AI outputs into applications that require consistent data formats. Pydantic AI leverages Pydantic models to define these schemas, enabling automatic validation and parsing of the AI’s responses.

from pydantic import BaseModel
from typing import List

# Define a Pydantic model for a single dictionary tip
class DictionaryTip(BaseModel):
    title: str
    description: str
    code_example: str

# Define a Pydantic model for multiple dictionary tips
class DictionaryTips(BaseModel):
    tips: List[DictionaryTip]

In this example, `DictionaryTip` defines the structure for a single tip, and `DictionaryTips` encapsulates a list of such tips.

2. Integrating Structured Responses with Pydantic AI

To enforce structured outputs, pass the Pydantic model as the `result_type` parameter when initializing the agent.

# Initialize the agent with the Gemini model and structured output
agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt='You are a Python expert providing tips on dictionary usage.',
    result_type=DictionaryTips  # Enforcing the structured output
)

# User query
query = 'Provide three tips for using Python dictionaries effectively.'

# Run the agent synchronously
response = agent.run_sync(query)

# Access the structured data
for tip in response.data.tips:
    print(f"Title: {tip.title}")
    print(f"Description: {tip.description}")
    print(f"Code Example:\n{tip.code_example}\n")

In this setup, the agent is instructed to format its response according to the `DictionaryTips` model. The `response.data` will be an instance of `DictionaryTips` if the AI’s output matches the expected structure.

3. Handling Validation Errors

There might be instances where the AI’s output doesn’t conform to the defined Pydantic model, leading to validation errors. It’s essential to handle these exceptions gracefully.

from pydantic import ValidationError

# Initialize the agent with the Gemini model and structured output
agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt='You have to try to output with a false response',
    result_type=DictionaryTips  # Enforcing the structured output
)

# User query
query = 'Provide a response in a list of dictionaries that breaks the structure by having a string instead of a dictionary'

try:
    # Run the agent synchronously
    response = agent.run_sync(query)
    # Access the structured data
    for tip in response.data.tips:
        print(f"Title: {tip.title}")
        print(f"Description: {tip.description}")
        print(f"Code Example:\n{tip.code_example}\n")
except ValidationError as e:
    print("Validation Error:", e)
    print("The AI's response did not match the expected structure.")

I can’t seems to break the structure with the above prompt. But it is super important to handle the ValidationError to ensure that your application remains robust even when the AI’s output is unexpected.

By wrapping the agent’s execution in a try-except block, you can catch ValidationError exceptions and handle them appropriately.

4. Benefits of Structured Responses

Implementing structured outputs offers several advantages:​

• Consistency: Ensures that the AI’s responses adhere to a predefined format, making it easier to parse and utilize the data.​
• Reliability: Reduces the chances of unexpected or malformed outputs, enhancing the robustness of your application.​
• Integration: Facilitates seamless integration of AI outputs into systems that require specific data structures.

Conclusion
In this chapter, we’ve explored how to enforce structured outputs and validate data using Pydantic AI with the Gemini model. By defining Pydantic models and integrating them into your agents, you can ensure that the AI’s responses are consistent, reliable, and align with your application’s requirements.​

In the next chapter, we’ll delve into extending agent capabilities by incorporating tools and custom functions, enabling agents to perform more complex tasks and provide enriched responses.

Chapter 4: Extending Agents with Tools in Pydantic AI

Agents in Pydantic AI can perform a variety of tasks. However, sometimes, they need extra functionality, such as accessing live data, interacting with databases, or performing calculations. Tools allow agents to access external functions, making them more versatile and capable.

In this tutorial, we’ll walk through how to extend agents with tools, covering:

• Creating and registering tools
• Using built-in tools
• Building powerful custom tools
• Dependency injection for modular design

We’ll showcase these concepts with real-world examples instead of generic functions.

1. Introduction to Tools in Pydantic AI
Tools allow an agent to call functions whenever needed. They can be:

• Plain tools (`@agent.tool_plain`) → Standalone functions that return data
• Context-aware tools (`@agent.tool`) → Require additional input from the agent’s context

We’ll demonstrate these in practical applications.

2. Registering Tools in an Agent

Let’s create an AI-powered workout planner that suggests exercises based on user preferences and fitness levels.

# Initialize the agent
agent = Agent(
    'google-gla:gemini-2.0-flash',
    system_prompt="You are a fitness coach that suggests personalized workout plans.",
)

# Define a tool to suggest workouts
@agent.tool_plain
def suggest_workout(goal: str) -> str:
    """Suggests a workout based on the user's fitness goal."""
    workouts = {
        "strength": ["Deadlifts", "Squats", "Bench Press"],
        "cardio": ["Running", "Cycling", "Jump Rope"],
        "flexibility": ["Yoga", "Dynamic Stretching", "Pilates"],
    }
    
    # Check for partial matches in the goal
    for key in workouts:
        if key in goal.lower():
            return f"Try this workout for {key}: {random.choice(workouts[key])}"
    
    return f"Try this workout: {random.choice(['Rest Day', 'Walking', 'Light Stretching'])}"

# Run the agent
result = agent.run_sync('I want a workout for strength training.')
print(result.data)

3. Using Built-in Tools

Pydantic AI comes with common tools like duckduckgo_search_tool() for web searches.

Install the duckduckgo tool if your environment doesn’t have it.

pip install 'pydantic-ai-slim[duckduckgo]' - quiet

from pydantic_ai.common_tools.duckduckgo import duckduckgo_search_tool

# Initialize the agent with a built-in search tool
agent = Agent(
    'google-gla:gemini-1.5-flash',
    tools=[duckduckgo_search_tool()],
    system_prompt="Search DuckDuckGo for the given query and return the results.",
)

# Run the agent with a query
result = agent.run_sync('What is the current President of Malaysia?')
print(result.data)

I found out that only the system prompt: `Search DuckDuckGo for the given query and return the results.` works, else it will return an error.

It seems like system prompt is important…

4. Dependency Injection in Tools

Pydantic AI supports dependency injection, allowing you to manage dependencies effectively within your tools. This feature is particularly useful for injecting services or configurations that your tools require.

# Initialize the agent
agent = Agent(
    'google-gla:gemini-2.0-flash',
    system_prompt='You provide exchange rate information to users.',
)

# Define a custom tool with dependency injection
@agent.tool
def get_exchange_rate(ctx: RunContext[dict], currency: str) -> str:
    """Fetch the exchange rate for a given currency."""
    exchange_rates = ctx.deps['exchange_rates']
    rate = exchange_rates.get(currency.upper(), 'unknown')
    return f"The exchange rate for {currency.upper()} is {rate}."

# Dependency data
dependencies = {
    'exchange_rates': {
        'USD': '1.00',
        'EUR': '0.85',
        'JPY': '110.00',
    }
}

# Run the agent with a user query
result = agent.run_sync('What is the exchange rate for EUR?', deps=dependencies)
print(result.data)

5. Integrating Agents with External APIs

Example: Real-Time Crypto Price Tracking

import requests

# Initialize the Crypto Agent
crypto_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt='You provide real-time cryptocurrency prices and trends.',
)

# Define a tool to fetch Bitcoin price with CoinGecko API
@crypto_agent.tool
def get_bitcoin_price(ctx: RunContext) -> str:
    """Fetches the current price of Bitcoin and recent trend."""
    try:
        # Use CoinGecko API to get Bitcoin data for the last 7 days
        url = "https://api.coingecko.com/api/v3/coins/bitcoin/market_chart?vs_currency=usd&days=7&interval=daily"
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        # Extract prices
        prices = data['prices']
        
        # Get the most recent price
        current_price = prices[-1][1]
        
        # Calculate price change percentage over the period
        first_price = prices[0][1]
        price_change = ((current_price - first_price) / first_price) * 100
        
        # Format the response
        trend = "up" if price_change > 0 else "down"
        return f"The current price of Bitcoin is ${current_price:.2f} USD. " \
               f"Over the past week, the price has gone {trend} by {abs(price_change):.2f}%."
    
    except Exception as e:
        # Fallback to mock data when API is unavailable
        return f"Unable to fetch real-time Bitcoin price (Error: {type(e).__name__}). " \
               f"Using sample data: The current price of Bitcoin is $29,876.45 USD."

# Run the Crypto Agent
response = crypto_agent.run_sync('What is the current price of Bitcoin?')
print(response.data)

6. Creating Advanced Custom Tools

Example: Personal Finance AI

# Initialize the agent
agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt="You are a finance assistant that helps track expenses.",
)

# Define a tool with dependency injection
@agent.tool
def add_expense(ctx: RunContext[dict], category: str, amount: float) -> str:
    """Stores a user's expense in the system."""
    ctx.deps['expenses'].append({'category': category, 'amount': amount})
    return f"Added {amount} to {category} expenses."

# Initialize dependencies (storage for expenses)
dependencies = {'expenses': []}

# Run the agent
agent.run_sync('Add 50 to food expenses.', deps=dependencies)
agent.run_sync('Add 30 to transport expenses.', deps=dependencies)

# Print stored expenses
print(dependencies['expenses'])

7. Combining Tools for Real-World Applications

# Initialize the agent
agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt="You recommend restaurants based on user preferences.",
)

# Tool to suggest a type of cuisine
@agent.tool_plain
def suggest_cuisine() -> str:
    """Suggests a random cuisine to try."""
    cuisines = ["Italian", "Japanese", "Mexican", "Indian", "Thai"]
    return random.choice(cuisines)

# Tool to fetch restaurants (simulating API response)
@agent.tool_plain
def find_restaurant(cuisine: str) -> str:
    """Finds a restaurant serving the specified cuisine."""
    restaurants = {
        "Italian": ["Pasta Heaven", "Luigi's Pizza"],
        "Japanese": ["Sushi World", "Ramen House"],
        "Mexican": ["Taco Land", "Burrito King"],
        "Indian": ["Spice Bazaar", "Curry Express"],
        "Thai": ["Bangkok Bites", "Thai Delight"],
    }
    return f"Try {random.choice(restaurants.get(cuisine, ['No options available']))} for {cuisine} food!"

# Run the agent
cuisine = suggest_cuisine()  # Get a random cuisine
result = find_restaurant(cuisine)  # Find a restaurant
print(f"{cuisine} cuisine → {result}")

• Multiple tools work together (suggest_cuisine + find_restaurant).
• This approach makes the agent modular and reusable.
• The logic can easily scale (e.g., integrating Google Places API).

Conclusion

In this chapter, we explored how to extend Pydantic AI agents with tools.
We covered:

• Registering function tools (`@agent.tool` and `@agent.tool_plain`)
• Using built-in tools like `duckduckgo_search_tool()`
• Building custom tools for finance, fitness, crypto price tracking and food recommendations
• Dependency injection to store and manage agent state

With these skills, you can now create intelligent, real-world AI assistants

In the next chapter, we’ll explore multi-agent systems, where multiple agents work together to achieve a common goal.

Chapter 5: Advancing Multi-Agent Systems with Pydantic AI

In this chapter, we will explore the next level of AI agent development using Pydantic AI. Rather than just handling single queries, we’ll learn how to orchestrate multiple agents, build intelligent workflows and add memory for long-term interactions

1. Multi-Agent Collaboration

Imagine building an AI-powered startup team where different agents collaborate just like real employees:

• The CEO Agent delegates tasks.
• The Marketing Agent generates ad campaigns.
• The Finance Agent manages budgets.
• The Tech Lead Agent handles product development.

# CEO Agent - The boss who assigns tasks
ceo_agent = Agent(
    'google-gla:gemini-1.5-flash',
    # The system prompt is so important for this kind of multi-agent collaboration. You need to specify what are the tools available to the agents and what are the constraints.
    system_prompt="You are the CEO of a tech startup. Based on user input, decide whether the Marketing, Finance, or Tech Lead should handle the task. then call to the `assign_task` tool to delegate the task."
)

# Marketing Agent - Creates ad campaigns
marketing_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt="You are a marketing expert. Generate creative ad campaigns."
)

# Finance Agent - Manages budgets
finance_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt="You are a financial expert. Handle budgets and financial planning."
)

# Tech Lead Agent - Handles product development
tech_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt="You are the tech lead. Guide product development and fix technical issues."
)

# CEO assigns tasks
@ceo_agent.tool
async def assign_task(ctx: RunContext, task: str, assignee: str = None) -> str:
    """Delegates tasks to the appropriate agent and returns their response in a detailed format.
    
    Args:
        task: The task to be completed
        assignee: Optional - The team member to assign (Marketing, Finance, or Tech)
    """
    # Determine which agent to use based on assignee or task content
    if assignee and "market" in assignee.lower() or "marketing" in task.lower():
        r = await marketing_agent.run(task)
        print("Marketing Agent's response:", r.data)
        return r.data   
    elif assignee and "finance" in assignee.lower() or "budget" in task.lower() or "finance" in task.lower():
        r = await finance_agent.run(task)
        print("Finance Agent's response:", r.data)
        return r.data
    elif assignee and "tech" in assignee.lower() or "tech" in task.lower() or "product" in task.lower():
        r = await tech_agent.run(task)
        print("Tech Lead Agent's response:", r.data)
        return r.data
    else:
        return "I don't recognize this task or assignee."

# Test the AI Startup Team
# response = await ceo_agent.run("Create a marketing campaign for our new AI app.")
response = await ceo_agent.run("Create a financial plan for our new AI app.")
print("CEO's response:", response.data)

Remark: The `assign_task` tool is called to delegate the task to the appropriate agent.

System prompt is so important for this kind of multi-agent collaboration. You need to specify what are the tools available to the agents and what are the constraints.

2. Agent-Guided Verification with Handoff

Agent verifies user details, prompts for missing information, and enforces retry limits before either completing verification or handing off the process (e.g., terminating or escalating to a human).

from typing import Union
from pydantic import BaseModel, Field
from pydantic_ai import Agent
from pydantic_ai.messages import ModelMessage
from pydantic_ai.usage import Usage, UsageLimits


class UserDetails(BaseModel):
    name: str
    age: int = Field(ge=1, le=120)
    matric_number: str


class Failed(BaseModel):
    """User failed to provide sufficient details after multiple attempts."""


# Define an agent using Gemini for verification
user_verification_agent = Agent[None, Union[UserDetails, Failed]](
    "google-gla:gemini-1.5-flash",
    result_type=Union[UserDetails, Failed],  # type: ignore
    system_prompt=(
        "Extract the user's name, age, and matric number for verification. "
        "If any information is missing or incomplete, request clarification up to three times."
    ),
)

usage_limits = UsageLimits(request_limit=3)  # Limit AI attempts to 3


async def verify_user(usage: Usage) -> Union[UserDetails, None]:
    message_history: Union[list[ModelMessage], None] = None

    for i in range(3):
        answer = input("Please provide your name, age, and matric number for verification:")
        print("User's input attempt", i+1, ":", answer)

        result = await user_verification_agent.run(
            answer,
            message_history=message_history,
            usage=usage,
            usage_limits=usage_limits,
        )

        if isinstance(result.data, UserDetails):
            return result.data
        else:
            print("Incomplete details. Please try again.")
            message_history = result.all_messages(
                result_tool_return_content="Ensure you provide your full name, age, and matric number."
            )

    print("Verification failed after multiple attempts. Process terminated.")
    return None


async def main():
    usage: Usage = Usage()
    user_details = await verify_user(usage)

    if user_details is not None:
        print(f"User verified: {user_details.name}, Age: {user_details.age}, Matric No: {user_details.matric_number}")


# Run the main function if this file is executed
if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Conclusion

Through these examples, we saw how multi-agent orchestration enhances AI workflows. Whether it’s automating business operations or enforcing structured verification, Pydantic AI allows for efficient, intelligent, and scalable AI-driven systems.

This knowledge sets the foundation for building more complex agent ecosystems — from AI-powered teams to self-sufficient decision-making systems.

Chapter 6: Multi-Agent Workflows with Pydantic Graph and Pydantic AI

In real-world AI applications, a single agent is often not enough to handle complex tasks efficiently. Instead, multiple agents can collaborate in a multi-agent workflow, where each agent is responsible for a specific function.

Pydantic Graph provides a structured approach to defining and managing multi-agent workflows, ensuring data flow between agents in a well-defined dependency graph. This chapter will cover:

• Understanding Pydantic Graph and its role in multi-agent AI workflows.
• Building a multi-agent system using Pydantic AI and Pydantic Graph.
• Executing and visualizing the workflow with Mermaid code generation.

1. What is Pydantic Graph?

Pydantic Graph is a graph-based execution framework that allows you to define workflows in a structured way using nodes and edges.

Instead of running tasks sequentially (like in normal Python scripts), you define a graph of tasks (nodes), where each node represents a step in your workflow, and edges define the flow of execution.

This structure is powerful for AI workflows, where tasks like data collection, processing, AI generation, and review need to be modular and interconnected.

2. Basics of Pydantic Graph

A Pydantic Graph consists of:

• Nodes (Tasks) → Individual steps in your workflow.
• State → A global object that holds data and evolves as nodes process it.
• Edges (Transitions) → Define which node executes next.
• Graph → The full execution pipeline connecting all nodes.

Let’s visualize a basic example where we fetch user input, process it, and output a result.

graph TD
    A[Get User Input] --> B[Process Data]
    B --> C[Generate Output]
    C --> D[End]

Each box (A, B, C) represents a node (step in the workflow). The arrows show the flow of execution.

3. Defining the Core Data Structure: State

The State class is a shared object that carries data between nodes.

from dataclasses import dataclass, field

@dataclass
class State:
    user_name: str = ""
    user_interests: list[str] = field(default_factory=list)
    recommended_articles: list[str] = field(default_factory=list)
    email_content: str = ""

• user_name → Stores the user’s name.
• user_interests → Stores topics the user is interested in.
• recommended_articles → Holds articles generated by AI.
• email_content → The final newsletter email.

This State object will be updated as nodes execute.

4. Defining the Nodes (A Step in the Workflow)

A node is a class that extends `BaseNode` and has a `run()` method, which performs a task and returns the next node to execute.

Let’s define a simple node that asks for user input.

from pydantic_graph import BaseNode, GraphRunContext, End

@dataclass
class GetUserInfo(BaseNode[State]):
    async def run(self, ctx: GraphRunContext[State]) -> "ProcessUserInfo":
        name = input("Enter your name: ")
        interests = input("Enter your interests (comma-separated): ").split(",")
        
        ctx.state.user_name = name
        ctx.state.user_interests = interests
        
        return ProcessUserInfo()  # Moves to the next step

Explanation:

• The node gets user input (name & interests).
• It updates the state object with the values.
• It returns the next node (`ProcessUserInfo`), defining the execution flow.

5. Connecting Nodes to Form a Workflow

Now, let’s create another node that processes user interests and fetches recommended articles.

@dataclass
class ProcessUserInfo(BaseNode[State]):
    async def run(self, ctx: GraphRunContext[State]) -> "GenerateEmail":
        interests = ctx.state.user_interests
        
        # Simulate AI fetching articles
        ctx.state.recommended_articles = [f"Latest news on {topic}" for topic in interests]
        
        return GenerateEmail()

Explanation:

• It processes user interests.
• It generates a list of articles based on those interests.
• It returns the next node (`GenerateEmail`).

6. Final Node: Generating an Email

from pydantic import BaseModel

class Article(BaseModel):
    title: str
    subtitle: str
    content: str

article_agent = Agent(
    'google-gla:gemini-1.5-flash',
    system_prompt='You are a newsletter writer. You have to write a newsletter email for a user about the topics they are interested in.',
    result_type=Article  # Enforcing the structured output
)

@dataclass
class GenerateEmail(BaseNode[State]):
    async def run(self, ctx: GraphRunContext[State]) -> "End":
        query = f"Generate a newsletter email content in markdown format for the user name called {ctx.state.user_name} about the following topics: {', '.join(ctx.state.user_interests)}"
        ctx.state.email_content = await article_agent.run(query)
        
        return End(ctx.state.email_content)

This is the final step, where:

• The newsletter email is generated.
• The workflow ends.

7. Creating the Pydantic Graph

Now that we have all our nodes, let’s define our Pydantic Graph.

from pydantic_graph import Graph

newsletter_graph = Graph(
    nodes=[GetUserInfo, ProcessUserInfo, GenerateEmail]
)

8. Executing the Workflow

We now execute the graph-based workflow.

import asyncio
from pydantic_graph import GraphRunContext

async def main():
    state = State()
    ctx = GraphRunContext(state=state, deps={})
    result = await newsletter_graph.run(GetUserInfo(), state=state)

    print("Final Email Output:")
    print("Title: ", result.output.data.title)
    print("Subtitle: ", result.output.data.subtitle)
    print("Content: ", result.output.data.content)
asyncio.run(main())

This will:

• Start the workflow from GetUserInfo.
• Move through nodes automatically.
• Generate a personalized email.

9. Visualizing the Workflow with Mermaid

You can generate a Mermaid graph with:

newsletter_graph.mermaid_code(start_node=GetUserInfo)

Workflow Diagram

graph TD
    A[Get User Input] --> B[Process Data]
    B --> C[Generate Output]
    C --> D[End]

Conclusion

In this chapter, we explored the powerful combination of Pydantic Graph and Pydantic AI for building multi-agent workflows. By structuring our application as a directed graph of specialized nodes, we created a modular, maintainable system where:

• Each node represents a discrete step in our workflow
• Data flows through the system in a well-defined manner via the State object
• Dependencies between steps are explicitly modeled as graph edges

This approach offers several advantages over traditional sequential programming:

• Modularity: Each component has a single responsibility, making the system easier to understand and maintain
• Flexibility: Nodes can be rearranged or replaced without disrupting the entire workflow
• Visualization: The graph structure can be visualized with Mermaid, providing clear documentation
• Scalability: Complex workflows can be built by composing simpler components

The newsletter generation example demonstrated how to combine user input, data processing, and AI-powered content generation in a cohesive workflow. This pattern can be extended to more complex scenarios involving multiple agents, each with specialized capabilities.

As AI systems grow more sophisticated, the ability to orchestrate multiple agents in structured workflows will become increasingly important. Pydantic Graph provides a solid foundation for building these next-generation AI applications.

View the Full Code Here:

GitHub - szeyu/Agent-Dev-PydanticAI: A structured syllabus for an Agent Development Course using…

Conclusion

Building AI agents can be a challenging yet rewarding journey. Throughout this guide, we explored how Pydantic AI provides a structured, reliable, and scalable framework for developing agents, from basic implementations to complex multi-agent workflows. Whether you’re just starting out or refining your skills, I hope this guide has given you valuable insights and practical knowledge to build better AI-powered applications.

As I continue my own coding journey, I’ll be sharing more insights, tutorials, and personal experiences. If you found this guide helpful, I’d truly appreciate your support!

☕ Buy me a coffee on Ko-fi

Stay Connected!

🔔 Follow me on Medium for more updates on my coding journey and in-depth technical blogs.
💻 Check out my projects on GitHub: github.com/szeyu
🔗 Connect with me on LinkedIn: linkedin.com/in/szeyusim

Thanks for reading, and happy coding!