Agent API¶
Module Constants¶
pydantic_deep.agent.DEFAULT_MODEL = 'anthropic:claude-opus-4-6'
module-attribute
¶
Primary model for the top-level deep agent.
pydantic_deep.prompts.BASE_PROMPT = build_system_prompt()
module-attribute
¶
create_deep_agent¶
pydantic_deep.agent.create_deep_agent(model=None, fallback_model=None, model_settings=None, summarization_model=None, instructions=None, output_style=None, styles_dir=None, tools=None, toolsets=None, extra_toolsets=None, mcp_servers=None, capabilities=None, subagents=None, skill_directories=None, skills=None, backend=None, include_todo=True, include_current_todos=False, include_filesystem=True, include_subagents=True, include_skills=True, include_builtin_subagents=True, include_plan=True, max_nesting_depth=1, subagent_registry=None, subagent_extra_toolsets=None, subagent_usage_limits=None, subagent_ask_timeout_seconds=DEFAULT_SUBAGENT_ASK_TIMEOUT_SECONDS, include_execute=None, interrupt_on=None, output_type=None, history_processors=None, eviction_token_limit=20000, max_binary_content=3, edit_format='hashline', context_manager=True, context_manager_max_tokens=None, on_context_update=None, on_before_compress=None, on_after_compress=None, on_eviction=None, context_files=None, context_discovery=False, include_memory=True, memory_dir=None, retries=3, hooks=None, patch_tool_calls=True, include_checkpoints=False, checkpoint_frequency='every_tool', max_checkpoints=20, checkpoint_store=None, include_teams=False, include_monitoring=True, include_improve=False, include_liteparse=False, stuck_loop_detection=True, periodic_reminder=None, web_search=True, web_fetch=True, thinking='high', include_history_archive=True, history_messages_path='.pydantic-deep/messages.json', cost_tracking=True, cost_budget_usd=None, on_cost_update=None, middleware=None, plans_dir=None, message_queue=None, forking=False, tool_search=False, instrument=None, **agent_kwargs)
¶
create_deep_agent(model: str | Model | None = None, fallback_model: str | Model | list[str | Model] | None = None, model_settings: dict[str, Any] | None = None, summarization_model: str | None = None, instructions: str | None = None, output_style: str | OutputStyle | None = None, styles_dir: str | list[str] | None = None, tools: Sequence[Tool[DeepAgentDeps] | Any] | None = None, toolsets: Sequence[AbstractToolset[DeepAgentDeps]] | None = None, extra_toolsets: Sequence[AbstractToolset[Any]] | None = None, mcp_servers: Sequence[AbstractToolset[Any]] | None = None, capabilities: Sequence[AbstractCapability[Any]] | None = None, subagents: list[SubAgentConfig] | None = None, skill_directories: list[dict[str, Any]] | list[str] | list[BackendSkillsDirectory] | None = None, skills: list[Skill] | None = None, backend: BackendProtocol | None = None, include_todo: bool = True, include_current_todos: bool = False, include_filesystem: bool = True, include_subagents: bool = True, include_skills: bool = True, include_builtin_subagents: bool = True, include_plan: bool = True, max_nesting_depth: int = 1, subagent_registry: DynamicAgentRegistry | None = None, subagent_extra_toolsets: Sequence[AbstractToolset[Any]] | None = None, subagent_usage_limits: UsageLimits | UsageLimitsFactory | None = None, subagent_ask_timeout_seconds: float = DEFAULT_SUBAGENT_ASK_TIMEOUT_SECONDS, include_execute: bool | None = None, interrupt_on: dict[str, bool] | None = None, output_type: None = None, history_processors: Sequence[HistoryProcessor[DeepAgentDeps]] | None = None, eviction_token_limit: int | None = 20000, max_binary_content: int | None = 3, edit_format: str = 'hashline', context_manager: bool = True, context_manager_max_tokens: int | None = None, on_context_update: Any | None = None, on_before_compress: Any | None = None, on_after_compress: Any | None = None, on_eviction: Any | None = None, context_files: list[str] | None = None, context_discovery: bool = False, include_memory: bool = True, memory_dir: str | None = None, retries: int = 3, hooks: list[Any] | None = None, patch_tool_calls: bool = True, include_checkpoints: bool = False, checkpoint_frequency: CheckpointFrequency = 'every_tool', max_checkpoints: int = 20, checkpoint_store: CheckpointStore | None = None, include_teams: bool = False, include_monitoring: bool = True, include_improve: bool = False, include_liteparse: bool = False, stuck_loop_detection: bool = True, periodic_reminder: PeriodicReminderConfig | bool | None = None, web_search: bool = True, web_fetch: bool = True, thinking: bool | str = 'high', include_history_archive: bool = True, history_messages_path: str = '.pydantic-deep/messages.json', cost_tracking: bool = True, cost_budget_usd: float | None = None, on_cost_update: Any | None = None, middleware: Sequence[Any] | None = None, plans_dir: str | None = None, message_queue: MessageQueue | None = None, forking: bool | LiveForkCapability = False, tool_search: bool = False, instrument: bool | None = None, **agent_kwargs: Any) -> Agent[DeepAgentDeps, str]
create_deep_agent(model: str | Model | None = None, fallback_model: str | Model | list[str | Model] | None = None, model_settings: dict[str, Any] | None = None, summarization_model: str | None = None, instructions: str | None = None, output_style: str | OutputStyle | None = None, styles_dir: str | list[str] | None = None, tools: Sequence[Tool[DeepAgentDeps] | Any] | None = None, toolsets: Sequence[AbstractToolset[DeepAgentDeps]] | None = None, extra_toolsets: Sequence[AbstractToolset[Any]] | None = None, mcp_servers: Sequence[AbstractToolset[Any]] | None = None, capabilities: Sequence[AbstractCapability[Any]] | None = None, subagents: list[SubAgentConfig] | None = None, skill_directories: list[dict[str, Any]] | list[str] | list[BackendSkillsDirectory] | None = None, skills: list[Skill] | None = None, backend: BackendProtocol | None = None, include_todo: bool = True, include_current_todos: bool = False, include_filesystem: bool = True, include_subagents: bool = True, include_skills: bool = True, include_builtin_subagents: bool = True, include_plan: bool = True, max_nesting_depth: int = 1, subagent_registry: DynamicAgentRegistry | None = None, subagent_extra_toolsets: Sequence[AbstractToolset[Any]] | None = None, subagent_usage_limits: UsageLimits | UsageLimitsFactory | None = None, subagent_ask_timeout_seconds: float = DEFAULT_SUBAGENT_ASK_TIMEOUT_SECONDS, include_execute: bool | None = None, interrupt_on: dict[str, bool] | None = None, *, output_type: OutputSpec[OutputDataT], history_processors: Sequence[HistoryProcessor[DeepAgentDeps]] | None = None, eviction_token_limit: int | None = 20000, max_binary_content: int | None = 3, edit_format: str = 'hashline', context_manager: bool = True, context_manager_max_tokens: int | None = None, on_context_update: Any | None = None, on_before_compress: Any | None = None, on_after_compress: Any | None = None, on_eviction: Any | None = None, context_files: list[str] | None = None, context_discovery: bool = False, include_memory: bool = True, memory_dir: str | None = None, retries: int = 3, hooks: list[Any] | None = None, patch_tool_calls: bool = True, include_checkpoints: bool = False, checkpoint_frequency: CheckpointFrequency = 'every_tool', max_checkpoints: int = 20, checkpoint_store: CheckpointStore | None = None, include_teams: bool = False, include_monitoring: bool = True, include_improve: bool = False, include_liteparse: bool = False, stuck_loop_detection: bool = True, periodic_reminder: PeriodicReminderConfig | bool | None = None, web_search: bool = True, web_fetch: bool = True, thinking: bool | str = 'high', include_history_archive: bool = True, history_messages_path: str = '.pydantic-deep/messages.json', cost_tracking: bool = True, cost_budget_usd: float | None = None, on_cost_update: Any | None = None, middleware: Sequence[Any] | None = None, plans_dir: str | None = None, message_queue: MessageQueue | None = None, forking: bool | LiveForkCapability = False, tool_search: bool = False, instrument: bool | None = None, **agent_kwargs: Any) -> Agent[DeepAgentDeps, OutputDataT]
Create a deep agent with planning, filesystem, subagent, and skills capabilities.
This factory function creates a fully-configured Agent with: - Todo toolset for task planning and tracking - Filesystem toolset for file operations - Subagent toolset for task delegation - Skills toolset for modular capability extension - Dynamic system prompts based on current state - Optional structured output via output_type - Optional history processing (e.g., summarization)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
model
|
str | Model | None
|
Model to use (default: anthropic:claude-opus-4-6). |
None
|
fallback_model
|
str | Model | list[str | Model] | None
|
One or more fallback models to try when the primary
model fails with a transient error (rate limit, 5xx, timeout).
Accepts a single model name/instance or an ordered list forming a
fallback chain. Auth/permission errors (401/403) never trigger
fallback. |
None
|
instructions
|
str | None
|
System prompt for the agent. When provided, replaces the
default |
None
|
output_style
|
str | OutputStyle | None
|
Output style to apply to agent responses. Can be a
built-in style name (a key of |
None
|
styles_dir
|
str | list[str] | None
|
Directory or list of directories to discover custom output styles from. Style files are markdown files with YAML frontmatter (name, description) in the directory root. |
None
|
tools
|
Sequence[Tool[DeepAgentDeps] | Any] | None
|
Additional tools to register. |
None
|
toolsets
|
Sequence[AbstractToolset[DeepAgentDeps]] | None
|
Additional toolsets to register. |
None
|
extra_toolsets
|
Sequence[AbstractToolset[Any]] | None
|
Extra toolsets appended after the built-in set on the main agent only (not propagated to subagents). |
None
|
subagent_extra_toolsets
|
Sequence[AbstractToolset[Any]] | None
|
Extra toolsets made available to spawned subagents (not the main agent). |
None
|
edit_format
|
str
|
Edit strategy for the file-edit tool (default "hashline"). |
'hashline'
|
mcp_servers
|
Sequence[AbstractToolset[Any]] | None
|
MCP server toolsets to attach (e.g. built via
|
None
|
capabilities
|
Sequence[AbstractCapability[Any]] | None
|
Additional capabilities to register. |
None
|
subagents
|
list[SubAgentConfig] | None
|
Subagent configurations for the task tool. |
None
|
skill_directories
|
list[dict[str, Any]] | list[str] | list[BackendSkillsDirectory] | None
|
Directories to discover skills from. Accepts plain string paths or BackendSkillsDirectory instances. |
None
|
skills
|
list[Skill] | None
|
Skill instances to register directly. |
None
|
backend
|
BackendProtocol | None
|
File storage backend (default: StateBackend). |
None
|
include_todo
|
bool
|
Whether to include the todo toolset. |
True
|
include_current_todos
|
bool
|
Inject the live todo list into the system prompt.
Off by default: the instructions open the provider's prompt-cache
prefix, so re-rendering them on every todo mutation invalidates the
whole cached prefix mid-run. The list is already in the append-only
message history ( |
False
|
include_filesystem
|
bool
|
Whether to include the filesystem toolset. |
True
|
include_subagents
|
bool
|
Whether to include the subagent toolset. |
True
|
include_skills
|
bool
|
Whether to include the skills toolset. |
True
|
include_builtin_subagents
|
bool
|
Whether to include built-in subagents (research). |
True
|
include_plan
|
bool
|
Whether to include the built-in 'planner' subagent that
provides Claude Code-style plan mode. The planner analyzes code,
asks clarifying questions via |
True
|
max_nesting_depth
|
int
|
Maximum subagent nesting depth. 1 (default) means subagents can spawn one level of their own subagents. Set to 0 to disable nested delegation. |
1
|
subagent_registry
|
DynamicAgentRegistry | None
|
Optional DynamicAgentRegistry instance. When provided, the task tool will also look up dynamically created agents from the registry (created via create_agent_factory_toolset). |
None
|
subagent_usage_limits
|
UsageLimits | UsageLimitsFactory | None
|
|
None
|
subagent_ask_timeout_seconds
|
float
|
How long a subagent blocked in |
DEFAULT_SUBAGENT_ASK_TIMEOUT_SECONDS
|
include_execute
|
bool | None
|
Whether to include the execute tool. If None (default), automatically determined based on whether backend is a SandboxProtocol. Set to True to force include even when backend is None (useful when backend is provided via deps at runtime). |
None
|
interrupt_on
|
dict[str, bool] | None
|
Map of tool names to approval requirements. e.g., {"execute": True, "write_file": True} |
None
|
output_type
|
OutputSpec[OutputDataT] | None
|
Structured output type (Pydantic model, dataclass, TypedDict). When specified, the agent will return this type instead of str. |
None
|
history_processors
|
Sequence[HistoryProcessor[DeepAgentDeps]] | None
|
Sequence of history processors to apply to messages before sending to the model. Useful for summarization, filtering, etc. |
None
|
eviction_token_limit
|
int | None
|
Token threshold for large tool output eviction. Tool outputs exceeding this limit are saved to files and replaced with a preview + file reference. Defaults to 20,000. Set to None to disable eviction. |
20000
|
max_binary_content
|
int | None
|
Maximum number of multimodal binary parts
(e.g. |
3
|
context_manager
|
bool
|
Whether to enable the ContextManagerMiddleware for automatic token tracking and auto-compression. When True (default), the middleware monitors token usage and triggers LLM-based summarization when approaching the token budget. Also provides tool output truncation when middleware wrapping is active. |
True
|
context_manager_max_tokens
|
int | None
|
Maximum token budget for the conversation. When None (default), auto-detected from genai-prices based on the model name. Falls back to 200,000 if model is not found. Used by ContextManagerMiddleware to calculate usage percentage and determine when to trigger auto-compression. Defaults to 200,000. |
None
|
on_context_update
|
Any | None
|
Callback for context usage updates. Called with
|
None
|
on_before_compress
|
Any | None
|
Optional callback invoked just before an automatic context compression runs. |
None
|
on_after_compress
|
Any | None
|
Optional callback invoked just after a context compression completes. |
None
|
on_eviction
|
Any | None
|
Optional callback invoked when a large tool output is
evicted from history by |
None
|
stuck_loop_detection
|
bool
|
Whether to enable |
True
|
summarization_model
|
str | None
|
Model to use for LLM-based context compression
summaries. Defaults to |
None
|
context_files
|
list[str] | None
|
List of paths to context files in the backend (e.g., ["/project/DEEP.md", "/project/SOUL.md"]). Files are loaded from the runtime backend (ctx.deps.backend) and injected into the system prompt. Missing files are silently skipped. |
None
|
context_discovery
|
bool
|
Whether to auto-discover context files at the backend root (/). Scans for AGENTS.md, SOUL.md. Defaults to False. |
False
|
include_memory
|
bool
|
Whether to include the agent memory toolset.
When True, the main agent and all subagents get persistent
memory stored as MEMORY.md files in the backend. Memory is
auto-loaded into the system prompt and writable via tools
(read_memory, write_memory, update_memory). Per-subagent
memory can be disabled via |
True
|
memory_dir
|
str | None
|
Base directory for memory files in the backend.
Each agent gets its own subdirectory:
|
None
|
retries
|
int
|
Maximum number of retries for tool calls. Defaults to 3. |
3
|
hooks
|
list[Any] | None
|
List of Hook instances for Claude Code-style lifecycle hooks. Hooks execute shell commands or Python handlers on tool events (PRE_TOOL_USE, POST_TOOL_USE, POST_TOOL_USE_FAILURE). Command hooks require a SandboxProtocol backend (LocalBackend or DockerSandbox). Adds HooksCapability to the agent. |
None
|
cost_tracking
|
bool
|
Whether to enable automatic cost tracking via CostTracking capability (from pydantic-ai-shields). When True (default), token usage and USD costs are tracked across runs. |
True
|
cost_budget_usd
|
float | None
|
Maximum allowed cumulative cost in USD. When exceeded, the next run raises BudgetExceededError. None (default) means unlimited. |
None
|
on_cost_update
|
Any | None
|
Callback for cost updates after each run. Called with a CostInfo object containing run and cumulative token/cost data. Supports sync and async callables. |
None
|
patch_tool_calls
|
bool
|
Whether to enable PatchToolCallsCapability that fixes orphaned tool calls in message history. Useful when resuming interrupted conversations. Defaults to True. |
True
|
include_checkpoints
|
bool
|
Whether to enable conversation checkpointing.
When True, adds CheckpointMiddleware (auto-saves snapshots)
and CheckpointToolset (save_checkpoint, list_checkpoints,
rewind_to tools). The checkpoint store is resolved from
|
False
|
checkpoint_frequency
|
CheckpointFrequency
|
When to auto-save checkpoints:
|
'every_tool'
|
max_checkpoints
|
int
|
Maximum number of checkpoints to keep. Oldest checkpoints are pruned when this limit is exceeded. Defaults to 20. |
20
|
checkpoint_store
|
CheckpointStore | None
|
Checkpoint storage backend. When None (default),
uses InMemoryCheckpointStore. Can also be set per-session
via |
None
|
include_teams
|
bool
|
Whether to include the team management toolset. |
False
|
include_improve
|
bool
|
Whether to include the self-improvement toolset
( |
False
|
include_liteparse
|
bool
|
Whether to include the LiteParse document parsing
toolset ( |
False
|
periodic_reminder
|
PeriodicReminderConfig | bool | None
|
Inject a task reminder every N turns to keep the
agent anchored on its original goal.
|
None
|
web_search
|
bool
|
Whether to include the |
True
|
web_fetch
|
bool
|
Whether to include the |
True
|
thinking
|
bool | str
|
Thinking/reasoning effort level. |
'high'
|
include_history_archive
|
bool
|
Whether to persist full conversation history
before context compression discards messages. Adds a
|
True
|
history_messages_path
|
str
|
Path to the messages.json file that stores
the full conversation history. Defaults to
|
'.pydantic-deep/messages.json'
|
middleware
|
Sequence[Any] | None
|
List of additional AbstractCapability instances to include. These extend the agent with custom lifecycle hooks. |
None
|
plans_dir
|
str | None
|
Directory to save plan files from the planner subagent.
Defaults to |
None
|
message_queue
|
MessageQueue | None
|
Optional :class: |
None
|
forking
|
bool | LiveForkCapability
|
Enable Live Run Forking. |
False
|
model_settings
|
dict[str, Any] | None
|
Provider-specific model settings (temperature, thinking,
etc.). Passed directly to the pydantic-ai Agent. Common keys:
|
None
|
instrument
|
bool | None
|
Enable OpenTelemetry/Logfire instrumentation. When True,
the agent emits spans for LLM calls, tool invocations, and token
usage. Requires |
None
|
**agent_kwargs
|
Any
|
Additional arguments passed to Agent constructor. |
{}
|
Returns:
| Type | Description |
|---|---|
Agent[DeepAgentDeps, OutputDataT] | Agent[DeepAgentDeps, str]
|
Configured Agent instance. Returns Agent[DeepAgentDeps, OutputDataT] if |
Agent[DeepAgentDeps, OutputDataT] | Agent[DeepAgentDeps, str]
|
output_type is specified, otherwise Agent[DeepAgentDeps, str]. |
Example
from pydantic import BaseModel
from pydantic_deep import (
create_deep_agent, DeepAgentDeps, StateBackend, create_summarization_processor
)
# Basic usage with string output
agent = create_deep_agent(
instructions="You are a coding assistant",
)
# With structured output
class CodeAnalysis(BaseModel):
language: str
issues: list[str]
suggestions: list[str]
agent = create_deep_agent(
output_type=CodeAnalysis,
)
# Context management is ON by default (token tracking + auto-compression)
# Disable it or customize:
agent = create_deep_agent(context_manager=False)
agent = create_deep_agent(
context_manager_max_tokens=128_000,
on_context_update=lambda pct, cur, mx: print(f"{pct:.0%} used"),
)
deps = DeepAgentDeps(backend=StateBackend())
result = await agent.run("Analyze this code", deps=deps)
Parameters¶
Core¶
| Parameter | Type | Default | Description |
|---|---|---|---|
model |
str \| Model \| None |
None (resolves to DEFAULT_MODEL = "anthropic:claude-opus-4-6") |
LLM model identifier |
fallback_model |
str \| Model \| list[str \| Model] \| None |
None |
Fallback model(s) tried on transient errors. See Fallback Models |
model_settings |
dict[str, Any] \| None |
None |
Model settings passed to the underlying model |
summarization_model |
str \| None |
None (defaults to DEFAULT_SUMMARIZATION_MODEL) |
Model used by the context manager for compression |
base_prompt |
str \| None |
None (defaults to BASE_PROMPT) |
Base system prompt to build on |
instructions |
str \| None |
None |
System prompt for the agent (replaces BASE_PROMPT when set) |
output_style |
str \| OutputStyle \| None |
None |
Output style (built-in name or custom) |
styles_dir |
str \| list[str] \| None |
None |
Directories for custom style files |
tools |
Sequence[Tool \| Any] \| None |
None |
Additional custom tools |
toolsets |
Sequence[AbstractToolset] \| None |
None |
Additional toolsets |
mcp_servers |
Sequence[AbstractToolset] \| None |
None |
MCP server toolsets to attach. See MCP |
capabilities |
Sequence[AbstractCapability] \| None |
None |
Additional capabilities to register |
backend |
BackendProtocol \| None |
StateBackend() |
File storage backend |
output_type |
OutputSpec \| None |
None |
Pydantic model for structured output |
edit_format |
str |
"hashline" |
Edit format used by the file edit tool |
retries |
int |
3 |
Max retries for tool calls |
instrument |
bool \| None |
None |
Enable instrumentation (Logfire/OpenTelemetry) |
Feature Toggles¶
| Parameter | Type | Default | Description |
|---|---|---|---|
include_todo |
bool |
True |
Include TodoToolset |
include_current_todos |
bool |
False |
Inject the live todo list into the system prompt (invalidates the provider's prompt cache on every mutation) |
include_filesystem |
bool |
True |
Include Console Toolset |
include_subagents |
bool |
True |
Include SubAgentToolset |
include_skills |
bool |
True |
Include SkillsToolset |
include_builtin_subagents |
bool |
True |
Include built-in subagents (research) |
include_plan |
bool |
True |
Include planner subagent |
include_execute |
bool \| None |
None |
Include execute tool (auto-detected) |
include_memory |
bool |
True |
Persistent agent memory |
include_checkpoints |
bool |
False |
Conversation checkpointing |
include_teams |
bool |
False |
Agent teams with shared todos |
include_improve |
bool |
False |
Self-improvement toolset |
include_liteparse |
bool |
False |
Document parsing tools (LiteParse) |
include_history_archive |
bool |
True |
Persist message history to disk |
stuck_loop_detection |
bool |
True |
Detect repetitive agent behavior |
forking |
bool \| LiveForkCapability |
False |
Live run forking (parallel branches) |
patch_tool_calls |
bool |
True |
Fix orphaned tool calls |
web_search |
bool |
True |
WebSearch capability |
web_fetch |
bool |
True |
WebFetch capability |
thinking |
bool \| str |
"high" |
Thinking effort (True/False/"minimal"/"low"/"medium"/"high"/"xhigh") |
Subagents¶
| Parameter | Type | Default | Description |
|---|---|---|---|
subagents |
list[SubAgentConfig] \| None |
None |
Subagent configurations |
max_nesting_depth |
int |
1 |
Max subagent nesting depth |
subagent_registry |
DynamicAgentRegistry \| None |
None |
Dynamic agent registry |
Skills¶
| Parameter | Type | Default | Description |
|---|---|---|---|
skill_directories |
list \| None |
None |
Skill discovery directories |
Context Management¶
| Parameter | Type | Default | Description |
|---|---|---|---|
context_manager |
bool |
True |
Token tracking + auto-compression |
context_manager_max_tokens |
int \| None |
None |
Token budget (auto-detected from model when None) |
on_context_update |
Callable \| None |
None |
Callback: (pct, current, max) |
on_before_compress |
Callable \| None |
None |
Callback fired before compression |
on_after_compress |
Callable \| None |
None |
Callback fired after compression |
on_eviction |
Callable \| None |
None |
Callback fired when a large output is evicted |
context_files |
list[str] \| None |
None |
Context file paths |
context_discovery |
bool |
False |
Auto-discover AGENTS.md, SOUL.md |
history_processors |
Sequence \| None |
None |
History processors |
eviction_token_limit |
int \| None |
20_000 |
Large output eviction threshold |
max_binary_content |
int \| None |
3 |
Max binary tool results kept before pruning |
Checkpointing¶
| Parameter | Type | Default | Description |
|---|---|---|---|
checkpoint_frequency |
str |
"every_tool" |
Auto-save frequency |
max_checkpoints |
int |
20 |
Max checkpoints to keep |
checkpoint_store |
CheckpointStore \| None |
None |
Checkpoint storage backend |
Memory¶
| Parameter | Type | Default | Description |
|---|---|---|---|
memory_dir |
str \| None |
"/.deep/memory" |
Base directory for memory files |
Cost Tracking¶
| Parameter | Type | Default | Description |
|---|---|---|---|
cost_tracking |
bool |
True |
Enable cost tracking |
cost_budget_usd |
float \| None |
None |
Max cumulative cost |
on_cost_update |
Callable \| None |
None |
Callback with CostInfo |
Middleware & Hooks¶
| Parameter | Type | Default | Description |
|---|---|---|---|
middleware |
Sequence[Any] \| None |
None |
Custom middleware |
hooks |
list[Hook] \| None |
None |
Claude Code-style hooks |
Other¶
| Parameter | Type | Default | Description |
|---|---|---|---|
interrupt_on |
dict[str, bool] \| None |
None |
Tools requiring approval |
plans_dir |
str \| None |
"/plans" |
Directory for plan files |
message_queue |
MessageQueue \| None |
None |
Steering/follow-up message queue. See Message Queue |
periodic_reminder |
PeriodicReminderConfig \| bool \| None |
None |
Periodic task reminders. See Periodic Reminder |
history_messages_path |
str |
".pydantic-deep/messages.json" |
Path for the message history archive |
**agent_kwargs |
Any |
- | Additional Agent constructor args |
Returns¶
Agent[DeepAgentDeps, str] or Agent[DeepAgentDeps, OutputDataT] - Configured Pydantic AI agent.
When output_type is provided, returns an agent typed with the output model.
Example¶
from pydantic_deep import create_deep_agent, SubAgentConfig
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-6",
instructions="You are a coding assistant.",
subagents=[
SubAgentConfig(
name="reviewer",
description="Reviews code",
instructions="Review code for issues.",
),
],
skill_directories=[
{"path": "~/.pydantic-deep/skills", "recursive": True},
],
interrupt_on={"execute": True},
)
create_default_deps¶
pydantic_deep.agent.create_default_deps(backend=None)
¶
Create default dependencies for a deep agent.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
backend
|
BackendProtocol | None
|
File storage backend (default: StateBackend). |
None
|
Returns:
| Type | Description |
|---|---|
DeepAgentDeps
|
DeepAgentDeps instance. |
Signature¶
Parameters¶
| Parameter | Type | Default | Description |
|---|---|---|---|
backend |
BackendProtocol \| None |
StateBackend() |
File storage backend |
Returns¶
DeepAgentDeps - Configured dependencies instance.
Example¶
from pydantic_deep import create_default_deps
from pydantic_ai_backends import LocalBackend
# With default StateBackend
deps = create_default_deps()
# With custom backend
deps = create_default_deps(backend=LocalBackend("/workspace"))
run_with_files¶
pydantic_deep.agent.run_with_files(agent, query, deps, files=None, *, upload_dir='/uploads')
async
¶
Run agent with file uploads.
This is a convenience function that uploads files to the backend before running the agent. The files are accessible via file tools (read_file, grep, glob, execute).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
agent
|
Agent[DeepAgentDeps, OutputDataT]
|
The agent to run. |
required |
query
|
str
|
The user query/prompt. |
required |
deps
|
DeepAgentDeps
|
Agent dependencies. |
required |
files
|
list[tuple[str, bytes]] | None
|
List of (filename, content) tuples to upload. |
None
|
upload_dir
|
str
|
Directory to store uploads (default: "/uploads") |
'/uploads'
|
Returns:
| Type | Description |
|---|---|
OutputDataT
|
Agent output (type depends on agent's output_type). |
Example
from pydantic_deep import create_deep_agent, DeepAgentDeps, run_with_files
from pydantic_ai_backends import StateBackend
agent = create_deep_agent()
deps = DeepAgentDeps(backend=StateBackend())
with open("sales.csv", "rb") as f:
result = await run_with_files(
agent,
"Analyze the sales data and find top products",
deps,
files=[("sales.csv", f.read())],
)
Convenience coroutine that uploads files to the backend before running the agent.
The uploaded files become accessible via the file tools (read_file, grep,
glob, execute).
Signature¶
async def run_with_files(
agent: Agent[DeepAgentDeps, OutputDataT],
query: str,
deps: DeepAgentDeps,
files: list[tuple[str, bytes]] | None = None,
*,
upload_dir: str = "/uploads",
) -> OutputDataT
Parameters¶
| Parameter | Type | Default | Description |
|---|---|---|---|
agent |
Agent[DeepAgentDeps, OutputDataT] |
Required | The agent to run |
query |
str |
Required | The user query/prompt |
deps |
DeepAgentDeps |
Required | Agent dependencies |
files |
list[tuple[str, bytes]] \| None |
None |
List of (filename, content) tuples to upload |
upload_dir |
str |
"/uploads" |
Directory to store uploads |
Returns¶
OutputDataT - Agent output (type depends on the agent's output_type).
Example¶
from pydantic_deep import create_deep_agent, DeepAgentDeps, run_with_files
from pydantic_ai_backends import StateBackend
agent = create_deep_agent()
deps = DeepAgentDeps(backend=StateBackend())
with open("sales.csv", "rb") as f:
result = await run_with_files(
agent,
"Analyze the sales data and find top products",
deps,
files=[("sales.csv", f.read())],
)
DeepAgentDeps¶
pydantic_deep.deps.DeepAgentDeps
dataclass
¶
Dependencies for deep agents.
This container holds all the state and resources needed by the agent and its tools during execution.
Attributes:
| Name | Type | Description |
|---|---|---|
backend |
AsyncBackendProtocol | Any
|
File storage backend (StateBackend, FilesystemBackend, etc.) |
files |
dict[str, FileData]
|
In-memory file cache (used with StateBackend) |
todos |
list[Todo]
|
Task list for planning |
subagents |
dict[str, Any]
|
Pre-configured subagents available for delegation |
checkpoint_store |
CheckpointStore | None
|
Per-session checkpoint store (e.g. InMemoryCheckpointStore).
When set, overrides the global store passed to |
__post_init__()
¶
Auto-wrap sync backends and wire StateBackend cache.
get_todo_prompt()
¶
Generate system prompt section for todos.
Ids are included so the model can address a task with
update_todo_status without re-reading the list.
get_files_summary()
¶
Generate summary of files in memory.
get_subagents_summary()
¶
Generate summary of available subagents.
upload_file(name, content, *, upload_dir='/uploads')
async
¶
Upload a file to the backend and track it.
The file is written to the backend and its metadata is stored for display in the system prompt.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
name
|
str
|
Original filename (e.g., "sales.csv") |
required |
content
|
bytes
|
File content as bytes |
required |
upload_dir
|
str
|
Directory to store uploads (default: "/uploads") |
'/uploads'
|
Returns:
| Type | Description |
|---|---|
str
|
The path where the file was stored (e.g., "/uploads/sales.csv") |
upload_files(files, *, upload_dir='/uploads')
async
¶
Upload multiple files to the backend.
Each file is written independently - failures on one file don't affect others. Failed uploads are silently skipped.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
files
|
list[tuple[str, bytes]]
|
List of (filename, content) tuples. |
required |
upload_dir
|
str
|
Directory to store uploads (default: "/uploads"). |
'/uploads'
|
Returns:
| Type | Description |
|---|---|
list[str]
|
List of paths for successfully uploaded files. |
get_uploads_summary()
¶
Generate summary of uploaded files for system prompt.
clone_for_subagent(max_depth=0)
¶
Create a new deps instance for a subagent.
Subagents get: - Same backend (shared) - Empty todos (isolated) - or same todos if share_todos=True - Empty subagents (no nested delegation by default) - Same files (shared) - Same uploads (shared) - Same ask_user callback (propagated) - Same checkpoint_store (shared) - Same message_queue (shared - subagents can steer the parent)
Intentionally NOT propagated (a subagent is its own isolation
domain, so these are reset rather than inherited):
- context_middleware: a CLI-only handle used by /compact and
/context over the parent's history; a subagent has its own
history and never runs those commands.
- monitor_manager: monitors are a top-level concern; a subagent
starts with none rather than inheriting/duplicating the parent's
background watches.
- fork_coordinator: allocated lazily per-run by the fork
capability, so a forking subagent gets its own (mirrors how
clone_for_branch resets it).
- _fork_depth: subagent nesting is bounded separately by
max_depth; fork depth restarts at 0 for the subagent.
- _branch_cost_tracking / _branch_id /
_parent_fork_coordinator: branch bookkeeping set by
ForkCoordinator.fork and only meaningful to an agent wired
with the fork capability; inert on a separately-compiled subagent.
Every other field (backend, files, uploads, ask_user, share_todos,
checkpoint_store, message_queue) is shared with the parent via
replace, so new shared fields propagate automatically.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
max_depth
|
int
|
Maximum nesting depth for subagent. If > 0, subagents dict is copied to allow nested delegation. |
0
|
Definition¶
@dataclass
class DeepAgentDeps:
backend: BackendProtocol = field(default_factory=StateBackend)
files: dict[str, FileData] = field(default_factory=dict)
todos: list[Todo] = field(default_factory=list)
subagents: dict[str, Any] = field(default_factory=dict)
uploads: dict[str, UploadedFile] = field(default_factory=dict)
ask_user: Callable | None = None
share_todos: bool = False
checkpoint_store: CheckpointStore | None = None
Attributes¶
| Attribute | Type | Description |
|---|---|---|
backend |
BackendProtocol |
File storage backend |
files |
dict[str, FileData] |
In-memory file cache |
todos |
list[Todo] |
Task list |
subagents |
dict[str, Any] |
Pre-configured subagent instances |
uploads |
dict[str, UploadedFile] |
Uploaded files metadata |
ask_user |
Callable \| None |
Callback for planner's ask_user tool |
share_todos |
bool |
When True, subagents share parent's todo list |
checkpoint_store |
CheckpointStore \| None |
Per-session checkpoint store |
Methods¶
get_todo_prompt¶
Generate system prompt section for current todos.
get_files_summary¶
Generate summary of files in memory.
get_subagents_summary¶
Generate summary of available subagents.
clone_for_subagent¶
Create isolated dependencies for a subagent.
- Same backend (shared)
- Empty todos (isolated)
- Empty subagents (no nested delegation)
- Same files (shared reference)
Example¶
from pydantic_deep import DeepAgentDeps, StateBackend, Todo
deps = DeepAgentDeps(
backend=StateBackend(),
todos=[
Todo(
content="Review code",
status="pending",
active_form="Reviewing code",
),
],
)
# Access todo prompt
print(deps.get_todo_prompt())
# Clone for subagent
subagent_deps = deps.clone_for_subagent()
assert subagent_deps.todos == [] # Isolated
assert subagent_deps.backend is deps.backend # Shared