API Reference

This page provides comprehensive API documentation for all modules in the pydantic-ai-toolsets package.

Main Package

Pydantic AI Toolsets - Collection of reasoning and agent toolsets.

This package provides a comprehensive set of toolsets for pydantic-ai agents, including reasoning strategies, reflection techniques, and multi-agent capabilities.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_cot_toolset, CoTStorage

storage = CoTStorage()
agent = Agent("openai:gpt-4", toolsets=[create_cot_toolset(storage)])
result = await agent.run("Solve this problem step by step")

class pydantic_ai_toolsets.AddPersonaResponseItem(*, persona_id: str, content: str, references: list[str] = <factory>)

Bases: BaseModel

Input model for the add_persona_response tool.

persona_id: str

content: str

references: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.AggregateItem(*, source_node_ids: list[str], aggregated_content: str, is_solution: bool = False)

Bases: BaseModel

Input model for the aggregate_nodes tool.

source_node_ids: list[str]

aggregated_content: str

is_solution: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.AgreeWithPositionItem(*, target_position_id: str, persona_id: str, content: str, reasoning: list[str] = <factory>)

Bases: BaseModel

Input model for the agree_with_position tool.

target_position_id: str

persona_id: str

content: str

reasoning: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.Answer(*, answer_id: str, question_id: str, answer_text: str, confidence_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, requires_followup: bool = False)

Bases: BaseModel

An answer to a question or sub-question.

Each answer corresponds to a specific question and can be used to answer parent questions or compose the final answer.

answer_id

Unique identifier for this answer.

Type:

str

question_id

ID of the question this answer addresses.

Type:

str

answer_text

The actual answer content.

Type:

str

confidence_score

Optional confidence score (0-100, higher is more confident).

Type:

float | None

requires_followup

Whether this answer needs further sub-questions.

Type:

bool

answer_id: str

question_id: str

answer_text: str

confidence_score: float | None

requires_followup: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.AnswerQuestionItem(*, question_id: str, answer_text: str, confidence_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, requires_followup: bool = False)

Bases: BaseModel

Input model for the answer_question tool.

question_id: str

answer_text: str

confidence_score: float | None

requires_followup: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.AskMainQuestionItem(*, question_text: str)

Bases: BaseModel

Input model for the ask_main_question tool.

This is the model that agents use when calling ask_main_question.

question_text: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.AskSubQuestionItem(*, parent_question_id: str, sub_question_text: str, reasoning: str)

Bases: BaseModel

Input model for the ask_sub_question tool.

parent_question_id: str

sub_question_text: str

reasoning: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.BackpropagateItem(*, node_id: str, reward: Annotated[float, Ge(ge=0.0), Le(le=1.0)])

Bases: BaseModel

Input model for the backpropagate tool.

node_id: str

reward: float

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.BeamCandidate(*, candidate_id: str, content: str, score: float | None = None, depth: Annotated[int, Ge(ge=0)] = 0, parent_id: str | None = None, is_terminal: bool = False, step_index: Annotated[int, Ge(ge=0)] = 0)

Bases: BaseModel

A candidate in the beam search execution.

Each candidate represents a reasoning state at a specific depth in the search. Candidates form a tree structure where each candidate can have multiple children (expansions) but only one parent.

candidate_id

Unique identifier for this candidate.

Type:

str

content

The reasoning content at this candidate.

Type:

str

score

Evaluation score for this candidate (0-100, higher is better).

Type:

float | None

depth

Depth level in the search tree (0 for initial candidates).

Type:

int

parent_id

ID of the parent candidate (None for root candidates).

Type:

str | None

is_terminal

Whether this candidate represents a terminal/solution state.

Type:

bool

step_index

Which beam step this candidate belongs to.

Type:

int

candidate_id: str

content: str

score: float | None

depth: int

parent_id: str | None

is_terminal: bool

step_index: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.BeamStep(*, step_index: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], candidate_ids: list[str] = <factory>, beam_width: ~typing.Annotated[int, ~annotated_types.Ge(ge=1)])

Bases: BaseModel

A step in the beam search execution.

Each step contains the top-k candidates (the “beam”) at that depth level. Steps are ordered sequentially, with step 0 containing initial candidates.

step_index

Sequential index of this step (0-based).

Type:

int

candidate_ids

List of candidate IDs in this step’s beam (top-k).

Type:

list[str]

beam_width

The beam width (k) used for this step.

Type:

int

step_index: int

candidate_ids: list[str]

beam_width: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.BeamStorage(*, track_usage: bool = False)

Bases: object

Default in-memory beam search storage.

Simple implementation that stores candidates and steps in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_beam_toolset, BeamStorage

storage = BeamStorage()
toolset = create_beam_toolset(storage=storage)

# After agent runs, access candidates and steps directly
print(storage.candidates)
print(storage.steps)

# With metrics tracking
storage = BeamStorage(track_usage=True)
toolset = create_beam_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property candidates: dict[str, BeamCandidate]

Get the current dictionary of candidates.

property steps: list[BeamStep]

Get the current list of beam steps.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about the beam search.

Returns:

Dictionary with candidate counts and beam metrics.

beam_width_history() → list[tuple[int, int]]

Get beam width at each step.

Returns:

List of (step_index, beam_width) tuples.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all candidates, steps, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (candidate_id or step index as string)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of candidates and steps.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for candidates and steps.

class pydantic_ai_toolsets.BeamStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for beam search storage implementations.

Any class that has candidates and steps properties can be used as storage for the beam search toolset.

class MyCustomStorage:
    def __init__(self):
        self._candidates: dict[str, BeamCandidate] = {}
        self._steps: list[BeamStep] = []

    @property
    def candidates(self) -> dict[str, BeamCandidate]:
        return self._candidates

    @candidates.setter
    def candidates(self, value: BeamCandidate) -> None:
        self._candidates[value.candidate_id] = value

    @property
    def steps(self) -> list[BeamStep]:
        return self._steps

    @steps.setter
    def steps(self, value: BeamStep) -> None:
        # Update or append step
        ...

property candidates: dict[str, BeamCandidate]

Get the current dictionary of candidates (candidate_id -> BeamCandidate).

property steps: list[BeamStep]

Get the current list of beam steps.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (candidate_id or step_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.BranchEvaluation(*, branch_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str, recommendation: Literal['continue', 'prune', 'merge', 'explore_deeper'])

Bases: BaseModel

Evaluation of a branch/path in the tree.

Used to assess the promise of different reasoning paths.

branch_id

Identifier for the branch being evaluated.

Type:

str

score

Evaluation score (0-100, higher is better).

Type:

float

reasoning

Explanation of why this score was assigned.

Type:

str

recommendation

Recommended action for this branch.

Type:

Literal[‘continue’, ‘prune’, ‘merge’, ‘explore_deeper’]

branch_id: str

score: float

reasoning: str

recommendation: Literal['continue', 'prune', 'merge', 'explore_deeper']

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.BranchEvaluationItem(*, branch_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str, recommendation: Literal['continue', 'prune', 'merge', 'explore_deeper'])

Bases: BaseModel

Input model for the evaluate_branch tool.

branch_id: str

score: float

reasoning: str

recommendation: Literal['continue', 'prune', 'merge', 'explore_deeper']

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.CoTStorage(*, track_usage: bool = False)

Bases: object

Default in-memory chain of thoughts storage.

Simple implementation that stores thoughts in memory. Use this for standalone agents or testing.

_thoughts

Internal list of recorded thoughts.

Type:

list[Thought]

_metrics

Optional usage metrics tracker (enabled via track_usage parameter).

Type:

UsageMetrics | None

from pydantic_ai_toolsets import create_cot_toolset, CoTStorage

storage = CoTStorage()
toolset = create_cot_toolset(storage=storage)

# After agent runs, access thoughts directly
print(storage.thoughts)

# With metrics tracking
storage = CoTStorage(track_usage=True)
toolset = create_cot_toolset(storage=storage)
# After agent runs
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property thoughts: list[Thought]

Get the current list of thoughts.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

Returns:

UsageMetrics instance if track_usage=True was set, otherwise None.

get_statistics() → dict[str, int | float]

Get summary statistics about the chain of thoughts.

Returns:

Dictionary with thought counts and metadata.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all thoughts and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (e.g., thought number as string)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of thoughts and statistics.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for each linkable thought.

class pydantic_ai_toolsets.CoTStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for chain of thoughts storage implementations.

Any class that has a thoughts property (read returns list, write appends Thought) can be used as storage for the CoT toolset.

class MyCustomStorage:
    def __init__(self):
        self._thoughts: list[Thought] = []

    @property
    def thoughts(self) -> list[Thought]:
        return self._thoughts

    @thoughts.setter
    def thoughts(self, value: Thought) -> None:
        self._thoughts.append(value)

property thoughts: list[Thought]

Get the current list of thoughts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (e.g., thought number as string)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.ComposeFinalAnswerItem(*, main_question_id: str, final_answer_text: str, answer_ids_used: Annotated[list[str], MinLen(min_length=1)])

Bases: BaseModel

Input model for the compose_final_answer tool.

main_question_id: str

final_answer_text: str

answer_ids_used: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.CreateCandidateItem(*, content: str, is_terminal: bool = False)

Bases: BaseModel

Input model for the create_candidate tool.

This is the model that agents use when calling create_candidate.

content: str

is_terminal: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.CreateOutputItem(*, content: str)

Bases: BaseModel

Input model for the create_output tool.

This is the model that agents use when calling create_output.

content: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.CreatePersonaItem(*, name: str, persona_type: ~typing.Literal['expert', 'thinking_style', 'stakeholder'], description: str, expertise_areas: list[str] = <factory>)

Bases: BaseModel

Input model for the create_persona tool.

name: str

persona_type: Literal['expert', 'thinking_style', 'stakeholder']

description: str

expertise_areas: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.CritiqueOutputItem(*, output_id: str, problems: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], strengths: list[str] = <factory>, overall_assessment: str, improvement_suggestions: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)])

Bases: BaseModel

Input model for the critique_output tool.

output_id: str

problems: list[str]

strengths: list[str]

overall_assessment: str

improvement_suggestions: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.CritiquePositionItem(*, target_position_id: str, persona_id: str, content: str, specific_points: Annotated[list[str], MinLen(min_length=1)])

Bases: BaseModel

Input model for the critique_position tool.

target_position_id: str

persona_id: str

content: str

specific_points: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.DefendPositionItem(*, position_id: str, persona_id: str, content: str, critiques_addressed: list[str] = <factory>, evidence: list[str] = <factory>)

Bases: BaseModel

Input model for the defend_position tool.

position_id: str

persona_id: str

content: str

critiques_addressed: list[str]

evidence: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.EdgeItem(*, source_id: str, target_id: str, edge_type: Literal['dependency', 'aggregation', 'refinement', 'reference', 'merge'] = 'dependency', weight: Annotated[float | None, Ge(ge=0), Le(le=1)] = None)

Bases: BaseModel

Input model for the create_edge tool.

source_id: str

target_id: str

edge_type: Literal['dependency', 'aggregation', 'refinement', 'reference', 'merge']

weight: float | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ExpandCandidateItem(*, candidate_id: str, expansions: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], is_terminal: list[bool] = <factory>)

Bases: BaseModel

Input model for the expand_candidate tool.

candidate_id: str

expansions: list[str]

is_terminal: list[bool]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ExpandNodeItem(*, node_id: str, children: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], is_terminal: list[bool] = <factory>)

Bases: BaseModel

Input model for the expand_node tool.

node_id: str

children: list[str]

is_terminal: list[bool]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ExtractWebContentItem(*, url: str, output_format: OutputFormat = OutputFormat.TEXT)

Bases: BaseModel

Input model for the extract_web_content tool.

This is the model that agents use when calling extract_web_content. Note: Content extraction only works with web and news search results, not image results.

url: str

output_format: OutputFormat

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ExtractedContent(*, content_id: str, url: str, content: str, output_format: OutputFormat = OutputFormat.TEXT, timestamp: str | None = None)

Bases: BaseModel

Extracted content from a webpage.

content_id

Unique identifier for this extracted content.

Type:

str

url

URL of the webpage that was extracted.

Type:

str

content

The extracted content text.

Type:

str

output_format

Format of the extracted content (txt or markdown).

Type:

pydantic_ai_toolsets.toolsets.search.types.OutputFormat

timestamp

Timestamp when the extraction was performed.

Type:

str | None

content_id: str

url: str

content: str

output_format: OutputFormat

timestamp: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.Feedback(*, feedback_id: str, output_id: str, feedback_type: FeedbackType, dimension: FeedbackDimension, description: str, suggestion: str, priority: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.5, is_actionable: bool = True)

Bases: BaseModel

Feedback analyzing an output and providing actionable suggestions.

Each feedback examines a specific output and provides structured feedback with specific types (additive, subtractive, transformative, corrective) and dimensions.

feedback_id

Unique identifier for this feedback.

Type:

str

output_id

ID of the output being analyzed.

Type:

str

feedback_type

Type of feedback (additive, subtractive, transformative, corrective).

Type:

pydantic_ai_toolsets.toolsets.self_refine.types.FeedbackType

dimension

Dimension being evaluated (factuality, coherence, completeness, style).

Type:

pydantic_ai_toolsets.toolsets.self_refine.types.FeedbackDimension

description

Detailed description of the feedback.

Type:

str

suggestion

Specific, actionable suggestion for improvement.

Type:

str

priority

Priority weight (0-1, higher is more important). Used for weighted feedback.

Type:

float

is_actionable

Whether this feedback provides actionable guidance.

Type:

bool

feedback_id: str

output_id: str

feedback_type: FeedbackType

dimension: FeedbackDimension

description: str

suggestion: str

priority: float

is_actionable: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.FeedbackDimension(*values)

Bases: str, Enum

Dimensions for evaluating feedback quality.

FACTUALITY = 'factuality'

COHERENCE = 'coherence'

COMPLETENESS = 'completeness'

STYLE = 'style'

class pydantic_ai_toolsets.FeedbackItem(*, feedback_type: FeedbackType, dimension: FeedbackDimension, description: str, suggestion: str, priority: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.5)

Bases: BaseModel

A single feedback item within ProvideFeedbackItem.

feedback_type: FeedbackType

dimension: FeedbackDimension

description: str

suggestion: str

priority: float

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.FeedbackType(*values)

Bases: str, Enum

Types of feedback that can be provided.

ADDITIVE = 'additive'

SUBTRACTIVE = 'subtractive'

TRANSFORMATIVE = 'transformative'

CORRECTIVE = 'corrective'

class pydantic_ai_toolsets.FinalAnswer(*, final_answer_id: str, main_question_id: str, final_answer_text: str, composed_from_answers: list[str] = <factory>, is_complete: bool = True)

Bases: BaseModel

The final composed answer to the main question.

The final answer is composed from answers to sub-questions and represents the complete solution to the original question.

final_answer_id

Unique identifier for this final answer.

Type:

str

main_question_id

ID of the main question this answers.

Type:

str

final_answer_text

The composed final answer content.

Type:

str

composed_from_answers

List of answer IDs used to compose this answer.

Type:

list[str]

is_complete

Whether the final answer is complete and ready.

Type:

bool

final_answer_id: str

main_question_id: str

final_answer_text: str

composed_from_answers: list[str]

is_complete: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.GenerateOutputItem(*, content: str, quality_threshold: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, iteration_limit: Annotated[int | None, Ge(ge=1)] = None)

Bases: BaseModel

Input model for the generate_output tool.

This is the model that agents use when calling generate_output.

content: str

quality_threshold: float | None

iteration_limit: int | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.GoTStorage(*, track_usage: bool = False)

Bases: object

Default in-memory graph of thoughts storage.

Simple implementation that stores nodes, edges, and evaluations in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_got_toolset, GoTStorage

storage = GoTStorage()
toolset = create_got_toolset(storage=storage)

# After agent runs, access nodes, edges, and evaluations directly
print(storage.nodes)
print(storage.edges)
print(storage.evaluations)

# With metrics tracking
storage = GoTStorage(track_usage=True)
toolset = create_got_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property nodes: dict[str, GraphNode]

Get the current dictionary of nodes.

property edges: dict[str, GraphEdge]

Get the current dictionary of edges.

property evaluations: dict[str, NodeEvaluation]

Get the current dictionary of node evaluations.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about the graph.

Returns:

Dictionary with node, edge, and evaluation counts.

graph_complexity() → dict[str, float]

Calculate graph complexity metrics.

Returns:

Dictionary with complexity metrics (density, avg degree).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all nodes, edges, evaluations, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id, edge_id, or evaluation_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of nodes, edges, and evaluations.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for nodes, edges, and evaluations.

class pydantic_ai_toolsets.GoTStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for graph of thoughts storage implementations.

Any class that has nodes, edges, and evaluations properties can be used as storage for the GoT toolset.

class MyCustomStorage:
    def __init__(self):
        self._nodes: dict[str, GraphNode] = {}
        self._edges: dict[str, GraphEdge] = {}
        self._evaluations: dict[str, NodeEvaluation] = {}

    @property
    def nodes(self) -> dict[str, GraphNode]:
        return self._nodes

    @nodes.setter
    def nodes(self, value: GraphNode) -> None:
        self._nodes[value.node_id] = value

    @property
    def edges(self) -> dict[str, GraphEdge]:
        return self._edges

    @edges.setter
    def edges(self, value: GraphEdge) -> None:
        self._edges[value.edge_id] = value

    @property
    def evaluations(self) -> dict[str, NodeEvaluation]:
        return self._evaluations

    @evaluations.setter
    def evaluations(self, value: NodeEvaluation) -> None:
        self._evaluations[value.node_id] = value

property nodes: dict[str, GraphNode]

Get the current dictionary of nodes (node_id -> GraphNode).

property edges: dict[str, GraphEdge]

Get the current dictionary of edges (edge_id -> GraphEdge).

property evaluations: dict[str, NodeEvaluation]

Get the current dictionary of node evaluations (node_id -> NodeEvaluation).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id, edge_id, or evaluation_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.GraphEdge(*, edge_id: str, source_id: str, target_id: str, edge_type: Literal['dependency', 'aggregation', 'refinement', 'reference', 'merge'] = 'dependency', weight: float | None = None)

Bases: BaseModel

An edge in a graph of thoughts execution.

Edges represent dependencies, logical connections, or information flow between nodes. They form a directed graph (DAG or with cycles).

edge_id

Unique identifier for this edge.

Type:

str

source_id

ID of the source node.

Type:

str

target_id

ID of the target node.

Type:

str

edge_type

Type of connection (dependency, aggregation, refinement, etc.).

Type:

Literal[‘dependency’, ‘aggregation’, ‘refinement’, ‘reference’, ‘merge’]

weight

Optional weight indicating strength/importance of the connection.

Type:

float | None

edge_id: str

source_id: str

target_id: str

edge_type: Literal['dependency', 'aggregation', 'refinement', 'reference', 'merge']

weight: float | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.GraphNode(*, node_id: str, content: str, evaluation_score: float | None = None, is_solution: bool = False, status: Literal['active', 'completed', 'pruned']='active', aggregated_from: list[str] = <factory>, refined_from: str | None = None, refinement_count: int = 0)

Bases: BaseModel

A node in a graph of thoughts execution.

Each node represents a reasoning state. Nodes form a directed graph structure where edges represent dependencies, logical connections, or information flow.

node_id

Unique identifier for this node.

Type:

str

content

The reasoning content at this node.

Type:

str

evaluation_score

Optional score evaluating this node (0-100).

Type:

float | None

is_solution

Whether this node represents a solution.

Type:

bool

status

Current status of this node.

Type:

Literal[‘active’, ‘completed’, ‘pruned’]

aggregated_from

List of node IDs whose insights were aggregated into this node.

Type:

list[str]

refined_from

Node ID that this node refines (if any).

Type:

str | None

refinement_count

Number of times this node has been refined.

Type:

int

node_id: str

content: str

evaluation_score: float | None

is_solution: bool

status: Literal['active', 'completed', 'pruned']

aggregated_from: list[str]

refined_from: str | None

refinement_count: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.InitiatePersonaDebateItem(*, topic: str, max_rounds: Annotated[int, Ge(ge=1)] = 5)

Bases: BaseModel

Input model for the initiate_persona_debate tool.

topic: str

max_rounds: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.InitiatePersonaSessionItem(*, problem: str, process_type: Literal['sequential', 'interactive', 'devils_advocate'], max_rounds: Annotated[int, Ge(ge=1)] = 3)

Bases: BaseModel

Input model for the initiate_persona_session tool.

problem: str

process_type: Literal['sequential', 'interactive', 'devils_advocate']

max_rounds: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.MCTSNode(*, node_id: str, content: str, visits: Annotated[int, ~annotated_types.Ge(ge=0)] = 0, wins: Annotated[float, ~annotated_types.Ge(ge=0)] = 0.0, parent_id: str | None = None, children_ids: list[str] = <factory>, is_terminal: bool = False, is_expanded: bool = False, depth: Annotated[int, ~annotated_types.Ge(ge=0)] = 0)

Bases: BaseModel

A node in a Monte Carlo Tree Search execution.

Each node represents a state in the search tree. Nodes track visit counts, win/reward totals, and maintain parent-child relationships for the tree structure.

node_id

Unique identifier for this node.

Type:

str

content

The reasoning/state content at this node.

Type:

str

visits

Number of times this node has been visited.

Type:

int

wins

Total reward/wins accumulated from simulations passing through this node.

Type:

float

parent_id

ID of the parent node (None for root node).

Type:

str | None

children_ids

List of child node IDs.

Type:

list[str]

is_terminal

Whether this node represents a terminal/solution state.

Type:

bool

is_expanded

Whether this node has been expanded (has children).

Type:

bool

depth

Depth level in the tree (0 for root).

Type:

int

node_id: str

content: str

visits: int

wins: float

parent_id: str | None

children_ids: list[str]

is_terminal: bool

is_expanded: bool

depth: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.MCTSStorage(*, track_usage: bool = False)

Bases: object

Default in-memory MCTS storage.

Simple implementation that stores nodes in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_mcts_toolset, MCTSStorage

storage = MCTSStorage()
toolset = create_mcts_toolset(storage=storage)

# After agent runs, access nodes directly
print(storage.nodes)

# With metrics tracking
storage = MCTSStorage(track_usage=True)
toolset = create_mcts_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property nodes: dict[str, MCTSNode]

Get the current dictionary of nodes.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

property iteration_count: int

Get the number of MCTS iterations performed.

increment_iteration() → None

Increment the iteration counter.

get_statistics() → dict[str, int | float]

Get summary statistics about the MCTS tree.

Returns:

Dictionary with node counts and tree metrics.

get_ucb1_stats() → list[tuple[str, float, int, float]]

Get UCB1 statistics for all nodes.

Returns:

List of (node_id, win_rate, visits, ucb1_value) tuples. UCB1 calculated with c=sqrt(2).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all nodes and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of nodes and iterations.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for nodes.

class pydantic_ai_toolsets.MCTSStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for MCTS storage implementations.

Any class that has a nodes property can be used as storage for the MCTS toolset.

class MyCustomStorage:
    def __init__(self):
        self._nodes: dict[str, MCTSNode] = {}

    @property
    def nodes(self) -> dict[str, MCTSNode]:
        return self._nodes

    @nodes.setter
    def nodes(self, value: MCTSNode) -> None:
        self._nodes[value.node_id] = value

property nodes: dict[str, MCTSNode]

Get the current dictionary of nodes (node_id -> MCTSNode).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.NodeEvaluation(*, node_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str, recommendation: Literal['keep', 'refine', 'aggregate', 'prune'])

Bases: BaseModel

Evaluation of a node in the graph.

Used to assess the quality and relevance of individual nodes.

node_id

Identifier for the node being evaluated.

Type:

str

score

Evaluation score (0-100, higher is better).

Type:

float

reasoning

Explanation of why this score was assigned.

Type:

str

recommendation

Recommended action for this node.

Type:

Literal[‘keep’, ‘refine’, ‘aggregate’, ‘prune’]

node_id: str

score: float

reasoning: str

recommendation: Literal['keep', 'refine', 'aggregate', 'prune']

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.NodeEvaluationItem(*, node_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str, recommendation: Literal['keep', 'refine', 'aggregate', 'prune'])

Bases: BaseModel

Input model for the evaluate_node tool.

node_id: str

score: float

reasoning: str

recommendation: Literal['keep', 'refine', 'aggregate', 'prune']

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.NodeItem(*, content: str, is_solution: bool = False)

Bases: BaseModel

