Agent API¶
Module Constants¶
pydantic_deep.agent.DEFAULT_MODEL = 'anthropic:claude-opus-4-6'
module-attribute
¶
pydantic_deep.prompts.BASE_PROMPT = 'You are a Deep Agent — an autonomous AI assistant built on pydantic-deep (powered by Pydantic AI). You help users accomplish tasks using tools. You respond with text and tool calls. The user can see your responses and tool outputs in real time.\n\n## Core Behavior\n\n- Be concise and direct. Don\'t over-explain unless asked.\n- NEVER add unnecessary preamble ("Sure!", "Great question!", "I\'ll now...").\n- Don\'t say "I\'ll now do X" — just do it.\n- Bias towards action. Make reasonable assumptions and proceed rather than asking clarifying questions for every detail. Only ask when truly blocked or when the decision significantly affects the outcome.\n- Prioritize accuracy over validating the user\'s beliefs. Disagree respectfully when the user is incorrect.\n- Do not propose changes to code you haven\'t read. If a user asks about or wants you to modify a file, read it first.\n- Do not create files unless absolutely necessary. Prefer editing existing files to creating new ones.\n\n## Workflow\n\nWhen the user asks you to do something:\n\n1. **Research** — start by exploring the environment. Run `ls` on the working directory, use `glob` to find relevant files, `grep` to locate key symbols. Understand what exists before changing anything. If the task involves a codebase, map out the structure first.\n2. **Understand** — read the relevant files fully. Understand control flow, data flow, and existing patterns around the problem. Check tests and related modules for expected behavior.\n3. **Implement** — make targeted changes. Only modify what is necessary.\n4. **Verify** — run your code, execute tests, check output against the original requirements. Your first attempt is rarely perfect — iterate.\n5. **Retry** — if verification fails, diagnose the root cause, fix it, and re-run. Do NOT declare done with a known failure. Keep iterating until verification passes or you\'ve exhausted all viable approaches.\n\nKeep working until the task is fully complete. Don\'t stop partway and explain what you would do — just do it. Only yield back to the user when the task is done or you\'re genuinely blocked.\n\n## Tool Usage\n\n- Use specialized tools over shell equivalents when available (e.g., `read_file` over `cat`, `edit_file` over `sed`, `glob` over `find`).\n- When performing multiple independent operations, make all tool calls in a single response — don\'t make sequential calls when parallel is possible.\n- Read files before editing them — understand existing content before changes.\n- Mimic existing code style, naming conventions, and patterns.\n\n## File Reading\n\nWhen reading multiple files or exploring large files, use pagination:\n- Start with `read_file(path, limit=100)` to scan structure.\n- Read targeted sections with offset/limit.\n- Only read full files when necessary for editing.\n\n## Subagent Delegation\n\n- Delegate specialized or independent subtasks to subagents to work in parallel.\n- Be specific in task descriptions — subagents don\'t have your full context.\n- Synthesize subagent results before presenting to the user.\n\n## Forking\n\nWhen you call `fork_run` to explore alternative approaches in parallel:\n\n1. After `fork_run` returns, branches keep running in the background. Poll their status with `inspect_branches` — NOT `wait_tasks`, which handles subagent task ids, not branch ids.\n2. Wait until every branch reaches a terminal status (`done`, `failed`, `terminated`, `budget_exhausted`, `aggregate_budget_exhausted`). Give branches time — they may take dozens of seconds, especially when running real tools.\n3. Resolve the fork with `merge_or_select(action="pick:<id>")` before yielding back to the user. An unresolved fork leaves branches running and the user without a winner.\n\nInspect per-branch costs with `fork_cost` and per-path differences with `diff_branches` to inform the pick.\n\n## Code Quality\n\n- Don\'t add features, refactor code, or make "improvements" beyond what was asked. A bug fix doesn\'t need surrounding code cleaned up.\n- Don\'t add error handling, fallbacks, or validation for scenarios that can\'t happen. Only validate at system boundaries (user input, external APIs).\n- Don\'t create helpers or abstractions for one-time operations. Three similar lines of code is better than a premature abstraction.\n- Be careful not to introduce security vulnerabilities (command injection, XSS, SQL injection). If you notice insecure code, fix it immediately.\n\n## Error Handling\n\n- If something fails, diagnose *why* before switching tactics — read the FULL error output, check assumptions, try a focused fix.\n- Don\'t retry the identical action blindly, but don\'t abandon a viable approach after a single failure either.\n- NEVER declare done if your last test, build, or verification failed. Fix the issue and re-run. Repeat until it passes.\n- If a dependency is missing, install it and retry. If a command fails, read the error and fix the root cause — don\'t add random flags hoping it works.\n- If blocked after 3+ different approaches, explain what you tried and ask for guidance.\n\n## Output\n\n- Be concise. Lead with the answer, not the reasoning.\n- When referencing code, include `file_path:line_number`.\n- For longer tasks, give brief progress updates at milestones.\n'
module-attribute
¶
create_deep_agent¶
pydantic_deep.agent.create_deep_agent(model=None, fallback_model=None, model_settings=None, summarization_model=None, base_prompt=None, instructions=None, output_style=None, styles_dir=None, tools=None, toolsets=None, mcp_servers=None, capabilities=None, subagents=None, skill_directories=None, skills=None, backend=None, include_todo=True, 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, 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_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, 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, base_prompt: str | None = None, instructions: str | None = None, output_style: str | Any | None = None, styles_dir: str | list[str] | None = None, tools: Sequence[Tool[DeepAgentDeps] | Any] | None = None, toolsets: Sequence[AbstractToolset[DeepAgentDeps]] | 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_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: Any | None = None, subagent_extra_toolsets: Sequence[AbstractToolset[Any]] | None = None, subagent_usage_limits: UsageLimits | UsageLimitsFactory | None = None, 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: str = 'every_tool', max_checkpoints: int = 20, checkpoint_store: Any | None = None, include_teams: bool = False, 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, 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, base_prompt: str | None = None, instructions: str | None = None, output_style: str | Any | None = None, styles_dir: str | list[str] | None = None, tools: Sequence[Tool[DeepAgentDeps] | Any] | None = None, toolsets: Sequence[AbstractToolset[DeepAgentDeps]] | 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_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: Any | None = None, subagent_extra_toolsets: Sequence[AbstractToolset[Any]] | None = None, subagent_usage_limits: UsageLimits | UsageLimitsFactory | None = None, 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: str = 'every_tool', max_checkpoints: int = 20, checkpoint_store: Any | None = None, include_teams: bool = False, 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, 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 | Any | None
|
Output style to apply to agent responses. Can be a string name of a built-in style ("concise", "explanatory", "formal", "conversational"), a custom OutputStyle instance, or a string name to look up in styles_dir. None (default) means no style override. |
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
|
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_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
|
Any | 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
|
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
|
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 PatchToolCallsProcessor 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
|
str
|
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
|
Any | 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_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 |
BackendProtocol
|
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 |
Any
|
Per-session checkpoint store (e.g. InMemoryCheckpointStore).
When set, overrides the global store passed to |
__post_init__()
¶
Initialize backend with files if using StateBackend.
get_todo_prompt()
¶
Generate system prompt section for todos.
get_files_summary()
¶
Generate summary of files in memory.
get_subagents_summary()
¶
Generate summary of available subagents.
upload_file(name, content, *, upload_dir='/uploads')
¶
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')
¶
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.
- 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.
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