Advanced Pydantic AI Agents: Building a Multi-Agent System in Pydantic AI

In the previous part of this series, we introduced the concept of message history and explored how agents in Pydantic AI can maintain conversational context across multiple interactions.

In this part, we temporarily set aside message history and focus instead on a new concept in Pydantic AI: the Multi-Agent Pattern. This pattern allows us to design modular, cooperative AI systems that distribute responsibilities and work between agents to complete a task.

Understanding the Multi-Agent Pattern in Pydantic AI

The Multi-Agent Pattern is a design principle in which several agents cooperate to complete complex tasks.

Each agent is responsible for a specific domain or capability, and together they form a distributed intelligence system.

Pydantic AI supports several forms of multi-agent coordination, each designed for different control flow strategies.

• Agent Delegation, where one agent calls another agent to perform a specific subtask.
• Programmatic Agent Hand-Off, where control is explicitly transferred from one agent to another during runtime.
• Graph-Based Control Flow, where agents are organized in a directed graph and the execution path is dynamically determined based on the context and results.

In this tutorial, we focus exclusively on the Agent Delegation pattern. It demonstrates how one agent can call another in a structured, type-safe way to delegate part of its work to another specialized agent.

Implementing Two Cooperative Agents

In our example, we divide the responsibilities between two agents.

1. TaskAgent handles user queries related to task creation.
2. ProjectManagementAgent manages project-level updates such as tracking open tasks.

The TaskAgent is responsible for creating tasks and delegates the update of project details to the ProjectManagementAgent through a tool call.

Let’s start by adapting the Task Agent.

task_agent.py

class TaskAgent():
    def __init__(self):
        self.agent = self._init_agent()
        self.pm_agent = ProjectManagementAgent()

    def _init_agent(self) -> Agent:
        agent = Agent(
            model = MistralModel(model_name=os.getenv("LLM_MISTRAL_MODEL"), provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))),
            output_type=TaskModel | Failed,
            deps_type=TaskDependency,
        )

        @agent.system_prompt
        def create_system_prompt(ctx: RunContext[TaskDependency]) -> str:
            return (f"""
                    You are a task creation agent.
                    Always follow the following instructions:
                    1. Update the project {ctx.deps.project} using the 'update_project' tool.
                    2. Create a brief task from the user's query. It is created by {ctx.deps.username} and belongs to the project {ctx.deps.project}.
                    """)

        @agent.tool_plain
        async def update_project(project_name: str) -> str:
            """Update the new created task's project."""
            result = await self.pm_agent.run(query=f"Update open tasks for {project_name}")

            if isinstance(result, Failed):
                return Failed(reason=f"Task creation failed: {result.reason}")

            return f"Updated project details for project {project_name}: {result.project}."

        return agent

pm_agent.py

class ProjectManagementAgent:
    def __init__(self):
        self.agent = self._init_agent()

    def _init_agent(self) -> Agent[ProjectDetails | Failed, None]:
        agent = Agent[ProjectDetails | Failed, None](
            model = MistralModel(model_name=os.getenv("LLM_MISTRAL_MODEL"), provider=MistralProvider(api_key=os.getenv("MISTRAL_API_KEY"))),
            output_type=ProjectDetails | Failed,
            system_prompt=("""
                           You are a project management agent with the task of updating the number of open tasks for a project.
                           """)
        )

        @agent.tool_plain
        def update_open_tasks(project_name: str) -> ProjectDetails | Failed:
            """Updates the number of open tasks for a given project."""
            project = PROJECTS.get(project_name, None)
            if project is None:
                return Failed(reason=f"Updating open tasks failed for project {project_name}. Project does not exist.")

            # create internal ProjectDetails python object and increase number of open tasks by one
            project_details = ProjectDetails(owned_by_group=project.get("group"), number_of_open_tasks=project.get("open_tasks"))
            project_details.number_of_open_tasks += 1

            # write the new project details back to the data
            # this reflects a database update instruction for a real-world application
            project["open_tasks"] = project_details.number_of_open_tasks

            return project_details

        return agent

With this structure, the TaskAgent no longer manages project data directly. Instead, it delegates this part of the process to the ProjectManagementAgent, which ensures a clear separation of functionality and improves maintainability.

This architectural separation also affects the data model. Each agent now has its own dedicated data class that represents its specific domain. The project-related data is handled by the ProjectManagementAgent using the new ProjectDetails class, while the TaskAgent works with the TaskModel, which now includes project-specific information through a nested field. Additionally, a Failed class was introduced to handle error states in a structured way.

model.py

class ProjectDetails(BaseModel):
    owned_by_group: str = Field(description="The group this project is owned by.")
    number_of_open_tasks: int = Field(description="The number of open tasks for the project.")

class TaskModel(BaseModel):
    task: str = Field(description="The title of the task.")
    description: str = Field(description="A brief description of the task specifing its steps.")
    priority: int = Field(description="The priority of the task on a scale from 1 (high) to 5 (low).")
    project: str = Field(description="The project the task belongs to.")
    created_by:str = Field(description="The name of the person who created the task.")
    project_details: ProjectDetails = Field(description="The details of the project this task belongs to.")