Input model for the create_node tool.

This is the model that agents use when calling create_node.

content: str

is_solution: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.OrchestrateRoundItem(*, round_number: Annotated[int, Ge(ge=1)])

Bases: BaseModel

Input model for the orchestrate_round tool.

round_number: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.OutputFormat(*values)

Bases: str, Enum

Output format for extracted web content.

TEXT = 'txt'

MARKDOWN = 'markdown'

class pydantic_ai_toolsets.Persona(*, persona_id: str, name: str, persona_type: ~typing.Literal['expert', 'thinking_style', 'stakeholder'], description: str, expertise_areas: list[str] = <factory>)

Bases: BaseModel

A persona representing a distinct viewpoint or expertise.

Personas can be expert personas (domain specialists), thinking style personas (cognitive approaches), or stakeholder personas (interested parties).

persona_id

Unique identifier for this persona.

Type:

str

name

Display name of the persona (e.g., “Clinical Doctor”, “Analytical Persona”).

Type:

str

persona_type

Type of persona - expert, thinking_style, or stakeholder.

Type:

Literal[‘expert’, ‘thinking_style’, ‘stakeholder’]

description

Detailed description of the persona’s background, expertise, and perspective.

Type:

str

expertise_areas

List of specific areas of expertise or focus.

Type:

list[str]

persona_id: str

name: str

persona_type: Literal['expert', 'thinking_style', 'stakeholder']

description: str

expertise_areas: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.PersonaAgreement(*, agreement_id: str, target_position_id: str, persona_id: str, round_number: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], content: str, reasoning: list[str] = <factory>)

Bases: BaseModel

An agreement by a persona with another persona’s position.

Personas can agree with positions made by other personas, providing reasoning for their agreement. This allows for coalition-building and consensus formation.

agreement_id

Unique identifier for this agreement.

Type:

str

target_position_id

ID of the position being agreed with.

Type:

str

persona_id

ID of the persona agreeing with the position.

Type:

str

round_number

The debate round in which this agreement was made.

Type:

int

content

The agreement content explaining why the persona agrees.

Type:

str

reasoning

Specific reasons or points that led to the agreement.

Type:

list[str]

agreement_id: str

target_position_id: str

persona_id: str

round_number: int

content: str

reasoning: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.PersonaCritique(*, critique_id: str, target_position_id: str, persona_id: str, round_number: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], content: str, specific_points: list[str] = <factory>)

Bases: BaseModel

A critique of a position made by a persona.

Each critique identifies weaknesses or challenges to a specific position. Critiques guide the defense and refinement process.

critique_id

Unique identifier for this critique.

Type:

str

target_position_id

ID of the position being critiqued.

Type:

str

persona_id

ID of the persona making this critique.

Type:

str

round_number

The debate round in which this critique was made.

Type:

int

content

The critique content.

Type:

str

specific_points

List of specific weaknesses or points raised.

Type:

list[str]

critique_id: str

target_position_id: str

persona_id: str

round_number: int

content: str

specific_points: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.PersonaDebateSession(*, debate_id: str, topic: str, max_rounds: Annotated[int, Ge(ge=1)] = 5, current_round: Annotated[int, Ge(ge=0)] = 0, status: Literal['active', 'completed', 'resolved'] = 'active', resolution: str | None = None, winner_persona_id: str | None = None, resolution_type: Literal['synthesis', 'winner', 'consensus'] | None = None)

Bases: BaseModel

Complete persona debate session.

Tracks the overall state of a debate between personas, including topic, personas, rounds, positions, critiques, agreements, and resolution.

debate_id

Unique identifier for this debate session.

Type:

str

topic

The debate topic/question.

Type:

str

max_rounds

Maximum number of debate rounds.

Type:

int

current_round

Current round number (0-indexed).

Type:

int

status

Current status of the debate (active, completed, resolved).

Type:

Literal[‘active’, ‘completed’, ‘resolved’]

resolution

Final resolution text (if resolved).

Type:

str | None

winner_persona_id

ID of winning persona (if resolved with winner).

Type:

str | None

resolution_type

Type of resolution (synthesis, winner, consensus).

Type:

Literal[‘synthesis’, ‘winner’, ‘consensus’] | None

debate_id: str

topic: str

max_rounds: int

current_round: int

status: Literal['active', 'completed', 'resolved']

resolution: str | None

winner_persona_id: str | None

resolution_type: Literal['synthesis', 'winner', 'consensus'] | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.PersonaDebateStorage(*, track_usage: bool = False)

Bases: object

Default in-memory persona debate storage.

Simple implementation that stores persona debate sessions, personas, positions, critiques, and agreements in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_persona_debate_toolset, PersonaDebateStorage

storage = PersonaDebateStorage()
toolset = create_persona_debate_toolset(storage=storage)

# After agent runs, access debate state directly
print(storage.session)
print(storage.personas)
print(storage.positions)
print(storage.critiques)
print(storage.agreements)

# With metrics tracking
storage = PersonaDebateStorage(track_usage=True)
toolset = create_persona_debate_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property session: PersonaDebateSession | None

Get the current persona debate session.

property personas: dict[str, Persona]

Get all personas (persona_id -> Persona).

property positions: dict[str, PersonaPosition]

Get all positions (position_id -> PersonaPosition).

property critiques: dict[str, PersonaCritique]

Get all critiques (critique_id -> PersonaCritique).

property agreements: dict[str, PersonaAgreement]

Get all agreements (agreement_id -> PersonaAgreement).

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about persona debate operations.

Returns:

