* feat: add input and output guardrails * chore: use single backticks * fix: run InputGuard only on the first model request * fix: replace list[Any] with Sequence[ModelMessage] * fix: pass raw output to OutputGuard, not str(result.output) * refactor: organize tests into TestCapabilityName classes * fix: drain cancelled tasks in InputGuard parallel finally * fix: re-raise task exceptions via await instead of .exception() * refactor: consolidate parallel cancel/drain into single finally * feat: support callable block_message; collapse InputGuard hooks block_message now accepts a callable so the refusal text can reflect the prompt/output that tripped the guard, rather than being frozen at construction time. InputGuard's sequential path moves from before_model_request into wrap_model_request, so a single hook covers both sequential and parallel modes instead of two hooks each branching on `parallel`. Tests move to tests/guardrails/ to match the tests/<capability>/ layout. * refactor: guards return bool | GuardResult; drop block_message A guard now returns either a bare bool or a GuardResult carrying a refusal message, replacing the separate block_message constructor field. The message is produced when the guard decides, so it can reflect the guard's own reasoning rather than a string frozen at construction time. Guards (and the GuardResult path) may optionally take a RunContext as a first parameter, detected from the signature like pydantic-ai's output validators, so deps- and history-aware guards are possible without closing over globals. Prompt/output-only guards are unchanged. * feat: guard outcomes — allow/block/replace/retry with OTel spans A guard now reports one of four outcomes via GuardResult classmethods (bool shorthand still works): allow, block, replace, retry. - replace lets a guard redact rather than refuse — InputGuard rewrites the prompt sent to the model, OutputGuard substitutes the output. - retry lets OutputGuard send a bad output back to the model; OutputGuard moves from after_run to after_output_process so it can raise ModelRetry and return a modified output. - replace and block emit spans on the run tracer so a redaction or refusal is visible in Logfire; redacted content is included only when RunContext.trace_include_content is set. InputGuard replace requires sequential mode and retry is rejected as a usage error, since neither is meaningful for input. * test: cover guardrail test helpers for the 100% gate The coverage gate measures test files too; the _prompt_text helper had unreached branches. Drop it and assert on message parts inline. * docs: scope streaming behavior; add streaming tests A pydantic-ai-correctness review surfaced two streaming points. Verified both empirically: - InputGuard(parallel=True) works under run_stream() — no deadlock. - OutputGuard GuardResult.retry() is unsupported under run_stream(): pydantic-ai does not retry output while streaming, so a retry verdict surfaces as UnexpectedModelBehavior. Document the retry limitation and that OutputGuard screens only the final output (partial chunks reach the caller first while streaming). Note that input redaction also rewrites persisted history and targets text prompts. Add streaming tests for both guards to lock the behavior in. * test: make the parallel-streaming test a real regression guard The test proving InputGuard(parallel=True) does not deadlock under run_stream() had no timeout — a reintroduced deadlock would hang CI instead of failing. Wrap it in asyncio.wait_for and document the reviewed concern it guards against. * feat: guardrail capability ordering + GuardResult hardening Address @adtyavrdhn's review on #249. - `InputGuard.get_ordering()` → `position='innermost'` so any message-morphing capability runs first and the guard sees the final prompt the model will receive. - `OutputGuard.get_ordering()` → `position='outermost', wrapped_by= [Instrumentation]` so the guard's block/redact spans are always captured by an enclosing `Instrumentation` span regardless of user list order. - `GuardResult` is `frozen=True, kw_only=True` with a `__post_init__` that rejects field combinations the four-outcome contract does not allow (e.g. `replace` without a replacement). `block` with no message stays valid — the default kicks in at the use site. - `_run_guard` and `after_output_process` dispatch via `match action:` with `assert_never` exhaustiveness guards. - `_trace_block` gates the refusal `message` attribute behind `ctx.trace_include_content`, matching `_trace_redaction` — the message can quote sensitive content from the guarded value. - New tests: ordering declarations, `__post_init__` validation, frozen enforcement, outer-cancellation no-leak regression guard (which confirms `asyncio.shield` around the cleanup `gather` is not needed — the outer cancel is already consumed by `asyncio.wait`). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * chore: bump pydantic-ai-slim to 2.4.0 in lock CI resolves pydantic packages to newest (uv.toml marks them exclude-newer=false); the merged lock was stale at 2.1.0 so uv sync --locked failed. Regenerate to 2.4.0. * test(code_mode): drop removed ToolSearchMatch.description field pydantic-ai 2.4.0 narrowed ToolSearchMatch to carry only `name`; the full ToolDefinition (with description) is surfaced separately. Update the test fixtures to the new shape. --------- Co-authored-by: Kacper Włodarczyk <kacperwlodarczyk@protonmail.com> Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Pydantic AI Harness
The batteries for your Pydantic AI agent.
Pydantic AI's capabilities and hooks API is how you give an agent its harness -- bundles of tools, lifecycle hooks, instructions, and model settings that extend what the agent can do without any framework changes.
Pydantic AI Harness is the official capability library for Pydantic AI, maintained by the Pydantic AI team. Pydantic AI core ships capabilities that require model or framework support, and capabilities fundamental to every agent -- web search, tool search, thinking. Everything else lives here: standalone building blocks you pick and choose to turn your agent into a coding agent, a research assistant, or anything else. This is also where new capabilities start -- as they stabilize and prove themselves broadly essential, they can graduate into core.
The capability matrix tracks where we are. Tell us what to prioritize.
Contents: Installation · Quick start · DynamicWorkflow · Capability matrix · An ecosystem agent · Help us prioritize · Build your own · Contributing · Version policy · Pydantic AI references · License
Installation
uv add pydantic-ai-harness
Extras for specific capabilities:
uv add "pydantic-ai-harness[codemode]" # CodeMode (adds the Monty sandbox)
uv add "pydantic-ai-harness[dynamic-workflow]" # DynamicWorkflow (adds the Monty sandbox)
uv add "pydantic-ai-harness[logfire]" # ManagedPrompt (Logfire-managed prompts)
uv add "pydantic-ai-harness[acp]" # ACP (serve an agent to editors over the Agent Client Protocol)
The code-mode extra is also supported as an alias.
Requires Python 3.10+ and pydantic-ai-slim>=2.1.0.
Quick start
uv add "pydantic-ai-slim[anthropic,mcp,duckduckgo,logfire]" "pydantic-ai-harness[code-mode]"
import logfire
from pydantic_ai import Agent
from pydantic_ai.capabilities import MCP, WebSearch
from pydantic_ai_harness import CodeMode
# See https://ai.pydantic.dev/logfire/ for setup details.
logfire.configure()
logfire.instrument_pydantic_ai()
agent = Agent(
'anthropic:claude-opus-4-7',
capabilities=[
# Wraps every tool into a single run_code tool, sandboxed by Monty
# (https://github.com/pydantic/monty -- pulled in by the [code-mode] extra).
# The model writes Python that calls multiple tools with loops, conditionals,
# asyncio.gather, and local filtering -- one model round-trip for N tool calls.
CodeMode(),
# Connect to any MCP server -- here, the open-source Hacker News server
# (https://github.com/cyanheads/hn-mcp-server). native=False forces the
# local MCP toolset so CodeMode can wrap the tools; without it,
# providers that natively support MCP server connectors execute the tools
# server-side and bypass the sandbox.
MCP('https://hn.caseyjhand.com/mcp', native=False),
# Provider-adaptive web search; native=False routes through the local
# DuckDuckGo fallback (the [duckduckgo] extra above) so CodeMode can batch
# web searches alongside the HN calls in a single run_code.
WebSearch(native=False),
],
)
result = agent.run_sync(
"Across the top, best, and 'show HN' Hacker News feeds, find the most-discussed "
"story with at least 100 points. Pull its comment thread, its submitter's profile, "
"and any web coverage. Summarize what you find in one paragraph."
)
print(result.output)
"""
The most-discussed HN story across top/best/show clearing 100 points is "Vibe coding
and agentic engineering are getting closer than I'd like" by Simon Willison (748 points,
853 comments, on the Best feed), submitted by long-time HNer e12e. The piece argues
that the two modes Willison once kept mentally separate -- throwaway "vibe coding" and
disciplined "agentic engineering" -- are blurring, since agents like Claude Code now
reliably handle non-trivial tasks like "build a JSON API endpoint that runs a SQL query"
with tests and docs on the first pass. The HN thread is unusually substantive, with
commenters debating whether LLMs created or merely *exposed* sloppy engineering
practices and warning of a "normalization of deviance" as engineers stop reviewing diffs.
"""
See this run as a public Logfire trace → Each run_code span fans out into the tool calls the model issued from inside the sandbox -- it's the easiest way to understand what code mode actually did.
Orchestrating sub-agents: DynamicWorkflow
CodeMode gives the model one script for its tools. DynamicWorkflow does the same for sub-agents. Without it, an orchestrator delegates one tool call at a time: call a sub-agent, wait, read the result into context, think, call the next one. Ten delegations cost ten model round-trips, and every intermediate result flows through the orchestrator's context whether it needed to see it or not.
With it, the model writes one Python script in which each sub-agent is an async function, and the whole tree runs in a single tool call:
from pydantic_ai import Agent
from pydantic_ai_harness.experimental.dynamic_workflow import DynamicWorkflow
reviewer = Agent('anthropic:claude-sonnet-4-6', name='reviewer', description='Reviews code for bugs.')
summarizer = Agent('anthropic:claude-sonnet-4-6', name='summarizer', description='Summarizes findings.')
orchestrator = Agent(
'anthropic:claude-opus-4-7',
capabilities=[DynamicWorkflow(agents=[reviewer, summarizer])],
)
The script the model writes looks like this -- fan out, chain, and only the last line's value returns to its context:
import asyncio
reports = await asyncio.gather(
reviewer(task="Review auth.py for bugs:\n<file contents>"),
reviewer(task="Review parser.py for bugs:\n<file contents>"),
)
await summarizer(task="Summarize these findings:\n" + "\n\n".join(reports))
It composes with the rest of the harness:
- Budgets:
max_agent_callsis an exact, host-enforced ceiling on sub-agent runs (it holds even under concurrent fan-out), and by default the whole tree's token spend lands on the parent run'susage. - On-demand:
defer_loading=Truekeeps the catalog out of the prompt until the model loads the capability, andreveal()adds a sub-agent mid-run without disturbing the prompt cache.
DynamicWorkflow ships under experimental while planned extensions (structured sub-agent inputs, durable workflows) settle the call contract; importing it emits a HarnessExperimentalWarning.
Capability matrix
We studied leading coding agents, agent frameworks, and Claw-style assistants to map every capability area that matters for production agents. Each one is tracked as an issue in this repo.
Vote on whatever is linked in the Status column -- PRs if we're actively building it, issues if it's planned -- to help us decide what to work on next.
| Category | Capability | Description | Status | Community alternatives |
|---|---|---|---|---|
| Tools & execution | Code mode | Sandboxed Python execution via Monty -- one run_code call replaces N tool calls |
✅ Docs | |
| Tool search | Progressive tool discovery for large tool sets | ✅ Pydantic AI | ||
| File system | Read, write, edit, search files with path traversal prevention | ✅ Docs | pydantic-ai-backend (vstorm‑co) | |
| Shell | Execute commands with allowlists, denylists, and timeouts | ✅ Docs | pydantic-ai-backend (vstorm‑co) | |
| Repo context injection | Auto-load CLAUDE.md/AGENTS.md and repo structure | ✅ Docs (experimental) | pydantic-deep (vstorm‑co) | |
| Docs lookup | On-demand read_pyai_docs tool for Pydantic AI docs |
✅ Docs (experimental) | ||
| Verification loop | Run tests after edits, auto-fix failures | 🚧 PR #169 | ||
| Editor integration | ACP | Serve an agent to editors (Zed, etc.) over the Agent Client Protocol -- streamed text, diff-rendered edits, tool approval | ✅ Docs (experimental) | |
| Prompt management | Managed prompt | Back an agent's instructions with a Logfire-managed prompt, editable without shipping code | ✅ Docs | |
| Context management | Sliding window | Trim conversation history to stay within token limits | ✅ Docs (experimental) | summarization-pydantic-ai (vstorm‑co) |
| Context compaction | LLM-powered summarization of older messages | ✅ Docs (experimental) | summarization-pydantic-ai (vstorm‑co) | |
| Limit warnings | Warn agent before hitting context/iteration limits | ✅ Docs (experimental) | summarization-pydantic-ai (vstorm‑co) | |
| Tool output management | Truncate, summarize, or spill large tool outputs | ✅ Docs (experimental) | ||
| System reminders | Inject periodic reminders to counteract instruction drift | 🚧 PR #181 | ||
| Memory & persistence | Memory | Persistent key-value memory across sessions | 🚧 PR #179 | pydantic-deep (vstorm‑co) |
| Session persistence | Save and restore full conversation state | ✅ Docs (experimental) | ||
| Checkpointing | Snapshot, resume (continue_run), and fork (fork_run) a run |
✅ Docs (experimental) | pydantic-deep (vstorm‑co) | |
| Media externalization | Offload large BinaryContent to content-addressed stores (building blocks) |
✅ Docs (experimental) | ||
| Agent orchestration | Sub-agents | Delegate subtasks to specialized child agents | ✅ Docs (experimental) | subagents-pydantic-ai (vstorm‑co) |
| Dynamic workflow | Orchestrate sub-agents from a model-written Python script -- fan-out, chaining, voting in one tool call | ✅ Docs (experimental) | ||
| Skills | Progressive tool loading -- search, activate, deactivate | 🚧 PR #183 | pydantic-ai-skills (DougTrajano), pydantic-deep (vstorm‑co) | |
| Planning | Break complex tasks into structured plans before execution | ✅ Docs (experimental) | ||
| Runtime authoring | Let an agent author, validate, and load real capabilities at runtime | ✅ Docs (experimental) | ||
| Task tracking | Track tasks, subtasks, and dependencies | 📝 #65 | pydantic-ai-todo (vstorm‑co) | |
| Teams | Multi-agent teams with shared state and message bus | 📝 #195 | pydantic-deep (vstorm‑co) | |
| Safety & guardrails | Input guardrails | Validate user input before the agent run starts | 🚧 PR #182 | pydantic-ai-shields (vstorm‑co) |
| Output guardrails | Validate model output after the run completes | 🚧 PR #182 | pydantic-ai-shields (vstorm‑co) | |
| Cost/token budgets | Enforce token and cost limits per run | 🚧 PR #182 | pydantic-ai-shields (vstorm‑co) | |
| Tool access control | Block tools or require approval before execution | 🚧 PR #182 | pydantic-ai-shields (vstorm‑co) | |
| Async guardrails | Run validation concurrently with model requests | 🚧 PR #182 | pydantic-ai-shields (vstorm‑co) | |
| Secret masking | Detect and redact secrets in agent I/O | 🚧 PR #172 | pydantic-ai-shields (vstorm‑co) | |
| Approval workflows | Require human approval for sensitive operations | 🚧 PR #173 | Pydantic AI (built‑in) | |
| Tool budget | Limit total tool calls or cost per run | 🚧 PR #168 | ||
| Reliability | Stuck loop detection | Detect and break out of repetitive agent loops | 🚧 PR #186 | |
| Tool error recovery | Retry failed tool calls with backoff and budget | 🚧 PR #171 | ||
| Tool orphan repair | Fix orphaned tool calls in conversation history | 🚧 PR #184 | ||
| Reasoning | Adaptive reasoning | Adjust thinking effort based on task complexity | 🚧 PR #174 | |
| Current time | Inject current date/time into system prompt | 🚧 PR #170 |
Packages by vstorm-co are endorsed by the Pydantic AI team. We're working with them to upstream some of their implementations into this repo.
An ecosystem agent
The Quick start above is deliberately small. Here's the other end of the spectrum -- an agent wired up with capabilities drawn from across the Pydantic AI ecosystem: this repo, core pydantic-ai, and the community packages we vouch for in the matrix above.
import logfire
from pydantic_ai import Agent
from pydantic_ai.capabilities import MCP, Thinking, ToolSearch, WebSearch
from pydantic_ai_harness import CodeMode
# Community packages, alphabetical:
from pydantic_ai_backends import ConsoleCapability
from pydantic_ai_shields import CostTracking, InputGuard, SecretRedaction, ToolGuard
from pydantic_ai_skills import SkillsCapability
from pydantic_ai_summarization import ContextManagerCapability
from pydantic_ai_todo import TodoCapability
from pydantic_deep import MemoryCapability, StuckLoopDetection
from subagents_pydantic_ai import SubAgentCapability, SubAgentConfig
# See https://ai.pydantic.dev/logfire/ for setup details.
logfire.configure()
logfire.instrument_pydantic_ai()
agent = Agent(
'anthropic:claude-opus-4-7',
capabilities=[
# --- Tool execution & discovery ---
# Wraps every tool into a single run_code, sandboxed by Monty.
CodeMode(),
# Progressive tool discovery for large tool sets; discovered tools fold into run_code.
ToolSearch(),
# --- Reasoning ---
# Provider-adaptive thinking; uses native extended thinking on supporting models.
Thinking(effort='xhigh'),
# --- Context management ---
# Sliding window + LLM compaction. By @vstorm-co:
# https://github.com/vstorm-co/summarization-pydantic-ai
# Pydantic AI also ships `AnthropicCompaction` and `OpenAICompaction` for
# provider-native compaction.
ContextManagerCapability(max_tokens=180_000),
# --- Tools ---
# Connect to any MCP server -- here, the open-source Hacker News server
# (https://github.com/cyanheads/hn-mcp-server).
MCP('https://hn.caseyjhand.com/mcp'),
# Provider-adaptive web search; falls back to a local DuckDuckGo implementation.
WebSearch(),
# Filesystem + shell. By @vstorm-co: https://github.com/vstorm-co/pydantic-ai-backend
ConsoleCapability(),
# --- Memory & persistence ---
# Persistent ./MEMORY.md per agent name. By @vstorm-co:
# https://github.com/vstorm-co/pydantic-deepagents
MemoryCapability(agent_name='harness-example'),
# --- Orchestration ---
# Agent skills (Anthropic's spec) by @DougTrajano:
# https://github.com/DougTrajano/pydantic-ai-skills
# @vstorm-co's pydantic-deep also offers skills loading; the two have different
# spec footprints (Doug's is closer to programmatic skills).
SkillsCapability(directories=['./skills']),
# Spawn sub-agents with their own toolsets and instructions. By @vstorm-co:
# https://github.com/vstorm-co/subagents-pydantic-ai
SubAgentCapability(subagents=[
SubAgentConfig(
name='researcher',
description='Deep research on a topic',
instructions='You are a thorough research assistant.',
),
]),
# Track tasks and subtasks; in-memory by default, AsyncPostgresStorage available.
# By @vstorm-co: https://github.com/vstorm-co/pydantic-ai-todo
TodoCapability(enable_subtasks=True),
# --- Safety & reliability ---
# The next four are by @vstorm-co: https://github.com/vstorm-co/pydantic-ai-shields
# Per-run cost cap with a callback hook.
CostTracking(budget_usd=5.0),
# Reject prompts that look like prompt-injection attempts.
InputGuard(guard=lambda p: 'ignore previous instructions' not in p.lower()),
# Block or require approval per tool name.
ToolGuard(blocked=['rm'], require_approval=['write_file']),
# Detect API keys/tokens in tool I/O and redact before they reach the model.
SecretRedaction(),
# Bail out if the agent gets stuck calling the same tools in a loop.
# By @vstorm-co: https://github.com/vstorm-co/pydantic-deepagents
StuckLoopDetection(),
],
)
This snippet is illustrative, not literally copy-pasteable: a few capabilities have setup requirements (a ./skills directory, a Postgres database for TodoCapability's persistent storage), and the community packages move independently of this one. The capability matrix tracks each one's status. As the harness ships first-party versions, the imports above will collapse onto fewer packages -- but the example will keep working, since the API surface is the same.
Help us prioritize
Vote on whatever is linked in the Status column above. If there's a PR, vote on the PR -- it means we're actively building it. If there's only an issue, vote on the issue.
Want something that's not on the list? Open a capability request.
Build your own
Capabilities are the primary extension point for Pydantic AI. Any of the existing capabilities in this repo can serve as a reference for building your own.
Publishing as a standalone package? Use the pydantic-ai-<name> naming convention. See Publishing capability packages.
Contributing
We welcome capability contributions. Here's how:
- Start with an issue. Open a capability request describing the behavior you want. This lets us discuss the approach and priority before code is written -- we can close an approach without closing the problem.
- Then open a PR. Once the issue exists, you're welcome to open a PR with an implementation. Link the issue in your PR. We review based on community interest -- upvotes on both the issue and PR count.
- Don't chase green CI. Get the approach working, then let us know. We'll take it from there -- we may push to your branch, rewrite, or open a follow-up PR. You'll be credited as the original author. (See the Pydantic AI contributing guide.)
Note
: PRs that modify
pyproject.tomloruv.lockfrom non-team members are auto-closed by CI to prevent supply chain risk. If you need a new dependency, open an issue.
Development
make install # install dependencies
make format # ruff format
make lint # ruff check
make typecheck # pyright strict
make test # pytest
make testcov # pytest with 100% branch coverage
Version policy
Pydantic AI Harness uses 0.x versioning to signal that APIs are still stabilizing. During 0.x:
- Minor releases (0.1 → 0.2) may include breaking changes -- renamed parameters, changed defaults, restructured APIs. As the library grows, especially as capabilities gain provider-native support (starting as a local implementation, then auto-switching to the provider's built-in API when available), we may need to reshape APIs we couldn't fully anticipate in the initial design.
- Patch releases (0.1.0 → 0.1.1) will not intentionally break existing behavior.
- All breaking changes are documented in release notes with migration guidance.
- Where practical, we'll keep the previous behavior available under a deprecated name or configuration option before removing it.
This is why Pydantic AI Harness is a separate package from Pydantic AI, which has a stricter version policy. As the core capabilities stabilize, we'll move toward 1.0 with stability guarantees to match.
Pydantic AI references
- Capabilities -- what capabilities are, built-in capabilities, building your own
- Hooks -- lifecycle hooks reference, ordering, error handling
- Extensibility -- publishing packages, third-party ecosystem
- Toolsets -- building tools for capabilities
- API reference -- full API docs
Part of the Pydantic Stack
The Pydantic Stack is everything you need to ship production-grade AI agents:
- Pydantic AI - Type-safe agent framework
- Pydantic Logfire - AI-first, full-stack observability
- Logfire AI Gateway - Unified LLM proxy
License
MIT -- see LICENSE.