class Failed(BaseModel):
    reason: str = Field(description="The reason for the failure.")

The Failed class standardizes error handling across agents. Instead of returning unstructured strings or exceptions, an agent can return a Failed object with a descriptive reason. The calling agent can then interpret the failure, propagate it, or respond accordingly.

This design keeps success and failure paths explicit and type-safe, making debugging and testing much easier. In essence, the separation of agents is mirrored by a separation of data models, which allows a clear, maintainable, and scalable approach for building cooperative multi-agent systems.

Running the Multi-Agent System

Let’s now execute our Pydantic AI multi-agent system and observe how the two agents interact under both normal and failure conditions.

app.py

from dotenv import load_dotenv
from LearnPydanticAI.task_agent import TaskAgent
from LearnPydanticAI.dependecies import TaskDependency

import asyncio

async def main():
    load_dotenv()  # take environment variables from .env.

    agent = TaskAgent()

    # successful run
    deps = TaskDependency(username="sudo", project="LearnPydanticAI")
    answer = await agent.run("I want to learn more about Pydantic AI.", deps=deps)
    print(answer)

    # failed run due to a non-existing project
    deps_fail = TaskDependency(username="sudo", project="NonExistingProject")
    failed_answer = await agent.run("I want to learn more about Pydantic AI.", deps=deps_fail)
    print(failed_answer)

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

In the first case, the agent successfully creates a task and updates the project information.

In the second case, we simulate an error scenario by providing a project name that does not exist. The ProjectManagementAgent detects that the project cannot be found and returns a Failedobject, which is then handled by the TaskAgent.

% poetry run python src/LearnPydanticAI/app.py
task='Learn about Pydantic AI' 
description='1. Research about Pydantic AI.\n2. Learn about its features and capabilities.\n3. Explore its documentation and tutorials.' 
priority=3 
project='LearnPydanticAI' 
created_by='sudo' 
project_details=ProjectDetails(owned_by_group='dev_team', number_of_open_tasks=3)

reason='Task creation failed: Updating open tasks failed for project NonExistingProject. Project does not exist.'

This shows how Pydantic AI agents can coordinate their behavior while maintaining robust error handling across agent boundaries.

What Happens Behind the Scenes

When the query "I want to learn more about Pydantic AI." is executed, the following sequence occurs:

1. The TaskAgent receives the query and prepares to create a new task.
2. Before creating the task, it invokes the update_project tool.
3. This tool delegates the call to the ProjectManagementAgent, which updates the number of open tasks for the specified project.
4. Once the update is successful, the result is returned to the TaskAgent.
5. The TaskAgent then completes the task creation process and outputs a structured TaskModel.

When a non-existing project name is passed to the agent, a different control flow is triggered.

1. The ProjectManagementAgent attempts to update the specified project but cannot find a corresponding entry in the PROJECTS dictionary.
2. It returns a Failed object with a descriptive error message that explains why the update could not be completed.
3. The TaskAgent receives this Failed response instead of a project update result.
4. As a result, the TaskAgent stops the task creation process and propagates the failure, ensuring that no inconsistent data is produced.

This control flow illustrates how multiple agents in Pydantic AI can coordinate their tasks, exchange structured information, and gracefully recover from errors. It also demonstrates the value of well-defined data models and typed outputs, which make error handling explicit and maintain the transparency of the system’s internal logic.

Relevance for Real-World Applications

The The Multi-Agent Pattern in Pydantic AI is a powerful concept for designing scalable and modular AI systems. It allows developers to distribute complex logic across specialized agents that can interact and support one another.

In production environments, this approach enables teams to:

• Divide large workflows into modular components.
• Maintain clear boundaries between domains and functionalities.
• Extend or replace individual agents without changing the overall architecture.
• Achieve parallel and distributed reasoning across multiple AI modules.
• Integrate domain-specific intelligence through dedicated agents.

Using Agent Delegation in Pydantic AI combines the flexibility of natural language interfaces with the robustness of structured and typed agent communication. It provides a scalable foundation for building collaborative AI systems in Python that are easy to maintain, extend, and adapt to complex real-world scenarios.

Conclusion to this Series

With this fifth part, the current tutorial series on Pydantic AI Agents comes to a close. We’ve gone from building a simple, structured agent to exploring advanced features such as dependencies, function tools, message history, and finally, the multi-agent pattern with agent delegation.

Together, these concepts form a solid foundation for designing intelligent, modular, and production-ready AI systems in Python. While this marks the end of the series, it’s only the beginning of what can be done with Pydantic AI. Continue exploring advanced concepts and mechanisms, ranging from graph-based agent orchestration to persistent memory architectures and Pydantic Evals.

Thank you for following along this journey into the world of structured AI agent design with Pydantic AI. If you’ve enjoyed this series, stay tuned for more insights on AI programming in Python and modern AI frameworks. I’ll continue sharing practical tutorials and engineering perspectives on this blog to help bridge the gap between AI research and real-world software development.

💻 Code on GitHub: hamluk/LearnPydanticAI/part-5

📘 Read Part 4: Extending Pydantic AI Agents with Chat History