Dictionary with debate counts and metrics.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all debate data and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (persona_id, position_id, critique_id, agreement_id, or session_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of personas, positions, critiques, and agreements.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for personas, positions, critiques, agreements, and session.

class pydantic_ai_toolsets.PersonaDebateStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for persona debate storage implementations.

Any class that has session, personas, positions, critiques, and agreements properties can be used as storage for the persona debate toolset.

class MyCustomStorage:
    def __init__(self):
        self._session: PersonaDebateSession | None = None
        self._personas: dict[str, Persona] = {}
        self._positions: dict[str, PersonaPosition] = {}
        self._critiques: dict[str, PersonaCritique] = {}
        self._agreements: dict[str, PersonaAgreement] = {}

    @property
    def session(self) -> PersonaDebateSession | None:
        return self._session

    @session.setter
    def session(self, value: PersonaDebateSession) -> None:
        self._session = value

    @property
    def personas(self) -> dict[str, Persona]:
        return self._personas

    @personas.setter
    def personas(self, value: Persona) -> None:
        self._personas[value.persona_id] = value

    @property
    def positions(self) -> dict[str, PersonaPosition]:
        return self._positions

    @positions.setter
    def positions(self, value: PersonaPosition) -> None:
        self._positions[value.position_id] = value

    @property
    def critiques(self) -> dict[str, PersonaCritique]:
        return self._critiques

    @critiques.setter
    def critiques(self, value: PersonaCritique) -> None:
        self._critiques[value.critique_id] = value

    @property
    def agreements(self) -> dict[str, PersonaAgreement]:
        return self._agreements

    @agreements.setter
    def agreements(self, value: PersonaAgreement) -> None:
        self._agreements[value.agreement_id] = value

property session: PersonaDebateSession | None

Get the current persona debate session.

property personas: dict[str, Persona]

Get all personas (persona_id -> Persona).

property positions: dict[str, PersonaPosition]

Get all positions (position_id -> PersonaPosition).

property critiques: dict[str, PersonaCritique]

Get all critiques (critique_id -> PersonaCritique).

property agreements: dict[str, PersonaAgreement]

Get all agreements (agreement_id -> PersonaAgreement).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (persona_id, position_id, critique_id, or agreement_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.PersonaPosition(*, position_id: str, persona_id: str, round_number: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], content: str, evidence: list[str] = <factory>, critiques_addressed: list[str] = <factory>, parent_position_id: str | None = None)

Bases: BaseModel

A position/argument made by a persona in a debate.

Each position represents an argument made by a specific persona in a specific round. Positions can be critiqued, defended, and agreed with by other personas.

position_id

Unique identifier for this position.

Type:

str

persona_id

ID of the persona making this position.

Type:

str

round_number

The debate round in which this position was made.

Type:

int

content

The actual argument content.

Type:

str

evidence

List of evidence citations (for evidence-based debates).

Type:

list[str]

critiques_addressed

List of critique IDs that this position addresses.

Type:

list[str]

parent_position_id

ID of parent position if this is a defense/refinement.

Type:

str | None

position_id: str

persona_id: str

round_number: int

content: str

evidence: list[str]

critiques_addressed: list[str]

parent_position_id: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.PersonaResponse(*, response_id: str, persona_id: str, content: str, references: list[str] = <factory>, round_number: Annotated[int, ~annotated_types.Ge(ge=0)] = 0)

Bases: BaseModel

A response from a persona to a problem or question.

Each persona provides independent analysis from their unique perspective. Responses can reference other responses in interactive dialogue patterns.

response_id

Unique identifier for this response.

Type:

str

persona_id

ID of the persona providing this response.

Type:

str

content

The persona’s analysis, insights, or perspective.

Type:

str

references

List of response IDs that this response references or responds to.

Type:

list[str]

round_number

Round number in interactive dialogue (0 for initial responses).

Type:

int

response_id: str

persona_id: str

content: str

references: list[str]

round_number: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.PersonaSession(*, session_id: str, problem: str, process_type: Literal['sequential', 'interactive', 'devils_advocate'], status: Literal['active', 'completed', 'synthesized'] = 'active', synthesis: str | None = None, max_rounds: Annotated[int, Ge(ge=1)] = 3, current_round: Annotated[int, Ge(ge=0)] = 0)

Bases: BaseModel

Complete persona session.

Tracks the overall state of a multi-persona analysis session, including the problem/question, personas, responses, and synthesis.

session_id

Unique identifier for this session.

Type:

str

problem

The problem or question being analyzed.

Type:

str

process_type

Type of process - sequential, interactive, or devils_advocate.

Type:

Literal[‘sequential’, ‘interactive’, ‘devils_advocate’]

status

Current status of the session (active, completed, synthesized).

Type:

Literal[‘active’, ‘completed’, ‘synthesized’]

synthesis

Final synthesis text (if synthesized).

Type:

str | None

max_rounds

Maximum number of dialogue rounds (for interactive/devils_advocate).

Type:

int

current_round

Current round number (0-indexed).

Type:

int

session_id: str

problem: str

process_type: Literal['sequential', 'interactive', 'devils_advocate']

status: Literal['active', 'completed', 'synthesized']

synthesis: str | None

max_rounds: int

current_round: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.PersonaStorage(*, track_usage: bool = False)

Bases: object

Default in-memory persona storage.

Simple implementation that stores persona sessions, personas, and responses in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_persona_toolset, PersonaStorage

storage = PersonaStorage()
toolset = create_persona_toolset(storage=storage)

# After agent runs, access persona state directly
print(storage.session)
print(storage.personas)
print(storage.responses)

# With metrics tracking
storage = PersonaStorage(track_usage=True)
toolset = create_persona_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property session: PersonaSession | None

Get the current persona session.

property personas: dict[str, Persona]

Get all personas (persona_id -> Persona).

property responses: dict[str, PersonaResponse]

Get all responses (response_id -> PersonaResponse).

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about persona operations.

Returns:

Dictionary with persona and response counts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all persona data and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (persona_id, response_id, or session_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of personas, responses, and session.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for personas, responses, and session.

class pydantic_ai_toolsets.PersonaStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for persona storage implementations.

Any class that has session, personas, and responses properties can be used as storage for the persona toolset.

class MyCustomStorage:
    def __init__(self):
        self._session: PersonaSession | None = None
        self._personas: dict[str, Persona] = {}
        self._responses: dict[str, PersonaResponse] = {}

    @property
    def session(self) -> PersonaSession | None:
        return self._session

    @session.setter
    def session(self, value: PersonaSession) -> None:
        self._session = value

    @property
    def personas(self) -> dict[str, Persona]:
        return self._personas

    @personas.setter
    def personas(self, value: Persona) -> None:
        self._personas[value.persona_id] = value

    @property
    def responses(self) -> dict[str, PersonaResponse]:
        return self._responses

    @responses.setter
    def responses(self, value: PersonaResponse) -> None:
        self._responses[value.response_id] = value

property session: PersonaSession | None

Get the current persona session.

property personas: dict[str, Persona]

Get all personas (persona_id -> Persona).

property responses: dict[str, PersonaResponse]

Get all responses (response_id -> PersonaResponse).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (persona_id, response_id, or session_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.ProposePositionItem(*, persona_id: str, content: str, evidence: list[str] = <factory>)

Bases: BaseModel

Input model for the propose_position tool.

persona_id: str

content: str

evidence: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ProvideFeedbackItem(**data: Any)

Bases: BaseModel

Input model for the provide_feedback tool.

output_id: str

feedback_items: list[FeedbackItem]

overall_assessment: str

should_continue_refining: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.PruneBeamItem(*, step_index: Annotated[int, Ge(ge=0)], beam_width: Annotated[int, Ge(ge=1)])

Bases: BaseModel

Input model for the prune_beam tool.

step_index: int

beam_width: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.Question(*, question_id: str, question_text: str, is_main: bool = False, parent_question_id: str | None = None, depth: Annotated[int, Ge(ge=0), Le(le=3)] = 0, status: QuestionStatus = QuestionStatus.PENDING)

Bases: BaseModel

A question being decomposed in the self-ask process.

Questions form a tree structure where the main question is the root (depth 0), and sub-questions are children (depth 1-3). Each question can spawn sub-questions up to the maximum depth of 3.

question_id

Unique identifier for this question.

Type:

str

question_text

The actual question text.

Type:

str

is_main

Whether this is the main question (depth 0).

Type:

bool

parent_question_id

ID of the parent question (None for main question).

Type:

str | None

depth

Depth level in the question tree (0 for main, 1-3 for sub-questions).

Type:

int

status

Current status of the question (pending, answered, composed).

Type:

pydantic_ai_toolsets.toolsets.self_ask.types.QuestionStatus

question_id: str

question_text: str

is_main: bool

parent_question_id: str | None

depth: int

status: QuestionStatus

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.QuestionStatus(*values)

Bases: str, Enum

Status of a question in the self-ask process.

PENDING = 'pending'

ANSWERED = 'answered'

COMPOSED = 'composed'

class pydantic_ai_toolsets.RefineItem(*, node_id: str, refined_content: str, is_solution: bool = False)

Bases: BaseModel

Input model for the refine_node tool.

node_id: str

refined_content: str

is_solution: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.RefineOutputItem(*, output_id: str, refined_content: str, is_final: bool = False, quality_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None)

Bases: BaseModel

Input model for the refine_output tool.

output_id: str

refined_content: str

is_final: bool

quality_score: float | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.RefinementOutput(*, output_id: str, content: str, iteration: Annotated[int, Ge(ge=0)] = 0, parent_id: str | None = None, is_final: bool = False, quality_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, quality_threshold: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, iteration_limit: Annotated[int | None, Ge(ge=1)] = None)

Bases: BaseModel

An output that can be refined through feedback loops.

Each output represents a version of the solution at a specific refinement iteration. Outputs form a chain where each output can receive feedback and be refined into a new output.

output_id

Unique identifier for this output.

Type:

str

content

The actual output content (solution, answer, etc.).

Type:

str

iteration

The refinement iteration number (0 for initial output).

Type:

int

parent_id

ID of the parent output that was refined to create this one (None for initial).

Type:

str | None

is_final

Whether this output is marked as final/satisfactory.

Type:

bool

quality_score

Optional quality score (0-100, higher is better).

Type:

float | None

quality_threshold

Optional quality threshold that must be met (0-100).

Type:

float | None

iteration_limit

Optional maximum number of iterations allowed.

Type:

int | None

output_id: str

content: str

iteration: int

parent_id: str | None

is_final: bool

quality_score: float | None

quality_threshold: float | None

iteration_limit: int | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ReflectionOutput(*, output_id: str, content: str, cycle: Annotated[int, Ge(ge=0)] = 0, parent_id: str | None = None, is_final: bool = False, quality_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None)

Bases: BaseModel

An output that can be critiqued and refined.

Each output represents a version of the solution at a specific refinement cycle. Outputs form a chain where each output can be critiqued and refined into a new output.

output_id

Unique identifier for this output.

Type:

str

content

The actual output content (solution, answer, etc.).

Type:

str

cycle

The refinement cycle number (0 for initial output).

Type:

int

parent_id

ID of the parent output that was refined to create this one (None for initial).

Type:

str | None

is_final

Whether this output is marked as final/satisfactory.

Type:

bool

quality_score

Optional quality score (0-100, higher is better).

Type:

float | None

output_id: str

content: str

cycle: int

parent_id: str | None

is_final: bool

quality_score: float | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ReflectionStorage(*, track_usage: bool = False)

Bases: object

Default in-memory reflection storage.

Simple implementation that stores outputs and critiques in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_reflection_toolset, ReflectionStorage

storage = ReflectionStorage()
toolset = create_reflection_toolset(storage=storage)

# After agent runs, access outputs and critiques directly
print(storage.outputs)
print(storage.critiques)

# With metrics tracking
storage = ReflectionStorage(track_usage=True)
toolset = create_reflection_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property outputs: dict[str, ReflectionOutput]

Get the current dictionary of outputs.

property critiques: dict[str, Critique]

Get the current dictionary of critiques.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about reflection operations.

Returns:

Dictionary with output and critique counts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all outputs, critiques, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (output_id or critique_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of outputs and critiques.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for outputs and critiques.

class pydantic_ai_toolsets.ReflectionStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for reflection storage implementations.

Any class that has outputs and critiques properties can be used as storage for the reflection toolset.

class MyCustomStorage:
    def __init__(self):
        self._outputs: dict[str, ReflectionOutput] = {}
        self._critiques: dict[str, Critique] = {}

    @property
    def outputs(self) -> dict[str, ReflectionOutput]:
        return self._outputs

    @outputs.setter
    def outputs(self, value: ReflectionOutput) -> None:
        self._outputs[value.output_id] = value

    @property
    def critiques(self) -> dict[str, Critique]:
        return self._critiques

    @critiques.setter
    def critiques(self, value: Critique) -> None:
        self._critiques[value.critique_id] = value

property outputs: dict[str, ReflectionOutput]

Get the current dictionary of outputs (output_id -> ReflectionOutput).

property critiques: dict[str, Critique]

Get the current dictionary of critiques (critique_id -> Critique).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (output_id or critique_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.ResolveDebateItem(*, resolution_type: ~typing.Literal['synthesis', 'winner', 'consensus'], resolution_content: str, winner_persona_id: str | None = None, synthesis_elements: list[str] = <factory>, consensus_points: list[str] = <factory>)

Bases: BaseModel

Input model for the resolve_debate tool.

resolution_type: Literal['synthesis', 'winner', 'consensus']

resolution_content: str

winner_persona_id: str | None

synthesis_elements: list[str]

consensus_points: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ScoreCandidateItem(*, candidate_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str)

Bases: BaseModel

Input model for the score_candidate tool.

candidate_id: str

score: float

reasoning: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.SearchResult(*, result_id: str, query: str, title: str, url: str, description: str | None = None, timestamp: str | None = None, source_type: SearchSource = SearchSource.WEB, date: str | None = None, image_url: str | None = None, image_width: int | None = None, image_height: int | None = None)

Bases: BaseModel

A single search result from a web search.

result_id

Unique identifier for this search result.

Type:

str

query

The search query that produced this result.

Type:

str

title

Title of the search result.

Type:

str

url

URL of the search result.

Type:

str

description

Description or snippet of the search result.

Type:

str | None

timestamp

Timestamp when the search was performed.

Type:

str | None

source_type

Type of search source (web, news, or images).

Type:

pydantic_ai_toolsets.toolsets.search.types.SearchSource

date

Date string for news results (e.g., “3 months ago”).

Type:

str | None

image_url

Direct image URL for image results.

Type:

str | None

image_width

Image width in pixels for image results.

Type:

int | None

image_height

Image height in pixels for image results.

Type:

int | None

result_id: str

query: str

title: str

url: str

description: str | None

timestamp: str | None

source_type: SearchSource

date: str | None

image_url: str | None

image_width: int | None

image_height: int | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.SearchStorage(*, track_usage: bool = False)

Bases: object

Default in-memory search storage.

Simple implementation that stores search results and extracted contents in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_search_toolset, SearchStorage

storage = SearchStorage()
toolset = create_search_toolset(storage=storage)

# After agent runs, access search results and extracted contents directly
print(storage.search_results)
print(storage.extracted_contents)

# With metrics tracking
storage = SearchStorage(track_usage=True)
toolset = create_search_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property search_results: dict[str, SearchResult]

Get the current dictionary of search results.

property extracted_contents: dict[str, ExtractedContent]

Get the current dictionary of extracted contents.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about search operations.

Returns:

Dictionary with search and extraction counts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all search results, extracted contents, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (search_result_id or extracted_content_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of search results and extracted contents.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for search results and extracted contents.

class pydantic_ai_toolsets.SearchStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for search storage implementations.

Any class that has search_results and extracted_contents properties can be used as storage for the search toolset.

class MyCustomStorage:
    def __init__(self):
        self._search_results: dict[str, SearchResult] = {}
        self._extracted_contents: dict[str, ExtractedContent] = {}

    @property
    def search_results(self) -> dict[str, SearchResult]:
        return self._search_results

    @search_results.setter
    def search_results(self, value: SearchResult) -> None:
        self._search_results[value.result_id] = value

    @property
    def extracted_contents(self) -> dict[str, ExtractedContent]:
        return self._extracted_contents

    @extracted_contents.setter
    def extracted_contents(self, value: ExtractedContent) -> None:
        self._extracted_contents[value.content_id] = value

property search_results: dict[str, SearchResult]

Get the current dictionary of search results (result_id -> SearchResult).

property extracted_contents: dict[str, ExtractedContent]

Get the current dictionary of extracted contents (content_id -> ExtractedContent).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (result_id or content_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.SearchWebItem(*, query: str, limit: Annotated[int, Ge(ge=1), Le(le=5)] = 5)

Bases: BaseModel

Input model for the search_web tool.

This is the model that agents use when calling search_web.

query: str

limit: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.SelectNodeItem(*, node_id: str | None = None, exploration_constant: Annotated[float, Ge(ge=0)] = 1.414)

Bases: BaseModel

Input model for the select_node tool.

Allows manual selection of a node, or automatic selection using UCB1.

node_id: str | None

exploration_constant: float

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.SelfAskStorage(*, track_usage: bool = False)

Bases: object

Default in-memory self-ask storage.

Simple implementation that stores questions, answers, and final answers in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_self_ask_toolset, SelfAskStorage

storage = SelfAskStorage()
toolset = create_self_ask_toolset(storage=storage)

# After agent runs, access questions, answers, and final answers directly
print(storage.questions)
print(storage.answers)
print(storage.final_answers)

# With metrics tracking
storage = SelfAskStorage(track_usage=True)
toolset = create_self_ask_toolset(storage=storage)
print(storage.metrics.total_tokens())

MAX_DEPTH: int = 3

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property questions: dict[str, Question]

Get the current dictionary of questions.

property answers: dict[str, Answer]

Get the current dictionary of answers.

property final_answers: dict[str, FinalAnswer]

Get the current dictionary of final answers.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about self-ask operations.

Returns:

Dictionary with question, answer, and final answer counts, plus max depth reached.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all questions, answers, final answers, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (question_id, answer_id, or final_answer_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of questions, answers, and final answers.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for questions, answers, and final answers.

class pydantic_ai_toolsets.SelfAskStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for self-ask storage implementations.

Any class that has questions, answers, and final_answers properties can be used as storage for the self-ask toolset.

class MyCustomStorage:
    def __init__(self):
        self._questions: dict[str, Question] = {}
        self._answers: dict[str, Answer] = {}
        self._final_answers: dict[str, FinalAnswer] = {}

    @property
    def questions(self) -> dict[str, Question]:
        return self._questions

    @questions.setter
    def questions(self, value: Question) -> None:
        self._questions[value.question_id] = value

    @property
    def answers(self) -> dict[str, Answer]:
        return self._answers

    @answers.setter
    def answers(self, value: Answer) -> None:
        self._answers[value.answer_id] = value

    @property
    def final_answers(self) -> dict[str, FinalAnswer]:
        return self._final_answers

    @final_answers.setter
    def final_answers(self, value: FinalAnswer) -> None:
        self._final_answers[value.final_answer_id] = value

property questions: dict[str, Question]

Get the current dictionary of questions (question_id -> Question).

property answers: dict[str, Answer]

Get the current dictionary of answers (answer_id -> Answer).

property final_answers: dict[str, FinalAnswer]

Get the current dictionary of final answers (final_answer_id -> FinalAnswer).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (question_id, answer_id, or final_answer_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.SelfRefineStorage(*, track_usage: bool = False)

Bases: object

Default in-memory self-refinement storage.

Simple implementation that stores outputs and feedbacks in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_self_refine_toolset, SelfRefineStorage

storage = SelfRefineStorage()
toolset = create_self_refine_toolset(storage=storage)

# After agent runs, access outputs and feedbacks directly
print(storage.outputs)
print(storage.feedbacks)

# With metrics tracking
storage = SelfRefineStorage(track_usage=True)
toolset = create_self_refine_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property outputs: dict[str, RefinementOutput]

Get the current dictionary of outputs.

property feedbacks: dict[str, Feedback]

Get the current dictionary of feedbacks.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about self-refinement operations.

Returns:

Dictionary with output and feedback counts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all outputs, feedbacks, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (output_id or feedback_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of outputs and feedbacks.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for outputs and feedbacks.

class pydantic_ai_toolsets.SelfRefineStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for self-refinement storage implementations.

Any class that has outputs and feedbacks properties can be used as storage for the self-refinement toolset.

class MyCustomStorage:
    def __init__(self):
        self._outputs: dict[str, RefinementOutput] = {}
        self._feedbacks: dict[str, Feedback] = {}

    @property
    def outputs(self) -> dict[str, RefinementOutput]:
        return self._outputs

    @outputs.setter
    def outputs(self, value: RefinementOutput) -> None:
        self._outputs[value.output_id] = value

    @property
    def feedbacks(self) -> dict[str, Feedback]:
        return self._feedbacks

    @feedbacks.setter
    def feedbacks(self, value: Feedback) -> None:
        self._feedbacks[value.feedback_id] = value

property outputs: dict[str, RefinementOutput]

Get the current dictionary of outputs (output_id -> RefinementOutput).

property feedbacks: dict[str, Feedback]

Get the current dictionary of feedbacks (feedback_id -> Feedback).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (output_id or feedback_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.SimulateItem(*, node_id: str, simulation_result: ~typing.Annotated[float, ~annotated_types.Ge(ge=0.0), ~annotated_types.Le(le=1.0)], simulation_path: list[str] = <factory>)

Bases: BaseModel

Input model for the simulate tool.

node_id: str

simulation_result: float

simulation_path: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.SynthesizeItem(*, synthesis_content: str, key_insights: list[str] = <factory>, conflicts_resolved: list[str] = <factory>)

Bases: BaseModel

Input model for the synthesize tool.

synthesis_content: str

key_insights: list[str]

conflicts_resolved: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.Thought(*, thought: str, thought_number: int, total_thoughts: int, is_revision: bool = False, revises_thought: int | None = None, branch_from_thought: int | None = None, branch_id: str | None = None, next_thought_needed: bool = True)

Bases: BaseModel

A thought in a chain of thoughts execution.

thought

The current thinking step content.

Type:

str

thought_number

Current number in sequence (1-based).

Type:

int

total_thoughts

Estimated total thoughts needed (can be adjusted).

Type:

int

is_revision

Whether this thought revises previous thinking.

Type:

bool

revises_thought

Which thought number is being reconsidered (if is_revision).

Type:

int | None

branch_from_thought

Branching point thought number (if branching).

Type:

int | None

branch_id

Identifier for the current branch (if branching).

Type:

str | None

next_thought_needed

Whether another thought step is needed.

Type:

bool

thought: str

thought_number: int

total_thoughts: int

is_revision: bool

revises_thought: int | None

branch_from_thought: int | None

branch_id: str | None

next_thought_needed: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ThoughtItem(*, thought: str, thought_number: int, total_thoughts: int, is_revision: bool = False, revises_thought: int | None = None, branch_from_thought: int | None = None, branch_id: str | None = None, next_thought_needed: bool = True)

Bases: BaseModel

Input model for the write_thoughts tool.

This is the model that agents use when calling write_thoughts. It has the same fields as Thought but with Field descriptions for LLM guidance.

thought: str

thought_number: int

total_thoughts: int

is_revision: bool

revises_thought: int | None

branch_from_thought: int | None

branch_id: str | None

next_thought_needed: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ThoughtNode(*, node_id: str, content: str, parent_id: str | None = None, depth: int = 0, branch_id: str | None = None, status: Literal['active', 'pruned', 'merged', 'completed']='active', evaluation_score: float | None = None, is_solution: bool = False, merged_from: list[str] = <factory>)

Bases: BaseModel

A node in a tree of thoughts execution.

Each node represents a reasoning state. Nodes form a tree structure where multiple branches can be explored simultaneously.

node_id

Unique identifier for this node.

Type:

str

content

The reasoning content at this node.

Type:

str

parent_id

ID of the parent node (None for root nodes).

Type:

str | None

depth

Depth level in the tree (0 for root).

Type:

int

branch_id

Identifier for the branch this node belongs to.

Type:

str | None

status

Current status of this node/branch.

Type:

Literal[‘active’, ‘pruned’, ‘merged’, ‘completed’]

evaluation_score

Optional score evaluating this path (0-100).

Type:

float | None

is_solution

Whether this node represents a solution.

Type:

bool

merged_from

List of node IDs whose insights were merged into this node.

Type:

list[str]

node_id: str

content: str

parent_id: str | None

depth: int

branch_id: str | None

status: Literal['active', 'pruned', 'merged', 'completed']

evaluation_score: float | None

is_solution: bool

merged_from: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.ToTStorage(*, track_usage: bool = False)

Bases: object

Default in-memory tree of thoughts storage.

Simple implementation that stores nodes and evaluations in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_tot_toolset, ToTStorage

storage = ToTStorage()
toolset = create_tot_toolset(storage=storage)

# After agent runs, access nodes and evaluations directly
print(storage.nodes)
print(storage.evaluations)

# With metrics tracking
storage = ToTStorage(track_usage=True)
toolset = create_tot_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property nodes: dict[str, ThoughtNode]

Get the current dictionary of nodes.

property evaluations: dict[str, BranchEvaluation]

Get the current dictionary of branch evaluations.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about the tree.

Returns:

Dictionary with node counts and tree metrics.

depth_statistics() → dict[int, int]

Get node count at each depth level.

Returns:

Dictionary mapping depth to node count.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all nodes, evaluations, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id or evaluation_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of nodes and evaluations.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for nodes and evaluations.

class pydantic_ai_toolsets.ToTStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for tree of thoughts storage implementations.

Any class that has nodes and evaluations properties can be used as storage for the ToT toolset.

class MyCustomStorage:
    def __init__(self):
        self._nodes: dict[str, ThoughtNode] = {}
        self._evaluations: dict[str, BranchEvaluation] = {}

    @property
    def nodes(self) -> dict[str, ThoughtNode]:
        return self._nodes

    @nodes.setter
    def nodes(self, value: ThoughtNode) -> None:
        self._nodes[value.node_id] = value

    @property
    def evaluations(self) -> dict[str, BranchEvaluation]:
        return self._evaluations

    @evaluations.setter
    def evaluations(self, value: BranchEvaluation) -> None:
        self._evaluations[value.branch_id] = value

property nodes: dict[str, ThoughtNode]

Get the current dictionary of nodes (node_id -> ThoughtNode).

property evaluations: dict[str, BranchEvaluation]

Get the current dictionary of branch evaluations (branch_id -> BranchEvaluation).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id or evaluation_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

class pydantic_ai_toolsets.Todo(*, todo_id: str, content: str, status: Literal['pending', 'in_progress', 'completed'], active_form: str)

Bases: BaseModel

A todo item for task tracking.

todo_id

Unique identifier for this todo item.

Type:

str

content

The task description in imperative form (e.g., ‘Implement feature X’).

Type:

str

status

Task status - ‘pending’, ‘in_progress’, or ‘completed’.

Type:

Literal[‘pending’, ‘in_progress’, ‘completed’]

active_form

Present continuous form shown during execution (e.g., ‘Implementing feature X’).

Type:

str

todo_id: str

content: str

status: Literal['pending', 'in_progress', 'completed']

active_form: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.TodoItem(*, content: str, status: Literal['pending', 'in_progress', 'completed'], active_form: str)

Bases: BaseModel

Input model for the write_todos tool.

This is the model that agents use when calling write_todos. It has the same fields as Todo but with Field descriptions for LLM guidance.

content: str

status: Literal['pending', 'in_progress', 'completed']

active_form: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.TodoStorage(*, track_usage: bool = False)

Bases: object

Default in-memory todo storage.

Simple implementation that stores todos in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_todo_toolset, TodoStorage

storage = TodoStorage()
toolset = create_todo_toolset(storage=storage)

# After agent runs, access todos directly
print(storage.todos)

# With metrics tracking
storage = TodoStorage(track_usage=True)
toolset = create_todo_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property todos: list[Todo]

Get the current list of todos.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about todos.

Returns:

Dictionary with todo counts and completion metrics.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all todos and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the todo item

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of todos.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for todos.

class pydantic_ai_toolsets.TodoStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for todo storage implementations.

Any class that has a todos property (read/write) implementing list[Todo] can be used as storage for the todo toolset.

class MyCustomStorage:
    def __init__(self):
        self._todos: list[Todo] = []

    @property
    def todos(self) -> list[Todo]:
        return self._todos

    @todos.setter
    def todos(self, value: list[Todo]) -> None:
        self._todos = value

property todos: list[Todo]

Get the current list of todos.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (todo_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

pydantic_ai_toolsets.calculate_ucb1(node: MCTSNode, parent_visits: int, exploration_constant: float) → float

Calculate UCB1 value for a node.

UCB1 = win_rate + c * sqrt(ln(parent_visits) / visits)

Parameters:

• node – The node to calculate UCB1 for.

• parent_visits – Total visits of the parent node.

• exploration_constant – The exploration constant (c).

Returns:

UCB1 value, or infinity if node hasn’t been visited.

pydantic_ai_toolsets.create_beam_toolset(storage: BeamStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a beam search toolset for beam-based reasoning exploration.

This toolset provides tools for AI agents to explore reasoning using beam search, maintaining a beam of top-k candidates at each step.

Parameters:

• storage – Optional storage backend. Defaults to in-memory BeamStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_beam_toolset, BeamStorage

# With storage and metrics
storage = BeamStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_beam_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.create_cot_toolset(storage: CoTStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a chain of thoughts toolset for reasoning exploration.

This toolset provides read_thoughts and write_thoughts tools for AI agents to document and explore their reasoning process during a session.

Parameters:

• storage – Optional storage backend. Defaults to in-memory CoTStorage. You can provide a custom storage implementing CoTStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection on the default storage. Ignored if custom storage is provided.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_cot_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_cot_toolset()])
result = await agent.run("Solve this complex problem step by step")

Example (with custom storage):

from pydantic_ai_toolsets import create_cot_toolset, CoTStorage

storage = CoTStorage()
toolset = create_cot_toolset(storage=storage)

# After agent runs, access thoughts directly
print(storage.thoughts)

Example (with usage tracking):

from pydantic_ai_toolsets import create_cot_toolset, CoTStorage

storage = CoTStorage(track_usage=True)
toolset = create_cot_toolset(storage=storage)

# After agent runs, check usage metrics
print(storage.metrics.total_tokens())
print(storage.metrics.invocation_count())

pydantic_ai_toolsets.create_got_toolset(storage: GoTStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a graph of thoughts toolset for graph-based reasoning.

This toolset provides tools for AI agents to explore reasoning using a directed graph structure with nodes (reasoning states) and edges (connections/dependencies).

Parameters:

• storage – Optional storage backend. Defaults to in-memory GoTStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_got_toolset, GoTStorage

# With storage and metrics
storage = GoTStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_got_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.create_mcts_toolset(storage: MCTSStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create an MCTS toolset for tree-based exploration with statistics.

This toolset provides tools for AI agents to explore reasoning using Monte Carlo Tree Search, balancing exploration and exploitation.

Parameters:

• storage – Optional storage backend. Defaults to in-memory MCTSStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_mcts_toolset, MCTSStorage

# With storage and metrics
storage = MCTSStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_mcts_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.create_persona_debate_toolset(storage: PersonaDebateStorageProtocol | None = None, *, id: str | None = None, agent_model: str | None = None, agent_configs: dict[str, dict[str, Any]] | None = None, auto_orchestrate: bool = False, track_usage: bool = False) → FunctionToolset[Any]

Create a persona debate toolset for multi-persona structured debates.

This toolset provides tools for AI agents to engage in structured debates between multiple personas, with support for creating and orchestrating multiple agent instances.

Parameters:

• storage – Optional storage backend. Defaults to in-memory PersonaDebateStorage. You can provide a custom storage implementing PersonaDebateStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• agent_model – Default model string for creating agents (e.g., “openai:gpt-4”). Required if agent_configs not provided or if auto_orchestrate=True.

• agent_configs –

  Per-persona agent configurations: {

  ”persona_id_1”: {“model”: “openai:gpt-4”, “system_prompt”: “…”}, “persona_id_2”: {“model”: “openai:gpt-4”, “system_prompt”: “…”},

  }

• auto_orchestrate – If True, tools automatically orchestrate agent interactions.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_persona_debate_toolset

agent = Agent("openai:gpt-4", toolsets=[create_persona_debate_toolset()])
result = await agent.run("Debate: Should we adopt microservices?")

Example (with multi-agent orchestration):

from pydantic_ai_toolsets import create_persona_debate_toolset, PersonaDebateStorage

storage = PersonaDebateStorage()
toolset = create_persona_debate_toolset(
    storage=storage,
    agent_model="openai:gpt-4",
    auto_orchestrate=True,
)

orchestrator = Agent("openai:gpt-4", toolsets=[toolset])
result = await orchestrator.run("Start a debate on microservices")

pydantic_ai_toolsets.create_persona_toolset(storage: PersonaStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a multi-persona toolset for diverse perspective analysis.

This toolset provides tools for AI agents to adopt multiple personas and synthesize diverse perspectives on problems.

Parameters:

• storage – Optional storage backend. Defaults to in-memory PersonaStorage. You can provide a custom storage implementing PersonaStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection in storage.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_persona_toolset

agent = Agent("openai:gpt-4", toolsets=[create_persona_toolset()])
result = await agent.run("Analyze: Should we invest in this startup?")

Example (with storage access):

from pydantic_ai_toolsets import create_persona_toolset, PersonaStorage

storage = PersonaStorage()
toolset = create_persona_toolset(storage=storage)

# After agent runs, access persona state directly
print(storage.session)
print(storage.personas)
print(storage.responses)

# With metrics tracking
storage = PersonaStorage(track_usage=True)
toolset = create_persona_toolset(storage=storage)
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.create_reflection_toolset(storage: ReflectionStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a reflection toolset for iterative output improvement.

This toolset provides tools for AI agents to improve outputs through critical analysis and refinement cycles.

Parameters:

• storage – Optional storage backend. Defaults to in-memory ReflectionStorage. You can provide a custom storage implementing ReflectionStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_reflection_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_reflection_toolset()])
result = await agent.run("Solve this problem using reflection")

Example (with custom storage):

from pydantic_ai_toolsets import create_reflection_toolset, ReflectionStorage

storage = ReflectionStorage()
toolset = create_reflection_toolset(storage=storage)

# After agent runs, access outputs and critiques directly
print(storage.outputs)
print(storage.critiques)

pydantic_ai_toolsets.create_search_toolset(storage: SearchStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a search toolset for web search and content extraction.

This toolset provides tools for AI agents to search the web and extract content from webpages, with support for tracking search history and extracted content.

Parameters:

• storage – Optional storage backend. Defaults to in-memory SearchStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_search_toolset, SearchStorage

# With storage and metrics
storage = SearchStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_search_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.create_self_ask_toolset(storage: SelfAskStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a self-ask toolset for question decomposition.

This toolset provides tools for AI agents to decompose complex questions into simpler sub-questions, answer them sequentially, and compose final answers.

Parameters:

• storage – Optional storage backend. Defaults to in-memory SelfAskStorage. You can provide a custom storage implementing SelfAskStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection in storage.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_self_ask_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_self_ask_toolset()])
result = await agent.run("What was the population of the city where the 2016 Summer Olympics were held?")

Example (with custom storage):

from pydantic_ai_toolsets import create_self_ask_toolset, SelfAskStorage

storage = SelfAskStorage()
toolset = create_self_ask_toolset(storage=storage)

# After agent runs, access questions, answers, and final answers directly
print(storage.questions)
print(storage.answers)
print(storage.final_answers)

pydantic_ai_toolsets.create_self_refine_toolset(storage: SelfRefineStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a self-refinement toolset for iterative output improvement.

This toolset provides tools for AI agents to improve outputs through structured feedback and refinement cycles, with support for quality thresholds and iteration limits.

Parameters:

• storage – Optional storage backend. Defaults to in-memory SelfRefineStorage. You can provide a custom storage implementing SelfRefineStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_self_refine_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_self_refine_toolset()])
result = await agent.run("Solve this problem using self-refinement")

Example (with custom storage):

from pydantic_ai_toolsets import create_self_refine_toolset, SelfRefineStorage

storage = SelfRefineStorage()
toolset = create_self_refine_toolset(storage=storage)

# After agent runs, access outputs and feedbacks directly
print(storage.outputs)
print(storage.feedbacks)

pydantic_ai_toolsets.create_todo_toolset(storage: TodoStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a todo toolset for task management.

This toolset provides read_todos and write_todos tools for AI agents to track and manage tasks during a session.

Parameters:

• storage – Optional storage backend. Defaults to in-memory TodoStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_todo_toolset, TodoStorage

# With storage and metrics
storage = TodoStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_todo_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.create_tot_toolset(storage: ToTStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a tree of thoughts toolset for multi-path reasoning exploration.

This toolset provides tools for AI agents to explore multiple reasoning paths simultaneously, evaluate branches, prune dead ends, and merge insights.

Parameters:

• storage – Optional storage backend. Defaults to in-memory ToTStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_tot_toolset, ToTStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_tot_toolset()])

# With storage and metrics
storage = ToTStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_tot_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.get_beam_system_prompt(storage: BeamStorageProtocol | None = None) → str

Generate dynamic system prompt section for beam search.

Parameters:

storage – Optional storage to read current beam from.

Returns:

System prompt section with current beam state, or base prompt if no candidates.

pydantic_ai_toolsets.get_cot_system_prompt(storage: CoTStorageProtocol | None = None) → str

Generate dynamic system prompt section for chain of thoughts.

Parameters:

storage – Optional storage to read current thoughts from.

Returns:

System prompt section with current thoughts, or base prompt if no thoughts.

pydantic_ai_toolsets.get_got_system_prompt(storage: GoTStorageProtocol | None = None) → str

Generate dynamic system prompt section for graph of thoughts.

Parameters:

storage – Optional storage to read current graph from.

Returns:

System prompt section with current graph state, or base prompt if empty.

pydantic_ai_toolsets.get_mcts_system_prompt(storage: MCTSStorageProtocol | None = None) → str

Generate dynamic system prompt section for MCTS.

Parameters:

storage – Optional storage to read current tree from.

Returns:

System prompt section with current tree state, or base prompt if empty.

pydantic_ai_toolsets.get_persona_debate_system_prompt() → str

Get the system prompt for persona debate-based reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

pydantic_ai_toolsets.get_persona_system_prompt(storage: PersonaStorageProtocol | None = None) → str

Generate dynamic system prompt section for personas.

Parameters:

storage – Optional storage to read current session from.

Returns:

System prompt section with current session info, or base prompt if no session.

pydantic_ai_toolsets.get_reflection_system_prompt() → str

Get the system prompt for reflection-based reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

pydantic_ai_toolsets.get_search_system_prompt() → str

Get the system prompt for search-based reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

pydantic_ai_toolsets.get_self_ask_system_prompt() → str

Get the system prompt for self-ask reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

pydantic_ai_toolsets.get_self_refine_system_prompt() → str

Get the system prompt for self-refinement-based reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

pydantic_ai_toolsets.get_todo_system_prompt(storage: TodoStorageProtocol | None = None) → str

Generate dynamic system prompt section for todos.

Parameters:

storage – Optional storage to read current todos from.

Returns:

System prompt section with current todos, or base prompt if no todos.

pydantic_ai_toolsets.get_tot_system_prompt(storage: ToTStorageProtocol | None = None) → str

Generate dynamic system prompt section for tree of thoughts.

Parameters:

storage – Optional storage to read current tree from.

Returns:

System prompt section with current tree state, or base prompt if no nodes.

class pydantic_ai_toolsets.MetaOrchestratorStorage(*, track_usage: bool = False)

Bases: object

Default in-memory meta-orchestrator storage.

Tracks active workflows, registered toolsets, transitions, and cross-toolset links. Use this for standalone agents or testing.

_registered_toolsets

Dictionary mapping toolset IDs to their metadata

Type:

dict[str, dict[str, Any]]

_active_workflows

Dictionary mapping workflow IDs to WorkflowState

Type:

dict[str, WorkflowState]

_links

List of all cross-toolset links

Type:

list[CrossToolsetLink]

_transitions

List of all toolset transitions

Type:

list[ToolsetTransition]

_workflow_registry

Registry of workflow templates

Type:

WorkflowRegistry

_metrics

Optional usage metrics tracker

Type:

UsageMetrics | None

from pydantic_ai_toolsets import create_meta_orchestrator_toolset, MetaOrchestratorStorage

storage = MetaOrchestratorStorage()
toolset = create_meta_orchestrator_toolset(storage=storage)

# After agent runs, access workflow state
workflow = storage.get_active_workflow()
print(workflow.current_stage)
print(storage.links)

# With metrics tracking
storage = MetaOrchestratorStorage(track_usage=True)
toolset = create_meta_orchestrator_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property registered_toolsets: dict[str, dict[str, Any]]

Get dictionary of registered toolsets.

property active_workflows: dict[str, WorkflowState]

Get dictionary of active workflows.

property links: list[CrossToolsetLink]

Get list of all cross-toolset links.

property transitions: list[ToolsetTransition]

Get list of all toolset transitions.

property workflow_registry: WorkflowRegistry

Get the workflow registry.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

register_toolset(toolset_id: str, toolset_info: dict[str, Any]) → None

Register a toolset with the orchestrator.

Parameters:

• toolset_id – Unique identifier for the toolset

• toolset_info – Dictionary with toolset metadata (e.g., type, label, tools)

track_transition(transition: ToolsetTransition) → None

Track a toolset transition and automatically update workflow progression.

Parameters:

transition – The transition to track

create_link(link: CrossToolsetLink) → None

Create a cross-toolset link.

Parameters:

link – The link to create

get_links_for_item(toolset_id: str, item_id: str) → list[CrossToolsetLink]

Get all links for a specific item.

Parameters:

• toolset_id – ID of the toolset

• item_id – ID of the item

Returns:

List of links where this item is source or target

start_workflow(workflow: WorkflowState) → None

Start a new workflow.

Parameters:

workflow – The workflow state to start

get_active_workflow() → WorkflowState | None

Get the currently active workflow.

Returns:

The most recently updated workflow, or None if no workflows exist

update_workflow(workflow_id: str, updates: dict[str, Any]) → None

Update a workflow’s state.

Parameters:

• workflow_id – ID of the workflow to update

• updates – Dictionary of updates to apply

get_unified_state() → dict[str, Any]

Get unified state across all registered toolsets.

Returns:

• active_toolsets: List of registered toolset IDs

• active_workflows: List of active workflow IDs

• total_links: Total number of cross-toolset links

• total_transitions: Total number of transitions

• current_workflow: Current workflow state if any

Return type:

Dictionary containing

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

class pydantic_ai_toolsets.WorkflowTemplate(name: str, toolsets: list[str], stages: list[Stage], handoff_instructions: dict[str, str], description: str | None = None)

Bases: object

A predefined workflow pattern combining multiple toolsets.

name

Unique identifier for this template

Type:

str

toolsets

List of toolset IDs in order

Type:

list[str]

stages

List of stages with transition conditions

Type:

list[pydantic_ai_toolsets.toolsets.meta_orchestrator.types.Stage]

handoff_instructions

Mapping of stage transitions to instructions

Type:

dict[str, str]

description

Optional description of when to use this template

Type:

str | None

name: str

toolsets: list[str]

stages: list[Stage]

handoff_instructions: dict[str, str]

description: str | None = None

__init__(name: str, toolsets: list[str], stages: list[Stage], handoff_instructions: dict[str, str], description: str | None = None) → None

pydantic_ai_toolsets.create_combined_toolset(toolsets: list[AbstractToolset[Any]], storages: dict[str, Any] | None = None, prefix_map: dict[str, str] | None = None, orchestrator: AbstractToolset[Any] | None = None, workflow_template: WorkflowTemplate | None = None, auto_prefix: bool = True) → tuple[CombinedToolset[Any], str]

Combine multiple toolsets with automatic collision resolution.

Uses official pydantic-ai API: - AbstractToolset.prefixed() to create aliased toolsets - CombinedToolset to combine toolsets

Strategy: 1. If auto_prefix=True, apply prefixes to all toolsets based on prefix_map

(prevents collisions proactively)

2. If auto_prefix=False, rely on CombinedToolset to detect collisions (raises UserError if collisions exist - user must handle)

3. Use CombinedToolset to combine all toolsets

4. Add orchestrator tools if provided

5. Combine system prompts from all toolsets

Parameters:

• toolsets – List of toolsets to combine

• storages – Optional dictionary mapping toolset IDs to storage instances

• prefix_map – Optional dictionary mapping toolset IDs to prefixes. If not provided, prefixes are inferred from toolset IDs.

• orchestrator – Optional meta-orchestrator toolset to add

• workflow_template – Optional workflow template for context

• auto_prefix – If True, automatically prefix all toolsets to prevent collisions

Returns:

Tuple of (CombinedToolset, combined_system_prompt)

from pydantic_ai_toolsets import create_cot_toolset, create_tot_toolset
from pydantic_ai_toolsets.toolsets.meta_orchestrator.helpers import create_combined_toolset

cot_toolset = create_cot_toolset()
tot_toolset = create_tot_toolset()

prefix_map = {
    "cot": "cot_",
    "tot": "tot_",
}

combined_toolset, combined_prompt = create_combined_toolset(
    toolsets=[cot_toolset, tot_toolset],
    prefix_map=prefix_map,
)

pydantic_ai_toolsets.create_meta_orchestrator_toolset(storage: MetaOrchestratorStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a meta-orchestrator toolset for workflow management.

This toolset provides tools for AI agents to orchestrate multi-toolset workflows, manage transitions between toolsets, and create cross-toolset links.

Parameters:

• storage – Optional storage backend. Defaults to in-memory MetaOrchestratorStorage. You can provide a custom storage implementing MetaOrchestratorStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection in storage.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_meta_orchestrator_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_meta_orchestrator_toolset()])
result = await agent.run("Start a research assistant workflow")

Example (with storage access):

from pydantic_ai_toolsets import create_meta_orchestrator_toolset, MetaOrchestratorStorage

storage = MetaOrchestratorStorage()
toolset = create_meta_orchestrator_toolset(storage=storage)

# After agent runs, access workflow state directly
workflow = storage.get_active_workflow()
print(workflow.current_stage)
print(storage.links)

pydantic_ai_toolsets.create_workflow_agent(model: str, workflow_template: WorkflowTemplate, toolsets: list[AbstractToolset[Any]], storages: dict[str, Any] | None = None, prefix_map: dict[str, str] | None = None, orchestrator_storage: Any | None = None, auto_prefix: bool = True, additional_system_prompt: str | None = None, output_type: type[BaseModel] | list[type[BaseModel]] | None = None) → Agent[Any, str]

Create an agent configured with a workflow template and combined toolsets.

This is a convenience function that: 1. Creates a meta-orchestrator toolset (if orchestrator_storage is provided) 2. Registers all toolsets with the orchestrator 3. Combines all toolsets with automatic prefixing 4. Creates an agent with the combined toolset and workflow-aware system prompt

Parameters:

• model – Model string for the agent (e.g., “openai:gpt-4”)

• workflow_template – Workflow template to use

• toolsets – List of toolsets to combine

• storages – Optional dictionary mapping toolset IDs to storage instances

• prefix_map – Optional dictionary mapping toolset IDs to prefixes. If not provided, prefixes are inferred from toolset IDs.

• orchestrator_storage – Optional MetaOrchestratorStorage instance. If provided, creates and registers orchestrator toolset.

• auto_prefix – If True, automatically prefix all toolsets to prevent collisions

• additional_system_prompt – Optional additional system prompt that will be appended to the combined prompt. Use this to add custom instructions or context without overriding the workflow-specific prompts.

• output_type – Optional Pydantic BaseModel class or list of BaseModel classes to use as the structured output type for the agent. If provided, the agent will return structured outputs matching this schema.

Returns:

Configured Agent instance with combined toolsets and workflow template

from pydantic_ai_toolsets import (
    RESEARCH_ASSISTANT,
    create_search_toolset,
    create_self_ask_toolset,
    create_self_refine_toolset,
    create_todo_toolset,
    SearchStorage,
    SelfAskStorage,
    SelfRefineStorage,
    TodoStorage,
    MetaOrchestratorStorage,
)
from pydantic_ai_toolsets.toolsets.meta_orchestrator.helpers import create_workflow_agent

# Create storages
storages = {
    "search": SearchStorage(),
    "self_ask": SelfAskStorage(),
    "self_refine": SelfRefineStorage(),
    "todo": TodoStorage(),
}

# Create toolsets
toolsets = [
    create_search_toolset(storages["search"], id="search"),
    create_self_ask_toolset(storages["self_ask"], id="self_ask"),
    create_self_refine_toolset(storages["self_refine"], id="self_refine"),
    create_todo_toolset(storages["todo"], id="todo"),
]

# Create orchestrator storage
orchestrator_storage = MetaOrchestratorStorage()

# Create agent with workflow template and additional instructions
agent = create_workflow_agent(
    model="openai:gpt-4",
    workflow_template=RESEARCH_ASSISTANT,
    toolsets=toolsets,
    storages=storages,
    orchestrator_storage=orchestrator_storage,
    additional_system_prompt="Always cite sources and provide URLs when available.",
)

# Use the agent
result = await agent.run("Research quantum computing breakthroughs")

# Example with output type:
from pydantic import BaseModel

class ResearchResult(BaseModel):
    summary: str
    sources: list[str]
    key_findings: list[str]

agent_with_output = create_workflow_agent(
    model="openai:gpt-4",
    workflow_template=RESEARCH_ASSISTANT,
    toolsets=toolsets,
    storages=storages,
    orchestrator_storage=orchestrator_storage,
    output_type=ResearchResult,
)
result = await agent_with_output.run("Research quantum computing breakthroughs")
print(result.output.summary)  # Access structured output

pydantic_ai_toolsets.get_template(name: str) → WorkflowTemplate | None

Get a workflow template by name.

Parameters:

name – Name of the template (e.g., ‘research_assistant’)

Returns:

WorkflowTemplate if found, None otherwise

pydantic_ai_toolsets.list_templates() → list[str]

List all available workflow template names.

Returns:

List of template names

pydantic_ai_toolsets.register_toolsets_with_orchestrator(orchestrator_storage: Any, toolsets: list[AbstractToolset[Any]], storages: dict[str, Any] | None = None) → None

Register toolsets with the meta-orchestrator storage.

Parameters:

• orchestrator_storage – MetaOrchestratorStorage instance

• toolsets – List of toolsets to register

• storages – Optional dictionary mapping toolset IDs to storage instances

Reasoning & Thinking Toolsets

Chain of Thought Reasoning

Chain of thoughts toolset for pydantic-ai agents.

Provides reasoning exploration and documentation capabilities for AI agents. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_cot_toolset, CoTStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_cot_toolset()])

# With storage access
storage = CoTStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_cot_toolset(storage)])
result = await agent.run("Solve this complex problem step by step")
print(storage.thoughts)  # Access thoughts directly

# With usage tracking
storage = CoTStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_cot_toolset(storage)])
result = await agent.run("Solve this problem")
print(storage.metrics.total_tokens())  # Check token usage

pydantic_ai_toolsets.toolsets.chain_of_thought_reasoning.create_cot_toolset(storage: CoTStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a chain of thoughts toolset for reasoning exploration.

This toolset provides read_thoughts and write_thoughts tools for AI agents to document and explore their reasoning process during a session.

Parameters:

• storage – Optional storage backend. Defaults to in-memory CoTStorage. You can provide a custom storage implementing CoTStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection on the default storage. Ignored if custom storage is provided.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_cot_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_cot_toolset()])
result = await agent.run("Solve this complex problem step by step")

Example (with custom storage):

from pydantic_ai_toolsets import create_cot_toolset, CoTStorage

storage = CoTStorage()
toolset = create_cot_toolset(storage=storage)

# After agent runs, access thoughts directly
print(storage.thoughts)

Example (with usage tracking):

from pydantic_ai_toolsets import create_cot_toolset, CoTStorage

storage = CoTStorage(track_usage=True)
toolset = create_cot_toolset(storage=storage)

# After agent runs, check usage metrics
print(storage.metrics.total_tokens())
print(storage.metrics.invocation_count())

pydantic_ai_toolsets.toolsets.chain_of_thought_reasoning.get_cot_system_prompt(storage: CoTStorageProtocol | None = None) → str

Generate dynamic system prompt section for chain of thoughts.

Parameters:

storage – Optional storage to read current thoughts from.

Returns:

System prompt section with current thoughts, or base prompt if no thoughts.

class pydantic_ai_toolsets.toolsets.chain_of_thought_reasoning.Thought(*, thought: str, thought_number: int, total_thoughts: int, is_revision: bool = False, revises_thought: int | None = None, branch_from_thought: int | None = None, branch_id: str | None = None, next_thought_needed: bool = True)

Bases: BaseModel

A thought in a chain of thoughts execution.

thought

The current thinking step content.

Type:

str

thought_number

Current number in sequence (1-based).

Type:

int

total_thoughts

Estimated total thoughts needed (can be adjusted).

Type:

int

is_revision

Whether this thought revises previous thinking.

Type:

bool

revises_thought

Which thought number is being reconsidered (if is_revision).

Type:

int | None

branch_from_thought

Branching point thought number (if branching).

Type:

int | None

branch_id

Identifier for the current branch (if branching).

Type:

str | None

next_thought_needed

Whether another thought step is needed.

Type:

bool

thought: str

thought_number: int

total_thoughts: int

is_revision: bool

revises_thought: int | None

branch_from_thought: int | None

branch_id: str | None

next_thought_needed: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.chain_of_thought_reasoning.ThoughtItem(*, thought: str, thought_number: int, total_thoughts: int, is_revision: bool = False, revises_thought: int | None = None, branch_from_thought: int | None = None, branch_id: str | None = None, next_thought_needed: bool = True)

Bases: BaseModel

Input model for the write_thoughts tool.

This is the model that agents use when calling write_thoughts. It has the same fields as Thought but with Field descriptions for LLM guidance.

thought: str

thought_number: int

total_thoughts: int

is_revision: bool

revises_thought: int | None

branch_from_thought: int | None

branch_id: str | None

next_thought_needed: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.chain_of_thought_reasoning.CoTStorage(*, track_usage: bool = False)

Bases: object

Default in-memory chain of thoughts storage.

Simple implementation that stores thoughts in memory. Use this for standalone agents or testing.

_thoughts

Internal list of recorded thoughts.

Type:

list[Thought]

_metrics

Optional usage metrics tracker (enabled via track_usage parameter).

Type:

UsageMetrics | None

from pydantic_ai_toolsets import create_cot_toolset, CoTStorage

storage = CoTStorage()
toolset = create_cot_toolset(storage=storage)

# After agent runs, access thoughts directly
print(storage.thoughts)

# With metrics tracking
storage = CoTStorage(track_usage=True)
toolset = create_cot_toolset(storage=storage)
# After agent runs
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property thoughts: list[Thought]

Get the current list of thoughts.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

Returns:

UsageMetrics instance if track_usage=True was set, otherwise None.

get_statistics() → dict[str, int | float]

Get summary statistics about the chain of thoughts.

Returns:

Dictionary with thought counts and metadata.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all thoughts and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (e.g., thought number as string)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of thoughts and statistics.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for each linkable thought.

class pydantic_ai_toolsets.toolsets.chain_of_thought_reasoning.CoTStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for chain of thoughts storage implementations.

Any class that has a thoughts property (read returns list, write appends Thought) can be used as storage for the CoT toolset.

class MyCustomStorage:
    def __init__(self):
        self._thoughts: list[Thought] = []

    @property
    def thoughts(self) -> list[Thought]:
        return self._thoughts

    @thoughts.setter
    def thoughts(self, value: Thought) -> None:
        self._thoughts.append(value)

property thoughts: list[Thought]

Get the current list of thoughts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (e.g., thought number as string)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Tree of Thought Reasoning

Tree of thoughts toolset for pydantic-ai agents.

Provides multi-path reasoning exploration capabilities for AI agents. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_tot_toolset, ToTStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_tot_toolset()])

# With storage access
storage = ToTStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_tot_toolset(storage)])
result = await agent.run("Solve this complex problem exploring multiple approaches")
print(storage.nodes)  # Access nodes directly
print(storage.evaluations)  # Access evaluations directly

# With usage tracking
storage = ToTStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_tot_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.tree_of_thought_reasoning.create_tot_toolset(storage: ToTStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a tree of thoughts toolset for multi-path reasoning exploration.

This toolset provides tools for AI agents to explore multiple reasoning paths simultaneously, evaluate branches, prune dead ends, and merge insights.

Parameters:

• storage – Optional storage backend. Defaults to in-memory ToTStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_tot_toolset, ToTStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_tot_toolset()])

