Single agent¶
One agent handles the whole task: the simplest pattern, and the one to start from.
A single agent takes a request and answers it, using a prompt and a typed output you define. There is no coordination, no routing and no external calls unless you add them, which makes it the easiest pattern to reason about, test and debug. This example is a complete, working version: a structured output type, instructions in a prompt file, usage limits, and commented recipes for what you will usually add next (tools, dynamic instructions, multi-turn history).
Use it when
- One agent can do the whole job with one prompt.
- Nothing needs to be specialised, delegated or run in parallel.
- You want the lowest complexity and the easiest thing to test.
- You are not sure which pattern you need: start here and add one when a real run shows what is missing.
Look elsewhere when
- The agent needs facts it can't know, or must act in other systems:
tool_calling. - The task is really several tasks with different prompts:
pipeline,routerorsupervisor. - It must remember earlier turns:
conversation.
What it shows
- A structured output type (
AgentOutput) and an injectable deps dataclass (AgentDeps) - Instructions loaded from a prompt file (
prompts/single.txt) USAGE_LIMITSas a guardrail on every run, andRaiseContentFilterError- Commented recipes for tools, dynamic instructions and multi-turn history
test_example.py: how to unit-test an agent withTestModel
See it run: sample_run.md is a recorded run against a real model: what each agent was asked, which tools it called, and what it returned.
Source¶
All of it is in examples/single/.
"""Single-agent pattern: one agent handles the entire task.
This is the simplest pattern — use it when:
- One agent can handle the full task
- No specialization or delegation is needed
- You want the lowest complexity
To use:
1. Define your output type (or use str for unstructured output)
2. Set your instructions
3. Add tools if needed
4. Call run_agent() and read `.output` from the RunResult it returns
"""
from __future__ import annotations
from dataclasses import dataclass
from pydantic import BaseModel
from pydantic_ai import ( # noqa: F401 — RunContext used in commented tool example below
Agent,
RunContext,
)
from pydantic_ai.capabilities import RaiseContentFilterError
from pydantic_ai.usage import UsageLimits
from agent.config import settings
from agent.logging import agent_label, configure_logging, get_logger
from agent.prompts.templates import load_prompt
from agent.runs import Flow, RunResult
logger = get_logger(__name__)
LABEL = agent_label(__name__) # names this agent's run spans in Logfire traces
# Guardrail against runaway agentic loops. A run that exceeds any limit
# raises UsageLimitExceeded instead of silently burning tokens. Tune per task:
# request_limit caps model round-trips (each tool-call iteration is one
# request), total_tokens_limit caps overall tokens for the run. Set
# AGENT_COST_LIMIT (USD) to add a spend cap — optional, off by default, and only
# useful for models with known pricing (see Settings.cost_limit).
USAGE_LIMITS = UsageLimits(
request_limit=10, total_tokens_limit=100_000, cost_limit=settings.cost_limit
)
# --- Output type ---
# Replace with your actual output schema, or use str for unstructured output.
class AgentOutput(BaseModel):
"""Replace with your actual output schema."""
result: str
confidence: float
# --- Dependencies ---
# Use a dataclass to inject runtime dependencies (DB connections, API clients, etc.)
# Remove if your agent needs no external dependencies.
@dataclass
class AgentDeps:
"""Runtime dependencies injected into the agent."""
# example_client: SomeAPIClient # Add your dependencies here
pass
# --- Agent definition ---
agent: Agent[AgentDeps, AgentOutput] = Agent(
settings.model,
name=LABEL,
output_type=AgentOutput,
deps_type=AgentDeps,
# Fail fast when the provider filters a response, instead of retrying a
# refused request or returning partial text.
capabilities=[RaiseContentFilterError()],
instructions=load_prompt("single"), # prompts/single.txt; copied to agent/prompts/<name>.txt
# Or inline: instructions="You are a helpful assistant."
)
# --- Tools ---
# Add tools here. See agent/tools/example.py for the full pattern.
# @agent.tool
# async def my_tool(ctx: RunContext[AgentDeps], query: str) -> str:
# """Tool description — this docstring is sent to the LLM."""
# return "result"
# --- Dynamic instructions (optional) ---
# Use @agent.instructions for instructions that depend on runtime state.
# @agent.instructions
# async def dynamic_instructions(ctx: RunContext[AgentDeps]) -> str:
# return f"Today is {date.today()}."
async def run_agent(user_input: str, deps: AgentDeps | None = None) -> RunResult[AgentOutput]:
"""Run the agent with the given user input.
Args:
user_input: The user's message or task description.
deps: Runtime dependencies. Created with defaults if not provided.
Returns:
A RunResult: `.output` is the validated AgentOutput, `.usage` the total usage, and
`.steps[0].result` the native Pydantic AI result (messages, run id, ...).
"""
if deps is None:
deps = AgentDeps()
logger.info("Running single agent", extra={"user_input": user_input})
flow = Flow(USAGE_LIMITS)
result = await flow.run(agent, user_input, deps=deps)
logger.info("Agent run complete", extra={"output": result.output})
return flow.finish(result.output)
# --- Multi-turn conversation example ---
# To maintain conversation history across multiple calls:
#
# history = []
# result1 = await agent.run("First message", deps=deps)
# history = result1.all_messages()
#
# result2 = await agent.run("Follow-up", deps=deps, message_history=history)
# history = result2.all_messages()
#
# Use all_messages(), not new_messages(), when carrying history forward —
# new_messages() returns only the messages from that single run, so assigning
# it to history would silently drop all earlier turns.
if __name__ == "__main__":
import asyncio
configure_logging()
result = asyncio.run(run_agent("Hello, what can you do?"))
print(result.output)
You are a helpful, accurate, and concise AI assistant.
When using tools:
- Use tools when they provide information you don't have
- Read error messages carefully — they tell you how to recover
- If a tool fails, try to recover before giving up
When responding:
- Be direct and specific
- Acknowledge uncertainty when you're not sure
- Ask for clarification if the request is ambiguous
"""Example unit tests using TestModel — no API calls, no cost.
This file stays with the `single` example (add_agent.py does not copy it); it is a
recipe for testing your own agent.
TestModel simulates agent behavior for fast, deterministic unit tests.
Import it from: from pydantic_ai.models.test import TestModel
Use TestModel(call_tools=[]) for agent-level tests: a default TestModel() calls every
tool with junk arguments, which breaks tools that validate input. See
tests/test_safety_net.py for opting in to a tool call.
"""
from pydantic_ai.models.test import TestModel
from agent.runs import RunResult
from examples.single.agent import AgentDeps, agent, run_agent
async def test_agent_runs_with_test_model():
"""Agent runs without error using TestModel (no API call)."""
with agent.override(model=TestModel(call_tools=[])):
result = await agent.run("Test input", deps=AgentDeps())
# TestModel returns a placeholder output that satisfies the output_type schema
assert result.output is not None
async def test_agent_accepts_string_input():
"""Agent accepts a string user prompt."""
with agent.override(model=TestModel(call_tools=[])):
result = await agent.run("Hello", deps=AgentDeps())
assert result is not None
async def test_agent_message_history():
"""Demonstrate multi-turn conversation history pattern."""
with agent.override(model=TestModel(call_tools=[])):
result1 = await agent.run("First message", deps=AgentDeps())
history = result1.all_messages() # all_messages(), not new_messages() — keeps all turns
result2 = await agent.run(
"Follow-up message",
deps=AgentDeps(),
message_history=history,
)
assert result2.output is not None
# History from both turns is available
assert len(result2.all_messages()) > len(result1.all_messages())
async def test_run_agent_returns_a_run_result():
"""run_agent wraps the native result: `.output`, total `.usage`, and the single step."""
with agent.override(model=TestModel(call_tools=[])):
result = await run_agent("Test input")
assert isinstance(result, RunResult)
assert result.output is not None
assert [step.agent for step in result.steps] == ["single"]
assert result.steps[0].result.output == result.output # the native result, untouched
assert result.usage.requests == 1
"""Live check: the single agent runs against the real model. Run with `pytest -m eval`."""
import pytest
from evals.trace import traced_run
from examples.live_support import assert_every_agent_ran, run_as_script
from examples.single import agent as module
pytestmark = pytest.mark.eval
async def test_the_agent_answers_with_a_valid_confidence():
traced = await traced_run(module.run_agent, "Explain what an AI agent is in one sentence.")
output = traced.result.output
assert isinstance(output, module.AgentOutput)
assert "agent" in output.result.lower()
assert 0.0 <= output.confidence <= 1.0
assert [step.agent for step in traced.result.steps] == ["single"]
assert traced.result.usage.requests >= 1 # more if the model retried its output
assert_every_agent_ran(module, traced.agents_ran)
async def test_the_demo_script_runs():
assert "result=" in await run_as_script("examples.single.agent")
Recorded run · gemini-3.1-flash-lite · 1 step · $0.0003
Recorded 2026-10-07 with google:gemini-3.1-flash-lite · 1 step · 339 tokens · $0.0003 · 1.3 s.
Model output varies between runs. Regenerate with uv run python scripts/record_example.py single.
Input
Hello, what can you do?
Steps
1. single
339 tokens · $0.0003
Prompt
Hello, what can you do?
Output
{
"result": "I am a helpful, accurate, and concise AI assistant. I can help you with a wide range of tasks, including:\n\n* **Answering questions:** Providing information on a vast range of topics.\n* **Writing and editing:** Creating emails, essays, stories, reports, or refining your existing text.\n* **Summarization:** Condensing long articles, documents, or conversations into key points.\n* **Coding and technical help:** Writing, debugging, and explaining code in various programming languages.\n* **Brainstorming:** Generating ideas for projects, marketing, or creative tasks.\n* **Translation:** Translating text between many different languages.\n* **Data Analysis and Formatting:** Helping organize information or performing calculations.\n\nIs there something specific you would like help with today?",
"confidence": 1.0
}
Result
run_agent(...).output
{
"result": "I am a helpful, accurate, and concise AI assistant. I can help you with a wide range of tasks, including:\n\n* **Answering questions:** Providing information on a vast range of topics.\n* **Writing and editing:** Creating emails, essays, stories, reports, or refining your existing text.\n* **Summarization:** Condensing long articles, documents, or conversations into key points.\n* **Coding and technical help:** Writing, debugging, and explaining code in various programming languages.\n* **Brainstorming:** Generating ideas for projects, marketing, or creative tasks.\n* **Translation:** Translating text between many different languages.\n* **Data Analysis and Formatting:** Helping organize information or performing calculations.\n\nIs there something specific you would like help with today?",
"confidence": 1.0
}