# With storage and metrics
storage = ToTStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_tot_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.tree_of_thought_reasoning.get_tot_system_prompt(storage: ToTStorageProtocol | None = None) → str

Generate dynamic system prompt section for tree of thoughts.

Parameters:

storage – Optional storage to read current tree from.

Returns:

System prompt section with current tree state, or base prompt if no nodes.

class pydantic_ai_toolsets.toolsets.tree_of_thought_reasoning.ThoughtNode(*, node_id: str, content: str, parent_id: str | None = None, depth: int = 0, branch_id: str | None = None, status: Literal['active', 'pruned', 'merged', 'completed']='active', evaluation_score: float | None = None, is_solution: bool = False, merged_from: list[str] = <factory>)

Bases: BaseModel

A node in a tree of thoughts execution.

Each node represents a reasoning state. Nodes form a tree structure where multiple branches can be explored simultaneously.

node_id

Unique identifier for this node.

Type:

str

content

The reasoning content at this node.

Type:

str

parent_id

ID of the parent node (None for root nodes).

Type:

str | None

depth

Depth level in the tree (0 for root).

Type:

int

branch_id

Identifier for the branch this node belongs to.

Type:

str | None

status

Current status of this node/branch.

Type:

Literal[‘active’, ‘pruned’, ‘merged’, ‘completed’]

evaluation_score

Optional score evaluating this path (0-100).

Type:

float | None

is_solution

Whether this node represents a solution.

Type:

bool

merged_from

List of node IDs whose insights were merged into this node.

Type:

list[str]

node_id: str

content: str

parent_id: str | None

depth: int

branch_id: str | None

status: Literal['active', 'pruned', 'merged', 'completed']

evaluation_score: float | None

is_solution: bool

merged_from: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.tree_of_thought_reasoning.NodeItem(*, content: str, parent_id: str | None = None, branch_id: str | None = None, is_solution: bool = False)

Bases: BaseModel

Input model for the create_node tool.

This is the model that agents use when calling create_node. It has the same fields as ThoughtNode but with Field descriptions for LLM guidance.

content: str

parent_id: str | None

branch_id: str | None

is_solution: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.tree_of_thought_reasoning.BranchEvaluation(*, branch_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str, recommendation: Literal['continue', 'prune', 'merge', 'explore_deeper'])

Bases: BaseModel

Evaluation of a branch/path in the tree.

Used to assess the promise of different reasoning paths.

branch_id

Identifier for the branch being evaluated.

Type:

str

score

Evaluation score (0-100, higher is better).

Type:

float

reasoning

Explanation of why this score was assigned.

Type:

str

recommendation

Recommended action for this branch.

Type:

Literal[‘continue’, ‘prune’, ‘merge’, ‘explore_deeper’]

branch_id: str

score: float

reasoning: str

recommendation: Literal['continue', 'prune', 'merge', 'explore_deeper']

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.tree_of_thought_reasoning.BranchEvaluationItem(*, branch_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str, recommendation: Literal['continue', 'prune', 'merge', 'explore_deeper'])

Bases: BaseModel

Input model for the evaluate_branch tool.

branch_id: str

score: float

reasoning: str

recommendation: Literal['continue', 'prune', 'merge', 'explore_deeper']

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.tree_of_thought_reasoning.ToTStorage(*, track_usage: bool = False)

Bases: object

Default in-memory tree of thoughts storage.

Simple implementation that stores nodes and evaluations in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_tot_toolset, ToTStorage

storage = ToTStorage()
toolset = create_tot_toolset(storage=storage)

# After agent runs, access nodes and evaluations directly
print(storage.nodes)
print(storage.evaluations)

# With metrics tracking
storage = ToTStorage(track_usage=True)
toolset = create_tot_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property nodes: dict[str, ThoughtNode]

Get the current dictionary of nodes.

property evaluations: dict[str, BranchEvaluation]

Get the current dictionary of branch evaluations.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about the tree.

Returns:

Dictionary with node counts and tree metrics.

depth_statistics() → dict[int, int]

Get node count at each depth level.

Returns:

Dictionary mapping depth to node count.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all nodes, evaluations, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id or evaluation_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of nodes and evaluations.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for nodes and evaluations.

class pydantic_ai_toolsets.toolsets.tree_of_thought_reasoning.ToTStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for tree of thoughts storage implementations.

Any class that has nodes and evaluations properties can be used as storage for the ToT toolset.

class MyCustomStorage:
    def __init__(self):
        self._nodes: dict[str, ThoughtNode] = {}
        self._evaluations: dict[str, BranchEvaluation] = {}

    @property
    def nodes(self) -> dict[str, ThoughtNode]:
        return self._nodes

    @nodes.setter
    def nodes(self, value: ThoughtNode) -> None:
        self._nodes[value.node_id] = value

    @property
    def evaluations(self) -> dict[str, BranchEvaluation]:
        return self._evaluations

    @evaluations.setter
    def evaluations(self, value: BranchEvaluation) -> None:
        self._evaluations[value.branch_id] = value

property nodes: dict[str, ThoughtNode]

Get the current dictionary of nodes (node_id -> ThoughtNode).

property evaluations: dict[str, BranchEvaluation]

Get the current dictionary of branch evaluations (branch_id -> BranchEvaluation).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id or evaluation_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Graph of Thought Reasoning

Graph of thoughts toolset for pydantic-ai agents.

Provides graph-based reasoning exploration capabilities for AI agents. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_got_toolset, GoTStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_got_toolset()])

# With storage access
storage = GoTStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_got_toolset(storage)])
result = await agent.run("Solve this complex interconnected problem")
print(storage.nodes)  # Access nodes directly
print(storage.edges)  # Access edges directly
print(storage.evaluations)  # Access evaluations directly

# With usage tracking
storage = GoTStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_got_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.create_got_toolset(storage: GoTStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a graph of thoughts toolset for graph-based reasoning.

This toolset provides tools for AI agents to explore reasoning using a directed graph structure with nodes (reasoning states) and edges (connections/dependencies).

Parameters:

• storage – Optional storage backend. Defaults to in-memory GoTStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_got_toolset, GoTStorage

# With storage and metrics
storage = GoTStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_got_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.get_got_system_prompt(storage: GoTStorageProtocol | None = None) → str

Generate dynamic system prompt section for graph of thoughts.

Parameters:

storage – Optional storage to read current graph from.

Returns:

System prompt section with current graph state, or base prompt if empty.

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.GraphNode(*, node_id: str, content: str, evaluation_score: float | None = None, is_solution: bool = False, status: Literal['active', 'completed', 'pruned']='active', aggregated_from: list[str] = <factory>, refined_from: str | None = None, refinement_count: int = 0)

Bases: BaseModel

A node in a graph of thoughts execution.

Each node represents a reasoning state. Nodes form a directed graph structure where edges represent dependencies, logical connections, or information flow.

node_id

Unique identifier for this node.

Type:

str

content

The reasoning content at this node.

Type:

str

evaluation_score

Optional score evaluating this node (0-100).

Type:

float | None

is_solution

Whether this node represents a solution.

Type:

bool

status

Current status of this node.

Type:

Literal[‘active’, ‘completed’, ‘pruned’]

aggregated_from

List of node IDs whose insights were aggregated into this node.

Type:

list[str]

refined_from

Node ID that this node refines (if any).

Type:

str | None

refinement_count

Number of times this node has been refined.

Type:

int

node_id: str

content: str

evaluation_score: float | None

is_solution: bool

status: Literal['active', 'completed', 'pruned']

aggregated_from: list[str]

refined_from: str | None

refinement_count: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.GraphEdge(*, edge_id: str, source_id: str, target_id: str, edge_type: Literal['dependency', 'aggregation', 'refinement', 'reference', 'merge'] = 'dependency', weight: float | None = None)

Bases: BaseModel

An edge in a graph of thoughts execution.

Edges represent dependencies, logical connections, or information flow between nodes. They form a directed graph (DAG or with cycles).

edge_id

Unique identifier for this edge.

Type:

str

source_id

ID of the source node.

Type:

str

target_id

ID of the target node.

Type:

str

edge_type

Type of connection (dependency, aggregation, refinement, etc.).

Type:

Literal[‘dependency’, ‘aggregation’, ‘refinement’, ‘reference’, ‘merge’]

weight

Optional weight indicating strength/importance of the connection.

Type:

float | None

edge_id: str

source_id: str

target_id: str

edge_type: Literal['dependency', 'aggregation', 'refinement', 'reference', 'merge']

weight: float | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.NodeEvaluation(*, node_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str, recommendation: Literal['keep', 'refine', 'aggregate', 'prune'])

Bases: BaseModel

Evaluation of a node in the graph.

Used to assess the quality and relevance of individual nodes.

node_id

Identifier for the node being evaluated.

Type:

str

score

Evaluation score (0-100, higher is better).

Type:

float

reasoning

Explanation of why this score was assigned.

Type:

str

recommendation

Recommended action for this node.

Type:

Literal[‘keep’, ‘refine’, ‘aggregate’, ‘prune’]

node_id: str

score: float

reasoning: str

recommendation: Literal['keep', 'refine', 'aggregate', 'prune']

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.NodeItem(*, content: str, is_solution: bool = False)

Bases: BaseModel

Input model for the create_node tool.

This is the model that agents use when calling create_node.

content: str

is_solution: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.EdgeItem(*, source_id: str, target_id: str, edge_type: Literal['dependency', 'aggregation', 'refinement', 'reference', 'merge'] = 'dependency', weight: Annotated[float | None, Ge(ge=0), Le(le=1)] = None)

Bases: BaseModel

Input model for the create_edge tool.

source_id: str

target_id: str

edge_type: Literal['dependency', 'aggregation', 'refinement', 'reference', 'merge']

weight: float | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.AggregateItem(*, source_node_ids: list[str], aggregated_content: str, is_solution: bool = False)

Bases: BaseModel

Input model for the aggregate_nodes tool.

source_node_ids: list[str]

aggregated_content: str

is_solution: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.RefineItem(*, node_id: str, refined_content: str, is_solution: bool = False)

Bases: BaseModel

Input model for the refine_node tool.

node_id: str

refined_content: str

is_solution: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.NodeEvaluationItem(*, node_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str, recommendation: Literal['keep', 'refine', 'aggregate', 'prune'])

Bases: BaseModel

Input model for the evaluate_node tool.

node_id: str

score: float

reasoning: str

recommendation: Literal['keep', 'refine', 'aggregate', 'prune']

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.GoTStorage(*, track_usage: bool = False)

Bases: object

Default in-memory graph of thoughts storage.

Simple implementation that stores nodes, edges, and evaluations in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_got_toolset, GoTStorage

storage = GoTStorage()
toolset = create_got_toolset(storage=storage)

# After agent runs, access nodes, edges, and evaluations directly
print(storage.nodes)
print(storage.edges)
print(storage.evaluations)

# With metrics tracking
storage = GoTStorage(track_usage=True)
toolset = create_got_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property nodes: dict[str, GraphNode]

Get the current dictionary of nodes.

property edges: dict[str, GraphEdge]

Get the current dictionary of edges.

property evaluations: dict[str, NodeEvaluation]

Get the current dictionary of node evaluations.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about the graph.

Returns:

Dictionary with node, edge, and evaluation counts.

graph_complexity() → dict[str, float]

Calculate graph complexity metrics.

Returns:

Dictionary with complexity metrics (density, avg degree).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all nodes, edges, evaluations, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id, edge_id, or evaluation_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of nodes, edges, and evaluations.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for nodes, edges, and evaluations.

class pydantic_ai_toolsets.toolsets.graph_of_thought_reasoning.GoTStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for graph of thoughts storage implementations.

Any class that has nodes, edges, and evaluations properties can be used as storage for the GoT toolset.

class MyCustomStorage:
    def __init__(self):
        self._nodes: dict[str, GraphNode] = {}
        self._edges: dict[str, GraphEdge] = {}
        self._evaluations: dict[str, NodeEvaluation] = {}

    @property
    def nodes(self) -> dict[str, GraphNode]:
        return self._nodes

    @nodes.setter
    def nodes(self, value: GraphNode) -> None:
        self._nodes[value.node_id] = value

    @property
    def edges(self) -> dict[str, GraphEdge]:
        return self._edges

    @edges.setter
    def edges(self, value: GraphEdge) -> None:
        self._edges[value.edge_id] = value

    @property
    def evaluations(self) -> dict[str, NodeEvaluation]:
        return self._evaluations

    @evaluations.setter
    def evaluations(self, value: NodeEvaluation) -> None:
        self._evaluations[value.node_id] = value

property nodes: dict[str, GraphNode]

Get the current dictionary of nodes (node_id -> GraphNode).

property edges: dict[str, GraphEdge]

Get the current dictionary of edges (edge_id -> GraphEdge).

property evaluations: dict[str, NodeEvaluation]

Get the current dictionary of node evaluations (node_id -> NodeEvaluation).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id, edge_id, or evaluation_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Beam Search Reasoning

Beam search toolset for pydantic-ai agents.

Provides beam search reasoning exploration capabilities for AI agents. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_beam_toolset, BeamStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_beam_toolset()])

# With storage access
storage = BeamStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_beam_toolset(storage)])
result = await agent.run("Solve this problem using beam search")
print(storage.candidates)  # Access candidates directly
print(storage.steps)  # Access steps directly

# With usage tracking
storage = BeamStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_beam_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.beam_search_reasoning.create_beam_toolset(storage: BeamStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a beam search toolset for beam-based reasoning exploration.

This toolset provides tools for AI agents to explore reasoning using beam search, maintaining a beam of top-k candidates at each step.

Parameters:

• storage – Optional storage backend. Defaults to in-memory BeamStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_beam_toolset, BeamStorage

# With storage and metrics
storage = BeamStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_beam_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.beam_search_reasoning.get_beam_system_prompt(storage: BeamStorageProtocol | None = None) → str

Generate dynamic system prompt section for beam search.

Parameters:

storage – Optional storage to read current beam from.

Returns:

System prompt section with current beam state, or base prompt if no candidates.

class pydantic_ai_toolsets.toolsets.beam_search_reasoning.BeamCandidate(*, candidate_id: str, content: str, score: float | None = None, depth: Annotated[int, Ge(ge=0)] = 0, parent_id: str | None = None, is_terminal: bool = False, step_index: Annotated[int, Ge(ge=0)] = 0)

Bases: BaseModel

A candidate in the beam search execution.

Each candidate represents a reasoning state at a specific depth in the search. Candidates form a tree structure where each candidate can have multiple children (expansions) but only one parent.

candidate_id

Unique identifier for this candidate.

Type:

str

content

The reasoning content at this candidate.

Type:

str

score

Evaluation score for this candidate (0-100, higher is better).

Type:

float | None

depth

Depth level in the search tree (0 for initial candidates).

Type:

int

parent_id

ID of the parent candidate (None for root candidates).

Type:

str | None

is_terminal

Whether this candidate represents a terminal/solution state.

Type:

bool

step_index

Which beam step this candidate belongs to.

Type:

int

candidate_id: str

content: str

score: float | None

depth: int

parent_id: str | None

is_terminal: bool

step_index: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.beam_search_reasoning.BeamStep(*, step_index: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], candidate_ids: list[str] = <factory>, beam_width: ~typing.Annotated[int, ~annotated_types.Ge(ge=1)])

Bases: BaseModel

A step in the beam search execution.

Each step contains the top-k candidates (the “beam”) at that depth level. Steps are ordered sequentially, with step 0 containing initial candidates.

step_index

Sequential index of this step (0-based).

Type:

int

candidate_ids

List of candidate IDs in this step’s beam (top-k).

Type:

list[str]

beam_width

The beam width (k) used for this step.

Type:

int

step_index: int

candidate_ids: list[str]

beam_width: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.beam_search_reasoning.CreateCandidateItem(*, content: str, is_terminal: bool = False)

Bases: BaseModel

Input model for the create_candidate tool.

This is the model that agents use when calling create_candidate.

content: str

is_terminal: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.beam_search_reasoning.ExpandCandidateItem(*, candidate_id: str, expansions: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], is_terminal: list[bool] = <factory>)

Bases: BaseModel

Input model for the expand_candidate tool.

candidate_id: str

expansions: list[str]

is_terminal: list[bool]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.beam_search_reasoning.PruneBeamItem(*, step_index: Annotated[int, Ge(ge=0)], beam_width: Annotated[int, Ge(ge=1)])

Bases: BaseModel

Input model for the prune_beam tool.

step_index: int

beam_width: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.beam_search_reasoning.ScoreCandidateItem(*, candidate_id: str, score: Annotated[float, Ge(ge=0), Le(le=100)], reasoning: str)

Bases: BaseModel

Input model for the score_candidate tool.

candidate_id: str

score: float

reasoning: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.beam_search_reasoning.BeamStorage(*, track_usage: bool = False)

Bases: object

Default in-memory beam search storage.

Simple implementation that stores candidates and steps in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_beam_toolset, BeamStorage

storage = BeamStorage()
toolset = create_beam_toolset(storage=storage)

# After agent runs, access candidates and steps directly
print(storage.candidates)
print(storage.steps)

# With metrics tracking
storage = BeamStorage(track_usage=True)
toolset = create_beam_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property candidates: dict[str, BeamCandidate]

Get the current dictionary of candidates.

property steps: list[BeamStep]

Get the current list of beam steps.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about the beam search.

Returns:

Dictionary with candidate counts and beam metrics.

beam_width_history() → list[tuple[int, int]]

Get beam width at each step.

Returns:

List of (step_index, beam_width) tuples.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all candidates, steps, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (candidate_id or step index as string)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of candidates and steps.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for candidates and steps.

class pydantic_ai_toolsets.toolsets.beam_search_reasoning.BeamStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for beam search storage implementations.

Any class that has candidates and steps properties can be used as storage for the beam search toolset.

class MyCustomStorage:
    def __init__(self):
        self._candidates: dict[str, BeamCandidate] = {}
        self._steps: list[BeamStep] = []

    @property
    def candidates(self) -> dict[str, BeamCandidate]:
        return self._candidates

    @candidates.setter
    def candidates(self, value: BeamCandidate) -> None:
        self._candidates[value.candidate_id] = value

    @property
    def steps(self) -> list[BeamStep]:
        return self._steps

    @steps.setter
    def steps(self, value: BeamStep) -> None:
        # Update or append step
        ...

property candidates: dict[str, BeamCandidate]

Get the current dictionary of candidates (candidate_id -> BeamCandidate).

property steps: list[BeamStep]

Get the current list of beam steps.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (candidate_id or step_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Monte Carlo Tree Search Reasoning

Monte Carlo Tree Search toolset for pydantic-ai agents.

Provides MCTS-based reasoning exploration capabilities for AI agents. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_mcts_toolset, MCTSStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_mcts_toolset()])

# With storage access
storage = MCTSStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_mcts_toolset(storage)])
result = await agent.run("Solve this decision problem using MCTS")
print(storage.nodes)  # Access nodes directly

# With usage tracking
storage = MCTSStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_mcts_toolset(storage)])
print(storage.metrics.total_tokens())
print(storage.iteration_count)

pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.create_mcts_toolset(storage: MCTSStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create an MCTS toolset for tree-based exploration with statistics.

This toolset provides tools for AI agents to explore reasoning using Monte Carlo Tree Search, balancing exploration and exploitation.

Parameters:

• storage – Optional storage backend. Defaults to in-memory MCTSStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_mcts_toolset, MCTSStorage

# With storage and metrics
storage = MCTSStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_mcts_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.get_mcts_system_prompt(storage: MCTSStorageProtocol | None = None) → str

Generate dynamic system prompt section for MCTS.

Parameters:

storage – Optional storage to read current tree from.

Returns:

System prompt section with current tree state, or base prompt if empty.

pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.calculate_ucb1(node: MCTSNode, parent_visits: int, exploration_constant: float) → float

Calculate UCB1 value for a node.

UCB1 = win_rate + c * sqrt(ln(parent_visits) / visits)

Parameters:

• node – The node to calculate UCB1 for.

• parent_visits – Total visits of the parent node.

• exploration_constant – The exploration constant (c).

Returns:

UCB1 value, or infinity if node hasn’t been visited.

class pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.MCTSNode(*, node_id: str, content: str, visits: Annotated[int, ~annotated_types.Ge(ge=0)] = 0, wins: Annotated[float, ~annotated_types.Ge(ge=0)] = 0.0, parent_id: str | None = None, children_ids: list[str] = <factory>, is_terminal: bool = False, is_expanded: bool = False, depth: Annotated[int, ~annotated_types.Ge(ge=0)] = 0)

Bases: BaseModel

A node in a Monte Carlo Tree Search execution.

Each node represents a state in the search tree. Nodes track visit counts, win/reward totals, and maintain parent-child relationships for the tree structure.

node_id

Unique identifier for this node.

Type:

str

content

The reasoning/state content at this node.

Type:

str

visits

Number of times this node has been visited.

Type:

int

wins

Total reward/wins accumulated from simulations passing through this node.

Type:

float

parent_id

ID of the parent node (None for root node).

Type:

str | None

children_ids

List of child node IDs.

Type:

list[str]

is_terminal

Whether this node represents a terminal/solution state.

Type:

bool

is_expanded

Whether this node has been expanded (has children).

Type:

bool

depth

Depth level in the tree (0 for root).

Type:

int

node_id: str

content: str

visits: int

wins: float

parent_id: str | None

children_ids: list[str]

is_terminal: bool

is_expanded: bool

depth: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.SelectNodeItem(*, node_id: str | None = None, exploration_constant: Annotated[float, Ge(ge=0)] = 1.414)

Bases: BaseModel

Input model for the select_node tool.

Allows manual selection of a node, or automatic selection using UCB1.

node_id: str | None

exploration_constant: float

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.ExpandNodeItem(*, node_id: str, children: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], is_terminal: list[bool] = <factory>)

Bases: BaseModel

Input model for the expand_node tool.

node_id: str

children: list[str]

is_terminal: list[bool]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.SimulateItem(*, node_id: str, simulation_result: ~typing.Annotated[float, ~annotated_types.Ge(ge=0.0), ~annotated_types.Le(le=1.0)], simulation_path: list[str] = <factory>)

Bases: BaseModel

Input model for the simulate tool.

node_id: str

simulation_result: float

simulation_path: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.BackpropagateItem(*, node_id: str, reward: Annotated[float, Ge(ge=0.0), Le(le=1.0)])

Bases: BaseModel

Input model for the backpropagate tool.

node_id: str

reward: float

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.MCTSStorage(*, track_usage: bool = False)

Bases: object

Default in-memory MCTS storage.

Simple implementation that stores nodes in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_mcts_toolset, MCTSStorage

storage = MCTSStorage()
toolset = create_mcts_toolset(storage=storage)

# After agent runs, access nodes directly
print(storage.nodes)

# With metrics tracking
storage = MCTSStorage(track_usage=True)
toolset = create_mcts_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property nodes: dict[str, MCTSNode]

Get the current dictionary of nodes.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

property iteration_count: int

Get the number of MCTS iterations performed.

increment_iteration() → None

Increment the iteration counter.

get_statistics() → dict[str, int | float]

Get summary statistics about the MCTS tree.

Returns:

Dictionary with node counts and tree metrics.

get_ucb1_stats() → list[tuple[str, float, int, float]]

Get UCB1 statistics for all nodes.

Returns:

List of (node_id, win_rate, visits, ucb1_value) tuples. UCB1 calculated with c=sqrt(2).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all nodes and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of nodes and iterations.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for nodes.

class pydantic_ai_toolsets.toolsets.monte_carlo_reasoning.MCTSStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for MCTS storage implementations.

Any class that has a nodes property can be used as storage for the MCTS toolset.

class MyCustomStorage:
    def __init__(self):
        self._nodes: dict[str, MCTSNode] = {}

    @property
    def nodes(self) -> dict[str, MCTSNode]:
        return self._nodes

    @nodes.setter
    def nodes(self, value: MCTSNode) -> None:
        self._nodes[value.node_id] = value

property nodes: dict[str, MCTSNode]

Get the current dictionary of nodes (node_id -> MCTSNode).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (node_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Reflection & Refinement Toolsets

Reflection

Reflection toolset for pydantic-ai agents.

Provides reflection-based reasoning capabilities for AI agents. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_reflection_toolset, ReflectionStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_reflection_toolset()])

# With storage access
storage = ReflectionStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_reflection_toolset(storage)])
result = await agent.run("Solve this problem using reflection")
print(storage.outputs)  # Access outputs directly

pydantic_ai_toolsets.toolsets.reflection.create_reflection_toolset(storage: ReflectionStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a reflection toolset for iterative output improvement.

This toolset provides tools for AI agents to improve outputs through critical analysis and refinement cycles.

Parameters:

• storage – Optional storage backend. Defaults to in-memory ReflectionStorage. You can provide a custom storage implementing ReflectionStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_reflection_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_reflection_toolset()])
result = await agent.run("Solve this problem using reflection")

Example (with custom storage):

from pydantic_ai_toolsets import create_reflection_toolset, ReflectionStorage

storage = ReflectionStorage()
toolset = create_reflection_toolset(storage=storage)

# After agent runs, access outputs and critiques directly
print(storage.outputs)
print(storage.critiques)

pydantic_ai_toolsets.toolsets.reflection.get_reflection_system_prompt() → str

Get the system prompt for reflection-based reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

class pydantic_ai_toolsets.toolsets.reflection.ReflectionOutput(*, output_id: str, content: str, cycle: Annotated[int, Ge(ge=0)] = 0, parent_id: str | None = None, is_final: bool = False, quality_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None)

Bases: BaseModel

An output that can be critiqued and refined.

Each output represents a version of the solution at a specific refinement cycle. Outputs form a chain where each output can be critiqued and refined into a new output.

output_id

Unique identifier for this output.

Type:

str

content

The actual output content (solution, answer, etc.).

Type:

str

cycle

The refinement cycle number (0 for initial output).

Type:

int

parent_id

ID of the parent output that was refined to create this one (None for initial).

Type:

str | None

is_final

Whether this output is marked as final/satisfactory.

Type:

bool

quality_score

Optional quality score (0-100, higher is better).

Type:

float | None

output_id: str

content: str

cycle: int

parent_id: str | None

is_final: bool

quality_score: float | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.reflection.Critique(*, critique_id: str, output_id: str, problems: list[str] = <factory>, strengths: list[str] = <factory>, overall_assessment: str, improvement_suggestions: list[str] = <factory>)

Bases: BaseModel

A critique analyzing an output and identifying problems.

Each critique examines a specific output and identifies areas for improvement.

critique_id

Unique identifier for this critique.

Type:

str

output_id

ID of the output being critiqued.

Type:

str

problems

List of specific problems identified (logical errors, missing info, etc.).

Type:

list[str]

strengths

List of strengths/positive aspects of the output.

Type:

list[str]

overall_assessment

Overall assessment of the output quality.

Type:

str

improvement_suggestions

Specific suggestions for improvement.

Type:

list[str]

critique_id: str

output_id: str

problems: list[str]

strengths: list[str]

overall_assessment: str

improvement_suggestions: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.reflection.CreateOutputItem(*, content: str)

Bases: BaseModel

Input model for the create_output tool.

This is the model that agents use when calling create_output.

content: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.reflection.CritiqueOutputItem(*, output_id: str, problems: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)], strengths: list[str] = <factory>, overall_assessment: str, improvement_suggestions: ~typing.Annotated[list[str], ~annotated_types.MinLen(min_length=1)])

Bases: BaseModel

Input model for the critique_output tool.

output_id: str

problems: list[str]

strengths: list[str]

overall_assessment: str

improvement_suggestions: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.reflection.RefineOutputItem(*, output_id: str, refined_content: str, is_final: bool = False, quality_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None)

Bases: BaseModel

Input model for the refine_output tool.

output_id: str

refined_content: str

is_final: bool

quality_score: float | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.reflection.ReflectionStorage(*, track_usage: bool = False)

Bases: object

Default in-memory reflection storage.

Simple implementation that stores outputs and critiques in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_reflection_toolset, ReflectionStorage

storage = ReflectionStorage()
toolset = create_reflection_toolset(storage=storage)

# After agent runs, access outputs and critiques directly
print(storage.outputs)
print(storage.critiques)

# With metrics tracking
storage = ReflectionStorage(track_usage=True)
toolset = create_reflection_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property outputs: dict[str, ReflectionOutput]

Get the current dictionary of outputs.

property critiques: dict[str, Critique]

Get the current dictionary of critiques.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about reflection operations.

Returns:

Dictionary with output and critique counts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all outputs, critiques, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (output_id or critique_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of outputs and critiques.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for outputs and critiques.

class pydantic_ai_toolsets.toolsets.reflection.ReflectionStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for reflection storage implementations.

Any class that has outputs and critiques properties can be used as storage for the reflection toolset.

class MyCustomStorage:
    def __init__(self):
        self._outputs: dict[str, ReflectionOutput] = {}
        self._critiques: dict[str, Critique] = {}

    @property
    def outputs(self) -> dict[str, ReflectionOutput]:
        return self._outputs

    @outputs.setter
    def outputs(self, value: ReflectionOutput) -> None:
        self._outputs[value.output_id] = value

    @property
    def critiques(self) -> dict[str, Critique]:
        return self._critiques

    @critiques.setter
    def critiques(self, value: Critique) -> None:
        self._critiques[value.critique_id] = value

property outputs: dict[str, ReflectionOutput]

Get the current dictionary of outputs (output_id -> ReflectionOutput).

property critiques: dict[str, Critique]

Get the current dictionary of critiques (critique_id -> Critique).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (output_id or critique_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Self-Refine

Self-refinement toolset for pydantic-ai agents.

Provides self-refinement capabilities for AI agents through iterative feedback and refinement cycles. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_self_refine_toolset, SelfRefineStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_self_refine_toolset()])

# With storage access
storage = SelfRefineStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_self_refine_toolset(storage)])
result = await agent.run("Solve this problem using self-refinement")
print(storage.outputs)  # Access outputs directly

pydantic_ai_toolsets.toolsets.self_refine.create_self_refine_toolset(storage: SelfRefineStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a self-refinement toolset for iterative output improvement.

This toolset provides tools for AI agents to improve outputs through structured feedback and refinement cycles, with support for quality thresholds and iteration limits.

Parameters:

• storage – Optional storage backend. Defaults to in-memory SelfRefineStorage. You can provide a custom storage implementing SelfRefineStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_self_refine_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_self_refine_toolset()])
result = await agent.run("Solve this problem using self-refinement")

Example (with custom storage):

from pydantic_ai_toolsets import create_self_refine_toolset, SelfRefineStorage

storage = SelfRefineStorage()
toolset = create_self_refine_toolset(storage=storage)

# After agent runs, access outputs and feedbacks directly
print(storage.outputs)
print(storage.feedbacks)

pydantic_ai_toolsets.toolsets.self_refine.get_self_refine_system_prompt() → str

Get the system prompt for self-refinement-based reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

class pydantic_ai_toolsets.toolsets.self_refine.RefinementOutput(*, output_id: str, content: str, iteration: Annotated[int, Ge(ge=0)] = 0, parent_id: str | None = None, is_final: bool = False, quality_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, quality_threshold: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, iteration_limit: Annotated[int | None, Ge(ge=1)] = None)

Bases: BaseModel

An output that can be refined through feedback loops.

Each output represents a version of the solution at a specific refinement iteration. Outputs form a chain where each output can receive feedback and be refined into a new output.

output_id

Unique identifier for this output.

Type:

str

content

The actual output content (solution, answer, etc.).

Type:

str

iteration

The refinement iteration number (0 for initial output).

Type:

int

parent_id

ID of the parent output that was refined to create this one (None for initial).

Type:

str | None

is_final

Whether this output is marked as final/satisfactory.

Type:

bool

quality_score

Optional quality score (0-100, higher is better).

Type:

float | None

quality_threshold

Optional quality threshold that must be met (0-100).

Type:

float | None

iteration_limit

Optional maximum number of iterations allowed.

Type:

int | None

output_id: str

content: str

iteration: int

parent_id: str | None

is_final: bool

quality_score: float | None

quality_threshold: float | None

iteration_limit: int | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_refine.Feedback(*, feedback_id: str, output_id: str, feedback_type: FeedbackType, dimension: FeedbackDimension, description: str, suggestion: str, priority: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.5, is_actionable: bool = True)

Bases: BaseModel

Feedback analyzing an output and providing actionable suggestions.

Each feedback examines a specific output and provides structured feedback with specific types (additive, subtractive, transformative, corrective) and dimensions.

feedback_id

Unique identifier for this feedback.

Type:

str

output_id

ID of the output being analyzed.

Type:

str

feedback_type

Type of feedback (additive, subtractive, transformative, corrective).

Type:

FeedbackType

dimension

Dimension being evaluated (factuality, coherence, completeness, style).

Type:

FeedbackDimension

description

Detailed description of the feedback.

Type:

str

suggestion

Specific, actionable suggestion for improvement.

Type:

str

priority

Priority weight (0-1, higher is more important). Used for weighted feedback.

Type:

float

is_actionable

Whether this feedback provides actionable guidance.

Type:

bool

feedback_id: str

output_id: str

feedback_type: FeedbackType

dimension: FeedbackDimension

description: str

suggestion: str

priority: float

is_actionable: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_refine.FeedbackType(*values)

Bases: str, Enum

Types of feedback that can be provided.

ADDITIVE = 'additive'

SUBTRACTIVE = 'subtractive'

TRANSFORMATIVE = 'transformative'

CORRECTIVE = 'corrective'

class pydantic_ai_toolsets.toolsets.self_refine.FeedbackDimension(*values)

Bases: str, Enum

Dimensions for evaluating feedback quality.

FACTUALITY = 'factuality'

COHERENCE = 'coherence'

COMPLETENESS = 'completeness'

STYLE = 'style'

class pydantic_ai_toolsets.toolsets.self_refine.FeedbackItem(*, feedback_type: FeedbackType, dimension: FeedbackDimension, description: str, suggestion: str, priority: Annotated[float, Ge(ge=0.0), Le(le=1.0)] = 0.5)

Bases: BaseModel

A single feedback item within ProvideFeedbackItem.

feedback_type: FeedbackType

dimension: FeedbackDimension

description: str

suggestion: str

priority: float

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_refine.GenerateOutputItem(*, content: str, quality_threshold: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, iteration_limit: Annotated[int | None, Ge(ge=1)] = None)

Bases: BaseModel

Input model for the generate_output tool.

This is the model that agents use when calling generate_output.

content: str

quality_threshold: float | None

iteration_limit: int | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_refine.ProvideFeedbackItem(*, output_id: str, feedback_items: Annotated[list[FeedbackItem], MinLen(min_length=1)], overall_assessment: str, should_continue_refining: bool = True)

Bases: BaseModel

Input model for the provide_feedback tool.

output_id: str

feedback_items: list[FeedbackItem]

overall_assessment: str

should_continue_refining: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_refine.RefineOutputItem(*, output_id: str, refined_content: str, is_final: bool = False, quality_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None)

Bases: BaseModel

Input model for the refine_output tool.

output_id: str

refined_content: str

is_final: bool

quality_score: float | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_refine.SelfRefineStorage(*, track_usage: bool = False)

Bases: object

Default in-memory self-refinement storage.

Simple implementation that stores outputs and feedbacks in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_self_refine_toolset, SelfRefineStorage

storage = SelfRefineStorage()
toolset = create_self_refine_toolset(storage=storage)

# After agent runs, access outputs and feedbacks directly
print(storage.outputs)
print(storage.feedbacks)

# With metrics tracking
storage = SelfRefineStorage(track_usage=True)
toolset = create_self_refine_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property outputs: dict[str, RefinementOutput]

Get the current dictionary of outputs.

property feedbacks: dict[str, Feedback]

Get the current dictionary of feedbacks.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about self-refinement operations.

Returns:

Dictionary with output and feedback counts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all outputs, feedbacks, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (output_id or feedback_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of outputs and feedbacks.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for outputs and feedbacks.

class pydantic_ai_toolsets.toolsets.self_refine.SelfRefineStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for self-refinement storage implementations.

Any class that has outputs and feedbacks properties can be used as storage for the self-refinement toolset.

class MyCustomStorage:
    def __init__(self):
        self._outputs: dict[str, RefinementOutput] = {}
        self._feedbacks: dict[str, Feedback] = {}

    @property
    def outputs(self) -> dict[str, RefinementOutput]:
        return self._outputs

    @outputs.setter
    def outputs(self, value: RefinementOutput) -> None:
        self._outputs[value.output_id] = value

    @property
    def feedbacks(self) -> dict[str, Feedback]:
        return self._feedbacks

    @feedbacks.setter
    def feedbacks(self, value: Feedback) -> None:
        self._feedbacks[value.feedback_id] = value

property outputs: dict[str, RefinementOutput]

Get the current dictionary of outputs (output_id -> RefinementOutput).

property feedbacks: dict[str, Feedback]

Get the current dictionary of feedbacks (feedback_id -> Feedback).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (output_id or feedback_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Self-Ask

Self-ask toolset for pydantic-ai agents.

Provides self-ask reasoning capabilities for AI agents. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_self_ask_toolset, SelfAskStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_self_ask_toolset()])

# With storage access
storage = SelfAskStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_self_ask_toolset(storage)])
result = await agent.run("What was the population of the city where the 2016 Summer Olympics were held?")
print(storage.questions)  # Access questions directly
print(storage.answers)  # Access answers directly
print(storage.final_answers)  # Access final answers directly

pydantic_ai_toolsets.toolsets.self_ask.create_self_ask_toolset(storage: SelfAskStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a self-ask toolset for question decomposition.

This toolset provides tools for AI agents to decompose complex questions into simpler sub-questions, answer them sequentially, and compose final answers.

Parameters:

• storage – Optional storage backend. Defaults to in-memory SelfAskStorage. You can provide a custom storage implementing SelfAskStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection in storage.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_self_ask_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_self_ask_toolset()])
result = await agent.run("What was the population of the city where the 2016 Summer Olympics were held?")

Example (with custom storage):

from pydantic_ai_toolsets import create_self_ask_toolset, SelfAskStorage

storage = SelfAskStorage()
toolset = create_self_ask_toolset(storage=storage)

# After agent runs, access questions, answers, and final answers directly
print(storage.questions)
print(storage.answers)
print(storage.final_answers)

pydantic_ai_toolsets.toolsets.self_ask.get_self_ask_system_prompt() → str

Get the system prompt for self-ask reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

class pydantic_ai_toolsets.toolsets.self_ask.Question(*, question_id: str, question_text: str, is_main: bool = False, parent_question_id: str | None = None, depth: Annotated[int, Ge(ge=0), Le(le=3)] = 0, status: QuestionStatus = QuestionStatus.PENDING)

Bases: BaseModel

A question being decomposed in the self-ask process.

Questions form a tree structure where the main question is the root (depth 0), and sub-questions are children (depth 1-3). Each question can spawn sub-questions up to the maximum depth of 3.

question_id

Unique identifier for this question.

Type:

str

question_text

The actual question text.

Type:

str

is_main

Whether this is the main question (depth 0).

Type:

bool

parent_question_id

ID of the parent question (None for main question).

Type:

str | None

depth

Depth level in the question tree (0 for main, 1-3 for sub-questions).

Type:

int

status

Current status of the question (pending, answered, composed).

Type:

QuestionStatus

question_id: str

question_text: str

is_main: bool

parent_question_id: str | None

depth: int

status: QuestionStatus

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_ask.Answer(*, answer_id: str, question_id: str, answer_text: str, confidence_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, requires_followup: bool = False)

Bases: BaseModel

An answer to a question or sub-question.

Each answer corresponds to a specific question and can be used to answer parent questions or compose the final answer.

answer_id

Unique identifier for this answer.

Type:

str

question_id

ID of the question this answer addresses.

Type:

str

answer_text

The actual answer content.

Type:

str

confidence_score

Optional confidence score (0-100, higher is more confident).

Type:

float | None

requires_followup

Whether this answer needs further sub-questions.

Type:

bool

answer_id: str

question_id: str

answer_text: str

confidence_score: float | None

requires_followup: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_ask.FinalAnswer(*, final_answer_id: str, main_question_id: str, final_answer_text: str, composed_from_answers: list[str] = <factory>, is_complete: bool = True)

Bases: BaseModel

The final composed answer to the main question.

The final answer is composed from answers to sub-questions and represents the complete solution to the original question.

final_answer_id

Unique identifier for this final answer.

Type:

str

main_question_id

ID of the main question this answers.

Type:

str

final_answer_text

The composed final answer content.

Type:

str

composed_from_answers

List of answer IDs used to compose this answer.

Type:

list[str]

is_complete

Whether the final answer is complete and ready.

Type:

bool

final_answer_id: str

main_question_id: str

final_answer_text: str

composed_from_answers: list[str]

is_complete: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_ask.QuestionStatus(*values)

Bases: str, Enum

Status of a question in the self-ask process.

PENDING = 'pending'

ANSWERED = 'answered'

COMPOSED = 'composed'

class pydantic_ai_toolsets.toolsets.self_ask.AskMainQuestionItem(*, question_text: str)

Bases: BaseModel

Input model for the ask_main_question tool.

This is the model that agents use when calling ask_main_question.

question_text: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_ask.AskSubQuestionItem(*, parent_question_id: str, sub_question_text: str, reasoning: str)

Bases: BaseModel

Input model for the ask_sub_question tool.

parent_question_id: str

sub_question_text: str

reasoning: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_ask.AnswerQuestionItem(*, question_id: str, answer_text: str, confidence_score: Annotated[float | None, Ge(ge=0.0), Le(le=100.0)] = None, requires_followup: bool = False)

Bases: BaseModel

Input model for the answer_question tool.

question_id: str

answer_text: str

confidence_score: float | None

requires_followup: bool

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_ask.ComposeFinalAnswerItem(*, main_question_id: str, final_answer_text: str, answer_ids_used: Annotated[list[str], MinLen(min_length=1)])

Bases: BaseModel

Input model for the compose_final_answer tool.

main_question_id: str

final_answer_text: str

answer_ids_used: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.self_ask.SelfAskStorage(*, track_usage: bool = False)

Bases: object

Default in-memory self-ask storage.

Simple implementation that stores questions, answers, and final answers in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_self_ask_toolset, SelfAskStorage

storage = SelfAskStorage()
toolset = create_self_ask_toolset(storage=storage)

# After agent runs, access questions, answers, and final answers directly
print(storage.questions)
print(storage.answers)
print(storage.final_answers)

# With metrics tracking
storage = SelfAskStorage(track_usage=True)
toolset = create_self_ask_toolset(storage=storage)
print(storage.metrics.total_tokens())

MAX_DEPTH: int = 3

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property questions: dict[str, Question]

Get the current dictionary of questions.

property answers: dict[str, Answer]

Get the current dictionary of answers.

property final_answers: dict[str, FinalAnswer]

Get the current dictionary of final answers.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about self-ask operations.

Returns:

Dictionary with question, answer, and final answer counts, plus max depth reached.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all questions, answers, final answers, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (question_id, answer_id, or final_answer_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of questions, answers, and final answers.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for questions, answers, and final answers.

class pydantic_ai_toolsets.toolsets.self_ask.SelfAskStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for self-ask storage implementations.

Any class that has questions, answers, and final_answers properties can be used as storage for the self-ask toolset.

class MyCustomStorage:
    def __init__(self):
        self._questions: dict[str, Question] = {}
        self._answers: dict[str, Answer] = {}
        self._final_answers: dict[str, FinalAnswer] = {}

    @property
    def questions(self) -> dict[str, Question]:
        return self._questions

    @questions.setter
    def questions(self, value: Question) -> None:
        self._questions[value.question_id] = value

    @property
    def answers(self) -> dict[str, Answer]:
        return self._answers

    @answers.setter
    def answers(self, value: Answer) -> None:
        self._answers[value.answer_id] = value

    @property
    def final_answers(self) -> dict[str, FinalAnswer]:
        return self._final_answers

    @final_answers.setter
    def final_answers(self, value: FinalAnswer) -> None:
        self._final_answers[value.final_answer_id] = value

property questions: dict[str, Question]

Get the current dictionary of questions (question_id -> Question).

property answers: dict[str, Answer]

Get the current dictionary of answers (answer_id -> Answer).

property final_answers: dict[str, FinalAnswer]

Get the current dictionary of final answers (final_answer_id -> FinalAnswer).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (question_id, answer_id, or final_answer_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Multi-Agent Toolsets

Multi-Persona Analysis

Multi-persona toolset for pydantic-ai agents.

Provides multi-persona analysis capabilities for AI agents through diverse viewpoints and perspectives. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_persona_toolset, PersonaStorage

# Simple usage
agent = Agent("openai:gpt-4", toolsets=[create_persona_toolset()])
result = await agent.run("Analyze: Should we invest in this startup?")

# With storage access
storage = PersonaStorage()
toolset = create_persona_toolset(storage=storage)

# After agent runs, access persona state directly
print(storage.session)
print(storage.personas)
print(storage.responses)

pydantic_ai_toolsets.toolsets.multi_persona_analysis.create_persona_toolset(storage: PersonaStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a multi-persona toolset for diverse perspective analysis.

This toolset provides tools for AI agents to adopt multiple personas and synthesize diverse perspectives on problems.

Parameters:

• storage – Optional storage backend. Defaults to in-memory PersonaStorage. You can provide a custom storage implementing PersonaStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection in storage.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_persona_toolset

agent = Agent("openai:gpt-4", toolsets=[create_persona_toolset()])
result = await agent.run("Analyze: Should we invest in this startup?")

Example (with storage access):

from pydantic_ai_toolsets import create_persona_toolset, PersonaStorage

storage = PersonaStorage()
toolset = create_persona_toolset(storage=storage)

# After agent runs, access persona state directly
print(storage.session)
print(storage.personas)
print(storage.responses)

# With metrics tracking
storage = PersonaStorage(track_usage=True)
toolset = create_persona_toolset(storage=storage)
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.multi_persona_analysis.get_persona_system_prompt(storage: PersonaStorageProtocol | None = None) → str

Generate dynamic system prompt section for personas.

Parameters:

storage – Optional storage to read current session from.

Returns:

System prompt section with current session info, or base prompt if no session.

class pydantic_ai_toolsets.toolsets.multi_persona_analysis.Persona(*, persona_id: str, name: str, persona_type: ~typing.Literal['expert', 'thinking_style', 'stakeholder'], description: str, expertise_areas: list[str] = <factory>)

Bases: BaseModel

A persona representing a distinct viewpoint or expertise.

Personas can be expert personas (domain specialists), thinking style personas (cognitive approaches), or stakeholder personas (interested parties).

persona_id

Unique identifier for this persona.

Type:

str

name

Display name of the persona (e.g., “Clinical Doctor”, “Analytical Persona”).

Type:

str

persona_type

Type of persona - expert, thinking_style, or stakeholder.

Type:

Literal[‘expert’, ‘thinking_style’, ‘stakeholder’]

description

Detailed description of the persona’s background, expertise, and perspective.

Type:

str

expertise_areas

List of specific areas of expertise or focus.

Type:

list[str]

persona_id: str

name: str

persona_type: Literal['expert', 'thinking_style', 'stakeholder']

description: str

expertise_areas: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_analysis.PersonaResponse(*, response_id: str, persona_id: str, content: str, references: list[str] = <factory>, round_number: Annotated[int, ~annotated_types.Ge(ge=0)] = 0)

Bases: BaseModel

A response from a persona to a problem or question.

Each persona provides independent analysis from their unique perspective. Responses can reference other responses in interactive dialogue patterns.

response_id

Unique identifier for this response.

Type:

str

persona_id

ID of the persona providing this response.

Type:

str

content

The persona’s analysis, insights, or perspective.

Type:

str

references

List of response IDs that this response references or responds to.

Type:

list[str]

round_number

Round number in interactive dialogue (0 for initial responses).

Type:

int

response_id: str

persona_id: str

content: str

references: list[str]

round_number: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_analysis.PersonaSession(*, session_id: str, problem: str, process_type: Literal['sequential', 'interactive', 'devils_advocate'], status: Literal['active', 'completed', 'synthesized'] = 'active', synthesis: str | None = None, max_rounds: Annotated[int, Ge(ge=1)] = 3, current_round: Annotated[int, Ge(ge=0)] = 0)

Bases: BaseModel

Complete persona session.

Tracks the overall state of a multi-persona analysis session, including the problem/question, personas, responses, and synthesis.

session_id

Unique identifier for this session.

Type:

str

problem

The problem or question being analyzed.

Type:

str

process_type

Type of process - sequential, interactive, or devils_advocate.

Type:

Literal[‘sequential’, ‘interactive’, ‘devils_advocate’]

status

Current status of the session (active, completed, synthesized).

Type:

Literal[‘active’, ‘completed’, ‘synthesized’]

synthesis

Final synthesis text (if synthesized).

Type:

str | None

max_rounds

Maximum number of dialogue rounds (for interactive/devils_advocate).

Type:

int

current_round

Current round number (0-indexed).

Type:

int

session_id: str

problem: str

process_type: Literal['sequential', 'interactive', 'devils_advocate']

status: Literal['active', 'completed', 'synthesized']

synthesis: str | None

max_rounds: int

current_round: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_analysis.CreatePersonaItem(*, name: str, persona_type: ~typing.Literal['expert', 'thinking_style', 'stakeholder'], description: str, expertise_areas: list[str] = <factory>)

Bases: BaseModel

Input model for the create_persona tool.

name: str

persona_type: Literal['expert', 'thinking_style', 'stakeholder']

description: str

expertise_areas: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_analysis.AddPersonaResponseItem(*, persona_id: str, content: str, references: list[str] = <factory>)

Bases: BaseModel

Input model for the add_persona_response tool.

persona_id: str

content: str

references: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_analysis.SynthesizeItem(*, synthesis_content: str, key_insights: list[str] = <factory>, conflicts_resolved: list[str] = <factory>)

Bases: BaseModel

Input model for the synthesize tool.

synthesis_content: str

key_insights: list[str]

conflicts_resolved: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_analysis.InitiatePersonaSessionItem(*, problem: str, process_type: Literal['sequential', 'interactive', 'devils_advocate'], max_rounds: Annotated[int, Ge(ge=1)] = 3)

Bases: BaseModel

Input model for the initiate_persona_session tool.

problem: str

process_type: Literal['sequential', 'interactive', 'devils_advocate']

max_rounds: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_analysis.PersonaStorage(*, track_usage: bool = False)

Bases: object

Default in-memory persona storage.

Simple implementation that stores persona sessions, personas, and responses in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_persona_toolset, PersonaStorage

storage = PersonaStorage()
toolset = create_persona_toolset(storage=storage)

# After agent runs, access persona state directly
print(storage.session)
print(storage.personas)
print(storage.responses)

# With metrics tracking
storage = PersonaStorage(track_usage=True)
toolset = create_persona_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property session: PersonaSession | None

Get the current persona session.

property personas: dict[str, Persona]

Get all personas (persona_id -> Persona).

property responses: dict[str, PersonaResponse]

Get all responses (response_id -> PersonaResponse).

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about persona operations.

Returns:

Dictionary with persona and response counts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all persona data and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (persona_id, response_id, or session_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of personas, responses, and session.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for personas, responses, and session.

class pydantic_ai_toolsets.toolsets.multi_persona_analysis.PersonaStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for persona storage implementations.

Any class that has session, personas, and responses properties can be used as storage for the persona toolset.

class MyCustomStorage:
    def __init__(self):
        self._session: PersonaSession | None = None
        self._personas: dict[str, Persona] = {}
        self._responses: dict[str, PersonaResponse] = {}

    @property
    def session(self) -> PersonaSession | None:
        return self._session

    @session.setter
    def session(self, value: PersonaSession) -> None:
        self._session = value

    @property
    def personas(self) -> dict[str, Persona]:
        return self._personas

    @personas.setter
    def personas(self, value: Persona) -> None:
        self._personas[value.persona_id] = value

    @property
    def responses(self) -> dict[str, PersonaResponse]:
        return self._responses

    @responses.setter
    def responses(self, value: PersonaResponse) -> None:
        self._responses[value.response_id] = value

property session: PersonaSession | None

Get the current persona session.

property personas: dict[str, Persona]

Get all personas (persona_id -> Persona).

property responses: dict[str, PersonaResponse]

Get all responses (response_id -> PersonaResponse).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (persona_id, response_id, or session_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Multi-Persona Debate

Persona debate toolset for pydantic-ai agents.

Provides structured debate capabilities between multiple personas for AI agents through multi-persona argumentation. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_persona_debate_toolset, PersonaDebateStorage

# Simple usage
agent = Agent("openai:gpt-4", toolsets=[create_persona_debate_toolset()])
result = await agent.run("Debate: Should we adopt microservices?")

# With multi-agent orchestration
storage = PersonaDebateStorage()
toolset = create_persona_debate_toolset(
    storage=storage,
    agent_model="openai:gpt-4",
    auto_orchestrate=True,
)

orchestrator = Agent("openai:gpt-4", toolsets=[toolset])
result = await orchestrator.run("Start a debate on microservices")
print(storage.session)  # Access debate state directly

pydantic_ai_toolsets.toolsets.multi_persona_debate.create_persona_debate_toolset(storage: PersonaDebateStorageProtocol | None = None, *, id: str | None = None, agent_model: str | None = None, agent_configs: dict[str, dict[str, Any]] | None = None, auto_orchestrate: bool = False, track_usage: bool = False) → FunctionToolset[Any]

Create a persona debate toolset for multi-persona structured debates.

This toolset provides tools for AI agents to engage in structured debates between multiple personas, with support for creating and orchestrating multiple agent instances.

Parameters:

• storage – Optional storage backend. Defaults to in-memory PersonaDebateStorage. You can provide a custom storage implementing PersonaDebateStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• agent_model – Default model string for creating agents (e.g., “openai:gpt-4”). Required if agent_configs not provided or if auto_orchestrate=True.

• agent_configs –

  Per-persona agent configurations: {

  ”persona_id_1”: {“model”: “openai:gpt-4”, “system_prompt”: “…”}, “persona_id_2”: {“model”: “openai:gpt-4”, “system_prompt”: “…”},

  }

• auto_orchestrate – If True, tools automatically orchestrate agent interactions.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_persona_debate_toolset

agent = Agent("openai:gpt-4", toolsets=[create_persona_debate_toolset()])
result = await agent.run("Debate: Should we adopt microservices?")

Example (with multi-agent orchestration):

from pydantic_ai_toolsets import create_persona_debate_toolset, PersonaDebateStorage

storage = PersonaDebateStorage()
toolset = create_persona_debate_toolset(
    storage=storage,
    agent_model="openai:gpt-4",
    auto_orchestrate=True,
)

orchestrator = Agent("openai:gpt-4", toolsets=[toolset])
result = await orchestrator.run("Start a debate on microservices")

pydantic_ai_toolsets.toolsets.multi_persona_debate.get_persona_debate_system_prompt() → str

Get the system prompt for persona debate-based reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

class pydantic_ai_toolsets.toolsets.multi_persona_debate.Persona(*, persona_id: str, name: str, persona_type: ~typing.Literal['expert', 'thinking_style', 'stakeholder'], description: str, expertise_areas: list[str] = <factory>)

Bases: BaseModel

A persona representing a distinct viewpoint or expertise.

Personas can be expert personas (domain specialists), thinking style personas (cognitive approaches), or stakeholder personas (interested parties).

persona_id

Unique identifier for this persona.

Type:

str

name

Display name of the persona (e.g., “Clinical Doctor”, “Analytical Persona”).

Type:

str

persona_type

Type of persona - expert, thinking_style, or stakeholder.

Type:

Literal[‘expert’, ‘thinking_style’, ‘stakeholder’]

description

Detailed description of the persona’s background, expertise, and perspective.

Type:

str

expertise_areas

List of specific areas of expertise or focus.

Type:

list[str]

persona_id: str

name: str

persona_type: Literal['expert', 'thinking_style', 'stakeholder']

description: str

expertise_areas: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.PersonaPosition(*, position_id: str, persona_id: str, round_number: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], content: str, evidence: list[str] = <factory>, critiques_addressed: list[str] = <factory>, parent_position_id: str | None = None)

Bases: BaseModel

A position/argument made by a persona in a debate.

Each position represents an argument made by a specific persona in a specific round. Positions can be critiqued, defended, and agreed with by other personas.

position_id

Unique identifier for this position.

Type:

str

persona_id

ID of the persona making this position.

Type:

str

round_number

The debate round in which this position was made.

Type:

int

content

The actual argument content.

Type:

str

evidence

List of evidence citations (for evidence-based debates).

Type:

list[str]

critiques_addressed

List of critique IDs that this position addresses.

Type:

list[str]

parent_position_id

ID of parent position if this is a defense/refinement.

Type:

str | None

position_id: str

persona_id: str

round_number: int

content: str

evidence: list[str]

critiques_addressed: list[str]

parent_position_id: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.PersonaCritique(*, critique_id: str, target_position_id: str, persona_id: str, round_number: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], content: str, specific_points: list[str] = <factory>)

Bases: BaseModel

A critique of a position made by a persona.

Each critique identifies weaknesses or challenges to a specific position. Critiques guide the defense and refinement process.

critique_id

Unique identifier for this critique.

Type:

str

target_position_id

ID of the position being critiqued.

Type:

str

persona_id

ID of the persona making this critique.

Type:

str

round_number

The debate round in which this critique was made.

Type:

int

content

The critique content.

Type:

str

specific_points

List of specific weaknesses or points raised.

Type:

list[str]

critique_id: str

target_position_id: str

persona_id: str

round_number: int

content: str

specific_points: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.PersonaAgreement(*, agreement_id: str, target_position_id: str, persona_id: str, round_number: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)], content: str, reasoning: list[str] = <factory>)

Bases: BaseModel

An agreement by a persona with another persona’s position.

Personas can agree with positions made by other personas, providing reasoning for their agreement. This allows for coalition-building and consensus formation.

agreement_id

Unique identifier for this agreement.

Type:

str

target_position_id

ID of the position being agreed with.

Type:

str

persona_id

ID of the persona agreeing with the position.

Type:

str

round_number

The debate round in which this agreement was made.

Type:

int

content

The agreement content explaining why the persona agrees.

Type:

str

reasoning

Specific reasons or points that led to the agreement.

Type:

list[str]

agreement_id: str

target_position_id: str

persona_id: str

round_number: int

content: str

reasoning: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.PersonaDebateSession(*, debate_id: str, topic: str, max_rounds: Annotated[int, Ge(ge=1)] = 5, current_round: Annotated[int, Ge(ge=0)] = 0, status: Literal['active', 'completed', 'resolved'] = 'active', resolution: str | None = None, winner_persona_id: str | None = None, resolution_type: Literal['synthesis', 'winner', 'consensus'] | None = None)

Bases: BaseModel

Complete persona debate session.

Tracks the overall state of a debate between personas, including topic, personas, rounds, positions, critiques, agreements, and resolution.

debate_id

Unique identifier for this debate session.

Type:

str

topic

The debate topic/question.

Type:

str

max_rounds

Maximum number of debate rounds.

Type:

int

current_round

Current round number (0-indexed).

Type:

int

status

Current status of the debate (active, completed, resolved).

Type:

Literal[‘active’, ‘completed’, ‘resolved’]

resolution

Final resolution text (if resolved).

Type:

str | None

winner_persona_id

ID of winning persona (if resolved with winner).

Type:

str | None

resolution_type

Type of resolution (synthesis, winner, consensus).

Type:

Literal[‘synthesis’, ‘winner’, ‘consensus’] | None

debate_id: str

topic: str

max_rounds: int

current_round: int

status: Literal['active', 'completed', 'resolved']

resolution: str | None

winner_persona_id: str | None

resolution_type: Literal['synthesis', 'winner', 'consensus'] | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.CreatePersonaItem(*, name: str, persona_type: ~typing.Literal['expert', 'thinking_style', 'stakeholder'], description: str, expertise_areas: list[str] = <factory>)

Bases: BaseModel

Input model for the create_persona tool.

name: str

persona_type: Literal['expert', 'thinking_style', 'stakeholder']

description: str

expertise_areas: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.ProposePositionItem(*, persona_id: str, content: str, evidence: list[str] = <factory>)

Bases: BaseModel

Input model for the propose_position tool.

persona_id: str

content: str

evidence: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.CritiquePositionItem(*, target_position_id: str, persona_id: str, content: str, specific_points: Annotated[list[str], MinLen(min_length=1)])

Bases: BaseModel

Input model for the critique_position tool.

target_position_id: str

persona_id: str

content: str

specific_points: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.AgreeWithPositionItem(*, target_position_id: str, persona_id: str, content: str, reasoning: list[str] = <factory>)

Bases: BaseModel

Input model for the agree_with_position tool.

target_position_id: str

persona_id: str

content: str

reasoning: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.DefendPositionItem(*, position_id: str, persona_id: str, content: str, critiques_addressed: list[str] = <factory>, evidence: list[str] = <factory>)

Bases: BaseModel

Input model for the defend_position tool.

position_id: str

persona_id: str

content: str

critiques_addressed: list[str]

evidence: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.ResolveDebateItem(*, resolution_type: ~typing.Literal['synthesis', 'winner', 'consensus'], resolution_content: str, winner_persona_id: str | None = None, synthesis_elements: list[str] = <factory>, consensus_points: list[str] = <factory>)

Bases: BaseModel

Input model for the resolve_debate tool.

resolution_type: Literal['synthesis', 'winner', 'consensus']

resolution_content: str

winner_persona_id: str | None

synthesis_elements: list[str]

consensus_points: list[str]

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.InitiatePersonaDebateItem(*, topic: str, max_rounds: Annotated[int, Ge(ge=1)] = 5)

Bases: BaseModel

Input model for the initiate_persona_debate tool.

topic: str

max_rounds: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.OrchestrateRoundItem(*, round_number: Annotated[int, Ge(ge=1)])

Bases: BaseModel

Input model for the orchestrate_round tool.

round_number: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.multi_persona_debate.PersonaDebateStorage(*, track_usage: bool = False)

Bases: object

Default in-memory persona debate storage.

Simple implementation that stores persona debate sessions, personas, positions, critiques, and agreements in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_persona_debate_toolset, PersonaDebateStorage

storage = PersonaDebateStorage()
toolset = create_persona_debate_toolset(storage=storage)

# After agent runs, access debate state directly
print(storage.session)
print(storage.personas)
print(storage.positions)
print(storage.critiques)
print(storage.agreements)

# With metrics tracking
storage = PersonaDebateStorage(track_usage=True)
toolset = create_persona_debate_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property session: PersonaDebateSession | None

Get the current persona debate session.

property personas: dict[str, Persona]

Get all personas (persona_id -> Persona).

property positions: dict[str, PersonaPosition]

Get all positions (position_id -> PersonaPosition).

property critiques: dict[str, PersonaCritique]

Get all critiques (critique_id -> PersonaCritique).

property agreements: dict[str, PersonaAgreement]

Get all agreements (agreement_id -> PersonaAgreement).

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about persona debate operations.

Returns:

Dictionary with debate counts and metrics.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all debate data and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (persona_id, position_id, critique_id, agreement_id, or session_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of personas, positions, critiques, and agreements.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for personas, positions, critiques, agreements, and session.

class pydantic_ai_toolsets.toolsets.multi_persona_debate.PersonaDebateStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for persona debate storage implementations.

Any class that has session, personas, positions, critiques, and agreements properties can be used as storage for the persona debate toolset.

class MyCustomStorage:
    def __init__(self):
        self._session: PersonaDebateSession | None = None
        self._personas: dict[str, Persona] = {}
        self._positions: dict[str, PersonaPosition] = {}
        self._critiques: dict[str, PersonaCritique] = {}
        self._agreements: dict[str, PersonaAgreement] = {}

    @property
    def session(self) -> PersonaDebateSession | None:
        return self._session

    @session.setter
    def session(self, value: PersonaDebateSession) -> None:
        self._session = value

    @property
    def personas(self) -> dict[str, Persona]:
        return self._personas

    @personas.setter
    def personas(self, value: Persona) -> None:
        self._personas[value.persona_id] = value

    @property
    def positions(self) -> dict[str, PersonaPosition]:
        return self._positions

    @positions.setter
    def positions(self, value: PersonaPosition) -> None:
        self._positions[value.position_id] = value

    @property
    def critiques(self) -> dict[str, PersonaCritique]:
        return self._critiques

    @critiques.setter
    def critiques(self, value: PersonaCritique) -> None:
        self._critiques[value.critique_id] = value

    @property
    def agreements(self) -> dict[str, PersonaAgreement]:
        return self._agreements

    @agreements.setter
    def agreements(self, value: PersonaAgreement) -> None:
        self._agreements[value.agreement_id] = value

property session: PersonaDebateSession | None

Get the current persona debate session.

property personas: dict[str, Persona]

Get all personas (persona_id -> Persona).

property positions: dict[str, PersonaPosition]

Get all positions (position_id -> PersonaPosition).

property critiques: dict[str, PersonaCritique]

Get all critiques (critique_id -> PersonaCritique).

property agreements: dict[str, PersonaAgreement]

Get all agreements (agreement_id -> PersonaAgreement).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (persona_id, position_id, critique_id, or agreement_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Utility Toolsets

Search

Search toolset for pydantic-ai agents.

Provides web search and content extraction capabilities for AI agents. Compatible with any pydantic-ai agent - requires FIRECRAWL_API_KEY environment variable.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_search_toolset, SearchStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_search_toolset()])

# With storage access
storage = SearchStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_search_toolset(storage)])
result = await agent.run("Search for information about Python async")
print(storage.search_results)  # Access search results directly

pydantic_ai_toolsets.toolsets.search.create_search_toolset(storage: SearchStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a search toolset for web search and content extraction.

This toolset provides tools for AI agents to search the web and extract content from webpages, with support for tracking search history and extracted content.

Parameters:

• storage – Optional storage backend. Defaults to in-memory SearchStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_search_toolset, SearchStorage

# With storage and metrics
storage = SearchStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_search_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.search.get_search_system_prompt() → str

Get the system prompt for search-based reasoning.

Returns:

System prompt string that can be used with pydantic-ai agents.

class pydantic_ai_toolsets.toolsets.search.SearchResult(*, result_id: str, query: str, title: str, url: str, description: str | None = None, timestamp: str | None = None, source_type: SearchSource = SearchSource.WEB, date: str | None = None, image_url: str | None = None, image_width: int | None = None, image_height: int | None = None)

Bases: BaseModel

A single search result from a web search.

result_id

Unique identifier for this search result.

Type:

str

query

The search query that produced this result.

Type:

str

title

Title of the search result.

Type:

str

url

URL of the search result.

Type:

str

description

Description or snippet of the search result.

Type:

str | None

timestamp

Timestamp when the search was performed.

Type:

str | None

source_type

Type of search source (web, news, or images).

Type:

SearchSource

date

Date string for news results (e.g., “3 months ago”).

Type:

str | None

image_url

Direct image URL for image results.

Type:

str | None

image_width

Image width in pixels for image results.

Type:

int | None

image_height

Image height in pixels for image results.

Type:

int | None

result_id: str

query: str

title: str

url: str

description: str | None

timestamp: str | None

source_type: SearchSource

date: str | None

image_url: str | None

image_width: int | None

image_height: int | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.search.ExtractedContent(*, content_id: str, url: str, content: str, output_format: OutputFormat = OutputFormat.TEXT, timestamp: str | None = None)

Bases: BaseModel

Extracted content from a webpage.

content_id

Unique identifier for this extracted content.

Type:

str

url

URL of the webpage that was extracted.

Type:

str

content

The extracted content text.

Type:

str

output_format

Format of the extracted content (txt or markdown).

Type:

OutputFormat

timestamp

Timestamp when the extraction was performed.

Type:

str | None

content_id: str

url: str

content: str

output_format: OutputFormat

timestamp: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.search.OutputFormat(*values)

Bases: str, Enum

Output format for extracted web content.

TEXT = 'txt'

MARKDOWN = 'markdown'

class pydantic_ai_toolsets.toolsets.search.SearchWebItem(*, query: str, limit: Annotated[int, Ge(ge=1), Le(le=5)] = 5)

Bases: BaseModel

Input model for the search_web tool.

This is the model that agents use when calling search_web.

query: str

limit: int

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.search.SearchNewsItem(*, query: str, limit: Annotated[int, Ge(ge=1), Le(le=50)] = 5, time_filter: TimeFilter | None = None, custom_date_min: str | None = None, custom_date_max: str | None = None)

Bases: BaseModel

Input model for the search_news tool.

This is the model that agents use when calling search_news.

query: str

limit: int

time_filter: TimeFilter | None

custom_date_min: str | None

custom_date_max: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.search.SearchImagesItem(*, query: str, limit: Annotated[int, Ge(ge=1), Le(le=50)] = 5, min_width: Annotated[int | None, Ge(ge=1)] = None, min_height: Annotated[int | None, Ge(ge=1)] = None, exact_width: Annotated[int | None, Ge(ge=1)] = None, exact_height: Annotated[int | None, Ge(ge=1)] = None)

Bases: BaseModel

Input model for the search_images tool.

This is the model that agents use when calling search_images.

query: str

limit: int

min_width: int | None

min_height: int | None

exact_width: int | None

exact_height: int | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.search.ExtractWebContentItem(*, url: str, output_format: OutputFormat = OutputFormat.TEXT)

Bases: BaseModel

Input model for the extract_web_content tool.

This is the model that agents use when calling extract_web_content. Note: Content extraction only works with web and news search results, not image results.

url: str

output_format: OutputFormat

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.search.SearchSource(*values)

Bases: str, Enum

Source type for search results.

WEB = 'web'

NEWS = 'news'

IMAGES = 'images'

class pydantic_ai_toolsets.toolsets.search.TimeFilter(*values)

Bases: str, Enum

Time filter for news search.

PAST_HOUR = 'qdr:h'

PAST_DAY = 'qdr:d'

PAST_WEEK = 'qdr:w'

PAST_MONTH = 'qdr:m'

PAST_YEAR = 'qdr:y'

CUSTOM = 'custom'

class pydantic_ai_toolsets.toolsets.search.SearchStorage(*, track_usage: bool = False)

Bases: object

Default in-memory search storage.

Simple implementation that stores search results and extracted contents in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_search_toolset, SearchStorage

storage = SearchStorage()
toolset = create_search_toolset(storage=storage)

# After agent runs, access search results and extracted contents directly
print(storage.search_results)
print(storage.extracted_contents)

# With metrics tracking
storage = SearchStorage(track_usage=True)
toolset = create_search_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property search_results: dict[str, SearchResult]

Get the current dictionary of search results.

property extracted_contents: dict[str, ExtractedContent]

Get the current dictionary of extracted contents.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about search operations.

Returns:

Dictionary with search and extraction counts.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all search results, extracted contents, and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (search_result_id or extracted_content_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of search results and extracted contents.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for search results and extracted contents.

class pydantic_ai_toolsets.toolsets.search.SearchStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for search storage implementations.

Any class that has search_results and extracted_contents properties can be used as storage for the search toolset.

class MyCustomStorage:
    def __init__(self):
        self._search_results: dict[str, SearchResult] = {}
        self._extracted_contents: dict[str, ExtractedContent] = {}

    @property
    def search_results(self) -> dict[str, SearchResult]:
        return self._search_results

    @search_results.setter
    def search_results(self, value: SearchResult) -> None:
        self._search_results[value.result_id] = value

    @property
    def extracted_contents(self) -> dict[str, ExtractedContent]:
        return self._extracted_contents

    @extracted_contents.setter
    def extracted_contents(self, value: ExtractedContent) -> None:
        self._extracted_contents[value.content_id] = value

property search_results: dict[str, SearchResult]

Get the current dictionary of search results (result_id -> SearchResult).

property extracted_contents: dict[str, ExtractedContent]

Get the current dictionary of extracted contents (content_id -> ExtractedContent).

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (result_id or content_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Todo

Todo toolset for pydantic-ai agents.

Provides task planning and tracking capabilities for AI agents. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_todo_toolset, TodoStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_todo_toolset()])

# With storage access
storage = TodoStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_todo_toolset(storage)])
result = await agent.run("Create 3 tasks")
print(storage.todos)  # Access todos directly

pydantic_ai_toolsets.toolsets.to_do.create_todo_toolset(storage: TodoStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a todo toolset for task management.

This toolset provides read_todos and write_todos tools for AI agents to track and manage tasks during a session.

Parameters:

• storage – Optional storage backend. Defaults to in-memory TodoStorage.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_todo_toolset, TodoStorage

# With storage and metrics
storage = TodoStorage(track_usage=True)
agent = Agent("openai:gpt-4.1", toolsets=[create_todo_toolset(storage)])
print(storage.metrics.total_tokens())

pydantic_ai_toolsets.toolsets.to_do.get_todo_system_prompt(storage: TodoStorageProtocol | None = None) → str

Generate dynamic system prompt section for todos.

Parameters:

storage – Optional storage to read current todos from.

Returns:

System prompt section with current todos, or base prompt if no todos.

class pydantic_ai_toolsets.toolsets.to_do.Todo(*, todo_id: str, content: str, status: Literal['pending', 'in_progress', 'completed'], active_form: str)

Bases: BaseModel

A todo item for task tracking.

todo_id

Unique identifier for this todo item.

Type:

str

content

The task description in imperative form (e.g., ‘Implement feature X’).

Type:

str

status

Task status - ‘pending’, ‘in_progress’, or ‘completed’.

Type:

Literal[‘pending’, ‘in_progress’, ‘completed’]

active_form

Present continuous form shown during execution (e.g., ‘Implementing feature X’).

Type:

str

todo_id: str

content: str

status: Literal['pending', 'in_progress', 'completed']

active_form: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.to_do.TodoItem(*, content: str, status: Literal['pending', 'in_progress', 'completed'], active_form: str)

Bases: BaseModel

Input model for the write_todos tool.

This is the model that agents use when calling write_todos. It has the same fields as Todo but with Field descriptions for LLM guidance.

content: str

status: Literal['pending', 'in_progress', 'completed']

active_form: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.to_do.TodoStorage(*, track_usage: bool = False)

Bases: object

Default in-memory todo storage.

Simple implementation that stores todos in memory. Use this for standalone agents or testing.

from pydantic_ai_toolsets import create_todo_toolset, TodoStorage

storage = TodoStorage()
toolset = create_todo_toolset(storage=storage)

# After agent runs, access todos directly
print(storage.todos)

# With metrics tracking
storage = TodoStorage(track_usage=True)
toolset = create_todo_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property todos: list[Todo]

Get the current list of todos.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

get_statistics() → dict[str, int | float]

Get summary statistics about todos.

Returns:

Dictionary with todo counts and completion metrics.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

clear() → None

Clear all todos and reset metrics.

property links: dict[str, list[str]]

Get outgoing links dictionary (item_id -> list of link IDs).

property linked_from: list[str]

Get incoming links list (link IDs where this storage is target).

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the todo item

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

get_state_summary() → str

Get a human-readable summary of the storage state.

Returns:

Formatted string summary of todos.

get_outputs_for_linking() → list[dict[str, str]]

Get list of linkable items with their IDs and descriptions.

Returns:

List of dictionaries with ‘id’ and ‘description’ keys for todos.

class pydantic_ai_toolsets.toolsets.to_do.TodoStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for todo storage implementations.

Any class that has a todos property (read/write) implementing list[Todo] can be used as storage for the todo toolset.

class MyCustomStorage:
    def __init__(self):
        self._todos: list[Todo] = []

    @property
    def todos(self) -> list[Todo]:
        return self._todos

    @todos.setter
    def todos(self, value: list[Todo]) -> None:
        self._todos = value

property todos: list[Todo]

Get the current list of todos.

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

add_link(item_id: str, link_id: str) → None

Add an outgoing link for an item.

Parameters:

• item_id – ID of the item (todo_id)

• link_id – ID of the link

add_linked_from(link_id: str) → None

Add an incoming link.

Parameters:

link_id – ID of the link

__init__(*args, **kwargs)

Meta-Orchestrator

Meta-orchestrator toolset for pydantic-ai agents.

Provides workflow orchestration and multi-toolset coordination capabilities for AI agents. Compatible with any pydantic-ai agent - no specific deps required.

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_meta_orchestrator_toolset, MetaOrchestratorStorage

# Simple usage
agent = Agent("openai:gpt-4.1", toolsets=[create_meta_orchestrator_toolset()])

# With storage access
storage = MetaOrchestratorStorage()
agent = Agent("openai:gpt-4.1", toolsets=[create_meta_orchestrator_toolset(storage)])
result = await agent.run("Start a research assistant workflow")
workflow = storage.get_active_workflow()
print(workflow.current_stage)

pydantic_ai_toolsets.toolsets.meta_orchestrator.create_meta_orchestrator_toolset(storage: MetaOrchestratorStorageProtocol | None = None, *, id: str | None = None, track_usage: bool = False) → FunctionToolset[Any]

Create a meta-orchestrator toolset for workflow management.

This toolset provides tools for AI agents to orchestrate multi-toolset workflows, manage transitions between toolsets, and create cross-toolset links.

Parameters:

• storage – Optional storage backend. Defaults to in-memory MetaOrchestratorStorage. You can provide a custom storage implementing MetaOrchestratorStorageProtocol for persistence or integration with other systems.

• id – Optional unique ID for the toolset.

• track_usage – If True, enables usage metrics collection in storage.

Returns:

FunctionToolset compatible with any pydantic-ai agent.

Example (standalone):

from pydantic_ai import Agent
from pydantic_ai_toolsets import create_meta_orchestrator_toolset

agent = Agent("openai:gpt-4.1", toolsets=[create_meta_orchestrator_toolset()])
result = await agent.run("Start a research assistant workflow")

Example (with storage access):

from pydantic_ai_toolsets import create_meta_orchestrator_toolset, MetaOrchestratorStorage

storage = MetaOrchestratorStorage()
toolset = create_meta_orchestrator_toolset(storage=storage)

# After agent runs, access workflow state directly
workflow = storage.get_active_workflow()
print(workflow.current_stage)
print(storage.links)

pydantic_ai_toolsets.toolsets.meta_orchestrator.get_meta_orchestrator_system_prompt() → str

Get the system prompt for the meta-orchestrator toolset.

Returns:

System prompt string describing when and how to use the meta-orchestrator tools.

pydantic_ai_toolsets.toolsets.meta_orchestrator.create_combined_toolset(toolsets: list[AbstractToolset[Any]], storages: dict[str, Any] | None = None, prefix_map: dict[str, str] | None = None, orchestrator: AbstractToolset[Any] | None = None, workflow_template: WorkflowTemplate | None = None, auto_prefix: bool = True) → tuple[CombinedToolset[Any], str]

Combine multiple toolsets with automatic collision resolution.

Uses official pydantic-ai API: - AbstractToolset.prefixed() to create aliased toolsets - CombinedToolset to combine toolsets

Strategy: 1. If auto_prefix=True, apply prefixes to all toolsets based on prefix_map

(prevents collisions proactively)

2. If auto_prefix=False, rely on CombinedToolset to detect collisions (raises UserError if collisions exist - user must handle)

3. Use CombinedToolset to combine all toolsets

4. Add orchestrator tools if provided

5. Combine system prompts from all toolsets

Parameters:

• toolsets – List of toolsets to combine

• storages – Optional dictionary mapping toolset IDs to storage instances

• prefix_map – Optional dictionary mapping toolset IDs to prefixes. If not provided, prefixes are inferred from toolset IDs.

• orchestrator – Optional meta-orchestrator toolset to add

• workflow_template – Optional workflow template for context

• auto_prefix – If True, automatically prefix all toolsets to prevent collisions

Returns:

Tuple of (CombinedToolset, combined_system_prompt)

from pydantic_ai_toolsets import create_cot_toolset, create_tot_toolset
from pydantic_ai_toolsets.toolsets.meta_orchestrator.helpers import create_combined_toolset

cot_toolset = create_cot_toolset()
tot_toolset = create_tot_toolset()

prefix_map = {
    "cot": "cot_",
    "tot": "tot_",
}

combined_toolset, combined_prompt = create_combined_toolset(
    toolsets=[cot_toolset, tot_toolset],
    prefix_map=prefix_map,
)

pydantic_ai_toolsets.toolsets.meta_orchestrator.create_workflow_agent(model: str, workflow_template: WorkflowTemplate, toolsets: list[AbstractToolset[Any]], storages: dict[str, Any] | None = None, prefix_map: dict[str, str] | None = None, orchestrator_storage: Any | None = None, auto_prefix: bool = True, additional_system_prompt: str | None = None, output_type: type[BaseModel] | list[type[BaseModel]] | None = None) → Agent[Any, str]

Create an agent configured with a workflow template and combined toolsets.

This is a convenience function that: 1. Creates a meta-orchestrator toolset (if orchestrator_storage is provided) 2. Registers all toolsets with the orchestrator 3. Combines all toolsets with automatic prefixing 4. Creates an agent with the combined toolset and workflow-aware system prompt

Parameters:

• model – Model string for the agent (e.g., “openai:gpt-4”)

• workflow_template – Workflow template to use

• toolsets – List of toolsets to combine

• storages – Optional dictionary mapping toolset IDs to storage instances

• prefix_map – Optional dictionary mapping toolset IDs to prefixes. If not provided, prefixes are inferred from toolset IDs.

• orchestrator_storage – Optional MetaOrchestratorStorage instance. If provided, creates and registers orchestrator toolset.

• auto_prefix – If True, automatically prefix all toolsets to prevent collisions

• additional_system_prompt – Optional additional system prompt that will be appended to the combined prompt. Use this to add custom instructions or context without overriding the workflow-specific prompts.

• output_type – Optional Pydantic BaseModel class or list of BaseModel classes to use as the structured output type for the agent. If provided, the agent will return structured outputs matching this schema.

Returns:

Configured Agent instance with combined toolsets and workflow template

from pydantic_ai_toolsets import (
    RESEARCH_ASSISTANT,
    create_search_toolset,
    create_self_ask_toolset,
    create_self_refine_toolset,
    create_todo_toolset,
    SearchStorage,
    SelfAskStorage,
    SelfRefineStorage,
    TodoStorage,
    MetaOrchestratorStorage,
)
from pydantic_ai_toolsets.toolsets.meta_orchestrator.helpers import create_workflow_agent

# Create storages
storages = {
    "search": SearchStorage(),
    "self_ask": SelfAskStorage(),
    "self_refine": SelfRefineStorage(),
    "todo": TodoStorage(),
}

# Create toolsets
toolsets = [
    create_search_toolset(storages["search"], id="search"),
    create_self_ask_toolset(storages["self_ask"], id="self_ask"),
    create_self_refine_toolset(storages["self_refine"], id="self_refine"),
    create_todo_toolset(storages["todo"], id="todo"),
]

# Create orchestrator storage
orchestrator_storage = MetaOrchestratorStorage()

# Create agent with workflow template and additional instructions
agent = create_workflow_agent(
    model="openai:gpt-4",
    workflow_template=RESEARCH_ASSISTANT,
    toolsets=toolsets,
    storages=storages,
    orchestrator_storage=orchestrator_storage,
    additional_system_prompt="Always cite sources and provide URLs when available.",
)

# Use the agent
result = await agent.run("Research quantum computing breakthroughs")

# Example with output type:
from pydantic import BaseModel

class ResearchResult(BaseModel):
    summary: str
    sources: list[str]
    key_findings: list[str]

agent_with_output = create_workflow_agent(
    model="openai:gpt-4",
    workflow_template=RESEARCH_ASSISTANT,
    toolsets=toolsets,
    storages=storages,
    orchestrator_storage=orchestrator_storage,
    output_type=ResearchResult,
)
result = await agent_with_output.run("Research quantum computing breakthroughs")
print(result.output.summary)  # Access structured output

pydantic_ai_toolsets.toolsets.meta_orchestrator.register_toolsets_with_orchestrator(orchestrator_storage: Any, toolsets: list[AbstractToolset[Any]], storages: dict[str, Any] | None = None) → None

Register toolsets with the meta-orchestrator storage.

Parameters:

• orchestrator_storage – MetaOrchestratorStorage instance

• toolsets – List of toolsets to register

• storages – Optional dictionary mapping toolset IDs to storage instances

class pydantic_ai_toolsets.toolsets.meta_orchestrator.WorkflowState(workflow_id: str, template_name: str, current_stage: int = 0, active_toolsets: list[str] = None, completed_stages: list[str] = None, transitions: list[ToolsetTransition] = None, links: list[CrossToolsetLink] = None, started_at: float | None = None, updated_at: float | None = None)

Bases: object

Tracks the state of an active workflow.

workflow_id

Unique identifier for this workflow instance

Type:

str

template_name

Name of the workflow template being used

Type:

str

current_stage

Index of the current stage

Type:

int

active_toolsets

List of toolset IDs currently active

Type:

list[str]

completed_stages

List of stage names that have been completed

Type:

list[str]

transitions

List of transitions that have occurred

Type:

list[pydantic_ai_toolsets.toolsets.meta_orchestrator.types.ToolsetTransition]

links

List of cross-toolset links created

Type:

list[pydantic_ai_toolsets.toolsets.meta_orchestrator.types.CrossToolsetLink]

started_at

Timestamp when workflow started

Type:

float | None

updated_at

Timestamp when workflow was last updated

Type:

float | None

workflow_id: str

template_name: str

current_stage: int = 0

active_toolsets: list[str] = None

completed_stages: list[str] = None

transitions: list[ToolsetTransition] = None

links: list[CrossToolsetLink] = None

started_at: float | None = None

updated_at: float | None = None

__post_init__()

Initialize default values for mutable fields.

__init__(workflow_id: str, template_name: str, current_stage: int = 0, active_toolsets: list[str] = None, completed_stages: list[str] = None, transitions: list[ToolsetTransition] = None, links: list[CrossToolsetLink] = None, started_at: float | None = None, updated_at: float | None = None) → None

class pydantic_ai_toolsets.toolsets.meta_orchestrator.WorkflowTemplate(name: str, toolsets: list[str], stages: list[Stage], handoff_instructions: dict[str, str], description: str | None = None)

Bases: object

A predefined workflow pattern combining multiple toolsets.

name

Unique identifier for this template

Type:

str

toolsets

List of toolset IDs in order

Type:

list[str]

stages

List of stages with transition conditions

Type:

list[pydantic_ai_toolsets.toolsets.meta_orchestrator.types.Stage]

handoff_instructions

Mapping of stage transitions to instructions

Type:

dict[str, str]

description

Optional description of when to use this template

Type:

str | None

name: str

toolsets: list[str]

stages: list[Stage]

handoff_instructions: dict[str, str]

description: str | None = None

__init__(name: str, toolsets: list[str], stages: list[Stage], handoff_instructions: dict[str, str], description: str | None = None) → None

class pydantic_ai_toolsets.toolsets.meta_orchestrator.Stage(name: str, toolset_id: str, transition_condition: str, description: str | None = None)

Bases: object

Represents a stage in a workflow template.

name

Human-readable name for this stage

Type:

str

toolset_id

ID of the toolset to use in this stage

Type:

str

transition_condition

Condition that must be met to transition to next stage

Type:

str

description

Optional description of what happens in this stage

Type:

str | None

name: str

toolset_id: str

transition_condition: str

description: str | None = None

__init__(name: str, toolset_id: str, transition_condition: str, description: str | None = None) → None

class pydantic_ai_toolsets.toolsets.meta_orchestrator.ToolsetTransition(from_toolset_id: str, to_toolset_id: str, reason: str, confidence: float = 1.0, conditions_met: list[str] | None = None)

Bases: object

Suggests when to switch from one toolset to another.

from_toolset_id

ID of the current toolset

Type:

str

to_toolset_id

ID of the recommended next toolset

Type:

str

reason

Explanation for why this transition is recommended

Type:

str

confidence

Confidence score (0.0 to 1.0)

Type:

float

conditions_met

List of conditions that triggered this transition

Type:

list[str] | None

from_toolset_id: str

to_toolset_id: str

reason: str

confidence: float = 1.0

conditions_met: list[str] | None = None

__init__(from_toolset_id: str, to_toolset_id: str, reason: str, confidence: float = 1.0, conditions_met: list[str] | None = None) → None

class pydantic_ai_toolsets.toolsets.meta_orchestrator.CrossToolsetLink(link_id: str, source_toolset_id: str, source_item_id: str, target_toolset_id: str, target_item_id: str, link_type: LinkType, created_at: float | None = None)

Bases: object

A link between outputs from different toolsets.

link_id

Unique identifier for this link

Type:

str

source_toolset_id

ID of the source toolset

Type:

str

source_item_id

ID of the item in the source toolset

Type:

str

target_toolset_id

ID of the target toolset

Type:

str

target_item_id

ID of the item in the target toolset

Type:

str

link_type

Type of relationship (refines, explores, synthesizes, references)

Type:

pydantic_ai_toolsets.toolsets.meta_orchestrator.types.LinkType

created_at

Timestamp when link was created

Type:

float | None

link_id: str

source_toolset_id: str

source_item_id: str

target_toolset_id: str

target_item_id: str

link_type: LinkType

created_at: float | None = None

__init__(link_id: str, source_toolset_id: str, source_item_id: str, target_toolset_id: str, target_item_id: str, link_type: LinkType, created_at: float | None = None) → None

class pydantic_ai_toolsets.toolsets.meta_orchestrator.LinkType(*values)

Bases: str, Enum

Types of links between toolset outputs.

REFINES = 'refines'

EXPLORES = 'explores'

SYNTHESIZES = 'synthesizes'

REFERENCES = 'references'

class pydantic_ai_toolsets.toolsets.meta_orchestrator.StartWorkflowItem(*, template_name: str, initial_context: dict[str, Any] | None = None)

Bases: BaseModel

Input model for starting a workflow.

template_name: str

initial_context: dict[str, Any] | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.meta_orchestrator.SuggestTransitionItem(*, current_toolset_id: str | None = None, current_state_summary: str | None = None)

Bases: BaseModel

Input model for suggesting a toolset transition.

current_toolset_id: str | None

current_state_summary: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.meta_orchestrator.LinkToolsetOutputsItem(*, source_toolset_id: str, source_item_id: str, target_toolset_id: str, target_item_id: str, link_type: LinkType)

Bases: BaseModel

Input model for creating a cross-toolset link.

source_toolset_id: str

source_item_id: str

target_toolset_id: str

target_item_id: str

link_type: LinkType

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.meta_orchestrator.GetWorkflowStatusItem(*, workflow_id: str | None = None)

Bases: BaseModel

Input model for getting workflow status.

workflow_id: str | None

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pydantic_ai_toolsets.toolsets.meta_orchestrator.MetaOrchestratorStorage(*, track_usage: bool = False)

Bases: object

Default in-memory meta-orchestrator storage.

Tracks active workflows, registered toolsets, transitions, and cross-toolset links. Use this for standalone agents or testing.

_registered_toolsets

Dictionary mapping toolset IDs to their metadata

Type:

dict[str, dict[str, Any]]

_active_workflows

Dictionary mapping workflow IDs to WorkflowState

Type:

dict[str, WorkflowState]

_links

List of all cross-toolset links

Type:

list[CrossToolsetLink]

_transitions

List of all toolset transitions

Type:

list[ToolsetTransition]

_workflow_registry

Registry of workflow templates

Type:

WorkflowRegistry

_metrics

Optional usage metrics tracker

Type:

UsageMetrics | None

from pydantic_ai_toolsets import create_meta_orchestrator_toolset, MetaOrchestratorStorage

storage = MetaOrchestratorStorage()
toolset = create_meta_orchestrator_toolset(storage=storage)

# After agent runs, access workflow state
workflow = storage.get_active_workflow()
print(workflow.current_stage)
print(storage.links)

# With metrics tracking
storage = MetaOrchestratorStorage(track_usage=True)
toolset = create_meta_orchestrator_toolset(storage=storage)
print(storage.metrics.total_tokens())

__init__(*, track_usage: bool = False) → None

Initialize storage with optional metrics tracking.

Parameters:

track_usage – If True, enables usage metrics collection.

property registered_toolsets: dict[str, dict[str, Any]]

Get dictionary of registered toolsets.

property active_workflows: dict[str, WorkflowState]

Get dictionary of active workflows.

property links: list[CrossToolsetLink]

Get list of all cross-toolset links.

property transitions: list[ToolsetTransition]

Get list of all toolset transitions.

property workflow_registry: WorkflowRegistry

Get the workflow registry.

property metrics: UsageMetrics | None

Get usage metrics if tracking is enabled.

register_toolset(toolset_id: str, toolset_info: dict[str, Any]) → None

Register a toolset with the orchestrator.

Parameters:

• toolset_id – Unique identifier for the toolset

• toolset_info – Dictionary with toolset metadata (e.g., type, label, tools)

track_transition(transition: ToolsetTransition) → None

Track a toolset transition and automatically update workflow progression.

Parameters:

transition – The transition to track

create_link(link: CrossToolsetLink) → None

Create a cross-toolset link.

Parameters:

link – The link to create

get_links_for_item(toolset_id: str, item_id: str) → list[CrossToolsetLink]

Get all links for a specific item.

Parameters:

• toolset_id – ID of the toolset

• item_id – ID of the item

Returns:

List of links where this item is source or target

start_workflow(workflow: WorkflowState) → None

Start a new workflow.

Parameters:

workflow – The workflow state to start

get_active_workflow() → WorkflowState | None

Get the currently active workflow.

Returns:

The most recently updated workflow, or None if no workflows exist

update_workflow(workflow_id: str, updates: dict[str, Any]) → None

Update a workflow’s state.

Parameters:

• workflow_id – ID of the workflow to update

• updates – Dictionary of updates to apply

get_unified_state() → dict[str, Any]

Get unified state across all registered toolsets.

Returns:

• active_toolsets: List of registered toolset IDs

• active_workflows: List of active workflow IDs

• total_links: Total number of cross-toolset links

• total_transitions: Total number of transitions

• current_workflow: Current workflow state if any

Return type:

Dictionary containing

summary() → dict[str, Any]

Get comprehensive JSON summary of storage state and metrics.

Returns:

Dictionary containing storage state, statistics, and usage metrics.

class pydantic_ai_toolsets.toolsets.meta_orchestrator.MetaOrchestratorStorageProtocol(*args, **kwargs)

Bases: Protocol

Protocol for meta-orchestrator storage implementations.

Any class that implements these methods can be used as storage for the meta-orchestrator toolset.

register_toolset(toolset_id: str, toolset_info: dict[str, Any]) → None

Register a toolset with the orchestrator.

track_transition(transition: ToolsetTransition) → None

Track a toolset transition.

create_link(link: CrossToolsetLink) → None

Create a cross-toolset link.

get_unified_state() → dict[str, Any]

Get unified state across all registered toolsets.

start_workflow(workflow: WorkflowState) → None

Start a new workflow.

get_active_workflow() → WorkflowState | None

Get the currently active workflow.

update_workflow(workflow_id: str, updates: dict[str, Any]) → None

Update a workflow’s state.

__init__(*args, **kwargs)

class pydantic_ai_toolsets.toolsets.meta_orchestrator.WorkflowRegistry(templates: dict[str, ~pydantic_ai_toolsets.toolsets.meta_orchestrator.types.WorkflowTemplate]=<factory>)

Bases: object

Registry for workflow templates.

templates: dict[str, WorkflowTemplate]

register(template: WorkflowTemplate) → None

Register a workflow template.

get(name: str) → WorkflowTemplate | None

Get a workflow template by name.

list_all() → list[str]

List all registered template names.

__init__(templates: dict[str, ~pydantic_ai_toolsets.toolsets.meta_orchestrator.types.WorkflowTemplate]=<factory>) → None

pydantic_ai_toolsets.toolsets.meta_orchestrator.get_template(name: str) → WorkflowTemplate | None

Get a workflow template by name.

Parameters:

name – Name of the template (e.g., ‘research_assistant’)

Returns:

WorkflowTemplate if found, None otherwise

pydantic_ai_toolsets.toolsets.meta_orchestrator.list_templates() → list[str]

List all available workflow template names.

Returns:

List of template names

Shared Utilities

Dynamic toolset aliasing using official pydantic-ai API.

pydantic_ai_toolsets.toolsets._shared.aliasing.get_prefix_for_toolset(toolset_id: str | None, toolset_label: str | None = None) → str

Get the prefix for a toolset based on its ID or label.

Parameters:

• toolset_id – Optional toolset ID (e.g., “cot”, “tot”, “self_ask”)

• toolset_label – Optional toolset label/name

Returns:

Prefix string (e.g., “cot_”, “tot_”, “self_ask_”)

prefix = get_prefix_for_toolset("cot")  # Returns "cot_"
prefix = get_prefix_for_toolset("self_ask")  # Returns "self_ask_"

pydantic_ai_toolsets.toolsets._shared.aliasing.create_aliased_toolset(base_toolset: AbstractToolset[Any], prefix: str) → AbstractToolset[Any]

Create an aliased version of a toolset with prefixed tool names.

Uses the official pydantic-ai API: AbstractToolset.prefixed()

This function wraps a toolset and prefixes all its tool names to avoid collisions when combining multiple toolsets.

Parameters:

• base_toolset – The original toolset to alias (unchanged)

• prefix – Prefix to add to all tool names (e.g., “cot_”, “tot_”)

Returns:

PrefixedToolset with aliased tool names

from pydantic_ai_toolsets import create_cot_toolset

cot_toolset = create_cot_toolset()
aliased_cot = create_aliased_toolset(cot_toolset, "cot_")
# Tools are now: cot_read_thoughts, cot_write_thoughts

Note

• The original toolset is NOT modified

• All tool names are prefixed (even if no collision exists)

• System prompts remain unchanged (handled separately)

Cross-toolset linking infrastructure for creating references between toolset outputs.

class pydantic_ai_toolsets.toolsets._shared.linking.LinkManager(_links: dict[str, ~pydantic_ai_toolsets.toolsets.meta_orchestrator.types.CrossToolsetLink] = <factory>, _links_by_source: dict[tuple[str, str], list[str]] = <factory>, _links_by_target: dict[tuple[str, str], list[str]] = <factory>)

Bases: object

Manages cross-toolset links between outputs from different toolsets.

This class provides a centralized way to create, track, and resolve links between items across different toolsets. Links can represent relationships like “refines”, “explores”, “synthesizes”, or “references”.

_links

Dictionary mapping link IDs to CrossToolsetLink objects

Type:

dict[str, pydantic_ai_toolsets.toolsets.meta_orchestrator.types.CrossToolsetLink]

_links_by_source

Dictionary mapping (toolset_id, item_id) to list of link IDs

Type:

dict[tuple[str, str], list[str]]

_links_by_target

Dictionary mapping (toolset_id, item_id) to list of link IDs

Type:

dict[tuple[str, str], list[str]]

from pydantic_ai_toolsets.toolsets._shared.linking import LinkManager

link_manager = LinkManager()

# Create a link
link_id = link_manager.create_link(
    source_toolset="search",
    source_id="result_123",
    target_toolset="self_ask",
    target_id="question_456",
    link_type=LinkType.REFERENCES
)

# Get all links for an item
links = link_manager.get_links("search", "result_123")

# Resolve a link
link = link_manager.resolve_link(link_id)

create_link(source_toolset: str, source_id: str, target_toolset: str, target_id: str, link_type: LinkType, link_id: str | None = None) → str

Create a link between outputs from different toolsets.

Parameters:

• source_toolset – ID of the source toolset

• source_id – ID of the item in the source toolset

• target_toolset – ID of the target toolset

• target_id – ID of the item in the target toolset

• link_type – Type of link (refines, explores, synthesizes, references)

• link_id – Optional custom link ID. If not provided, generates a unique ID.

Returns:

The link ID (either provided or generated)

Raises:

ValueError – If source and target are the same item

get_links(toolset_id: str, item_id: str) → list[CrossToolsetLink]

Get all links for a specific item (both outgoing and incoming).

Parameters:

• toolset_id – ID of the toolset

• item_id – ID of the item

Returns:

List of CrossToolsetLink objects where the item is either source or target

get_outgoing_links(toolset_id: str, item_id: str) → list[CrossToolsetLink]

Get outgoing links from a specific item.

Parameters:

• toolset_id – ID of the toolset

• item_id – ID of the item

Returns:

List of CrossToolsetLink objects where the item is the source

get_incoming_links(toolset_id: str, item_id: str) → list[CrossToolsetLink]

Get incoming links to a specific item.

Parameters:

• toolset_id – ID of the toolset

• item_id – ID of the item

Returns:

List of CrossToolsetLink objects where the item is the target

resolve_link(link_id: str) → CrossToolsetLink | None

Get a link by its ID.

Parameters:

link_id – ID of the link to resolve

Returns:

CrossToolsetLink if found, None otherwise

delete_link(link_id: str) → bool

Delete a link by its ID.

Parameters:

link_id – ID of the link to delete

Returns:

True if link was deleted, False if not found

clear() → None

Clear all links.

get_all_links() → list[CrossToolsetLink]

Get all links.

Returns:

List of all CrossToolsetLink objects

get_statistics() → dict[str, Any]

Get statistics about links.

Returns:

Dictionary with link counts by type and total counts

__init__(_links: dict[str, ~pydantic_ai_toolsets.toolsets.meta_orchestrator.types.CrossToolsetLink] = <factory>, _links_by_source: dict[tuple[str, str], list[str]] = <factory>, _links_by_target: dict[tuple[str, str], list[str]] = <factory>) → None

Shared metrics infrastructure for usage cost tracking.

Provides common types and utilities for tracking tool invocations, token estimates, and performance metrics across all toolset packages.

pydantic_ai_toolsets.toolsets._shared.metrics.estimate_tokens(text: str) → int

Estimate token count for a string.

Uses a simple heuristic: ~4 characters per token on average. This is a rough estimate suitable for cost tracking purposes.

Parameters:

text – The text to estimate tokens for.

Returns:

Estimated token count.

class pydantic_ai_toolsets.toolsets._shared.metrics.ToolInvocation(tool_name: str, timestamp: float, input_tokens: int = 0, output_tokens: int = 0, duration_ms: float = 0.0)

Bases: object

Record of a single tool invocation.

tool_name

Name of the tool that was invoked.

Type:

str

timestamp

Unix timestamp when the tool was called.

Type:

float

input_tokens

Estimated tokens in the input/parameters.

Type:

int

output_tokens

Estimated tokens in the output/response.

Type:

int

duration_ms

Execution duration in milliseconds.

Type:

float

tool_name: str

timestamp: float

input_tokens: int = 0

output_tokens: int = 0

duration_ms: float = 0.0

__init__(tool_name: str, timestamp: float, input_tokens: int = 0, output_tokens: int = 0, duration_ms: float = 0.0) → None

class pydantic_ai_toolsets.toolsets._shared.metrics.UsageMetrics(invocations: list[ToolInvocation] = <factory>)

Bases: object

Aggregated usage metrics for a toolset session.

Tracks all tool invocations and provides summary statistics. Thread-safe for basic append operations.

metrics = UsageMetrics()
metrics.record_invocation("read_thoughts", "", "Chain of thoughts:\n...")
print(f"Total tokens: {metrics.total_tokens()}")

invocations: list[ToolInvocation]

record_invocation(tool_name: str, input_text: str, output_text: str, duration_ms: float = 0.0) → ToolInvocation

Record a tool invocation with automatic token estimation.

Parameters:

• tool_name – Name of the tool.

• input_text – Serialized input parameters.

• output_text – Tool output/response.

• duration_ms – Execution time in milliseconds.

Returns:

The created ToolInvocation record.

total_input_tokens() → int

Get total estimated input tokens across all invocations.

total_output_tokens() → int

Get total estimated output tokens across all invocations.

total_tokens() → int

Get total estimated tokens (input + output).

invocation_count() → dict[str, int]

Get count of invocations per tool.

Returns:

Dictionary mapping tool names to invocation counts.

total_duration_ms() → float

Get total execution time across all invocations.

average_duration_ms() → float

Get average execution time per invocation.

to_dict() → dict[str, Any]

Export metrics as a dictionary for serialization.

Returns:

Dictionary with all metrics data.

clear() → None

Clear all recorded invocations.

__init__(invocations: list[ToolInvocation] = <factory>) → None

pydantic_ai_toolsets.toolsets._shared.metrics.create_tracking_wrapper(metrics: UsageMetrics, tool_name: str, func: Callable[[...], T]) → Callable[[...], T]

Create a wrapper that tracks tool invocations.

Parameters:

• metrics – UsageMetrics instance to record to.

• tool_name – Name of the tool being wrapped.

• func – The tool function to wrap.

Returns:

Wrapped function that records metrics.

System prompt combination for multi-toolset workflows.

pydantic_ai_toolsets.toolsets._shared.system_prompts.identify_toolset_type(toolset: AbstractToolset[Any]) → str

Identify toolset type from its ID or label.

Parameters:

toolset – The toolset to identify

Returns:

Toolset type string (e.g., “cot”, “tot”, “self_ask”)

pydantic_ai_toolsets.toolsets._shared.system_prompts.build_tool_name_mapping(prompt: str, prefix: str) → dict[str, str]

Build mapping of original tool names to prefixed names from prompt.

Extracts tool names mentioned in backticks and creates mapping.

Parameters:

• prompt – System prompt text

• prefix – Prefix to add to tool names

Returns:

Dictionary mapping original tool names to prefixed names

pydantic_ai_toolsets.toolsets._shared.system_prompts.update_prompt_tool_names(prompt: str, tool_name_mapping: dict[str, str]) → str

Update tool names in a system prompt to reflect aliased names.

Parameters:

• prompt – Original system prompt

• tool_name_mapping – Mapping of original tool names to aliased names e.g., {“read_thoughts”: “cot_read_thoughts”}

Returns:

Updated prompt with aliased tool names

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_workflow_instructions(workflow_template: WorkflowTemplate, prefix_map: dict[str, str] | None = None) → str

Generate workflow-specific instructions from a template.

Parameters:

• workflow_template – The workflow template

• prefix_map – Optional mapping of toolset IDs to prefixes

Returns:

Workflow instructions string

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_search_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Search toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_self_ask_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Self-Ask toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_self_refine_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Self-Refine toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_todo_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Todo toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_reflection_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Reflection toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_cot_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Chain of Thought toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_tot_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Tree of Thought toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_got_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Graph of Thought toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_mcts_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for MCTS toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_beam_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Beam Search toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_persona_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Multi-Persona Analysis toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_persona_debate_combination_prompt(toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], position: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate combination-specific prompt for Multi-Persona Debate toolset.

pydantic_ai_toolsets.toolsets._shared.system_prompts.generate_combination_prompt_for_toolset(toolset_type: str, toolset: AbstractToolset[Any], storage: Any | None, other_toolsets: list[AbstractToolset[Any]], toolset_order: int, prefix_map: dict[str, str] | None, workflow_template: WorkflowTemplate | None) → str

Generate a combination-specific prompt for a single toolset.

Parameters:

• toolset_type – Type identifier (e.g., “search”, “self_ask”)

• toolset – The toolset to generate prompt for

• storage – Optional storage for dynamic prompts

• other_toolsets – Other toolsets in the combination

• toolset_order – Order of toolset (index)

• prefix_map – Mapping of toolset IDs to prefixes

• workflow_template – Optional workflow template

Returns:

Combination-specific prompt for this toolset

pydantic_ai_toolsets.toolsets._shared.system_prompts.combine_system_prompts(toolsets: list[AbstractToolset[Any]], storages: dict[str, Any] | None = None, prefix_map: dict[str, str] | None = None, workflow_template: WorkflowTemplate | None = None, use_combination_prompts: bool = True) → str

Combine system prompts from multiple toolsets.

Parameters:

• toolsets – List of toolsets to combine prompts from

• storages – Optional mapping of toolset IDs to storage objects

• prefix_map – Optional mapping of toolset IDs to prefixes

• workflow_template – Optional workflow template for context

• use_combination_prompts – If True, use combination-specific prompts

Returns:

Combined system prompt with all toolset instructions

On this page