pydantic-pydantic-ai-4781-4687

.
./.claude
./.claude/skills
./.gemini
./.git
./.git/branches
./.git/hooks
./.git/info
./.git/logs
./.git/objects
./.git/refs
./.github

./.github/ISSUE_TEMPLATE
./.github/workflows
./agent_docs
./clai
./clai/clai
./docs
./docs/.cursor
./docs/.hooks
./docs/.overrides
./docs/.partials
./docs/api
./docs/durable_execution
./docs/evals
./docs/examples
./docs/extra
./docs/graph
./docs/img
./docs/javascripts
./docs/mcp
./docs/models
./docs/ui
./docs-site
./docs-site/src
./examples
./examples/pydantic_ai_examples
./pydantic_ai_slim
./pydantic_ai_slim/pydantic_ai
./pydantic_evals
./pydantic_evals/pydantic_evals
./pydantic_graph
./pydantic_graph/pydantic_graph
./scripts
./tests
./tests/assets
./tests/cassettes
./tests/evals
./tests/example_modules
./tests/ext
./tests/graph
./tests/models
./tests/profiles
./tests/providers

pydantic_ai_slim/pydantic_ai/agent/__init__.py:186:    _prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = dataclasses.field(repr=False)
pydantic_ai_slim/pydantic_ai/agent/__init__.py:218:        prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:250:        prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:280:        prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:325:            prepare_tools: Custom function to prepare the tool definition of all tools for each step, except output tools.
pydantic_ai_slim/pydantic_ai/agent/__init__.py:427:        self._prepare_tools = prepare_tools
pydantic_ai_slim/pydantic_ai/agent/__init__.py:500:        prepare_tools: ToolsPrepareFunc[Any] | None = None,
pydantic_ai_slim/pydanti
…[5586 chars omitted]…
n_step); in that case we
pydantic_ai_slim/pydantic_ai/_agent_graph.py:936:        ctx.deps.tool_manager = await ctx.deps.tool_manager.for_run_step(run_context)
pydantic_ai_slim/pydantic_ai/_tool_manager.py:100:    async def for_run_step(self, ctx: RunContext[AgentDepsT]) -> ToolManager[AgentDepsT]:

pydantic_ai_slim/pydantic_ai/_tool_manager.py:112:        toolset = await self.toolset.for_run_step(ctx)

from __future__ import annotations

from collections.abc import AsyncIterable, Awaitable, Callable, Sequence
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING, Any

from pydantic_ai import _system_prompt
from pydantic_ai._instructions import AgentInstructions, normalize_instructions
from pydantic_ai.messages import AgentStreamEvent, ModelResponse, ToolCallPart
from pydantic_ai.settings import ModelSettings, merge_model_settings
from pydantic_ai.tools import AgentBuiltinTool, AgentDepsT, RunContext, ToolDefinition
from pydantic_ai.toolsets import AbstractToolset, AgentToolset, CombinedToolset

from pydantic_ai.toolsets._dynamic import DynamicToolset

from .abstract import AbstractCapability, ModelRequestContext

if TYPE_CHECKING:
    from pydantic_ai import _agent_graph
    from pydantic_ai.result import FinalResult
    from pydantic_ai.run import AgentRunResul
…[5772 chars omitted]…
tDepsT],
        *,
        request_context: ModelRequestContext,
        response: ModelResponse,
    ) -> ModelResponse:
        for capability in reversed(self.capabilities):
            response = await capability.after_model_request(ctx, request_context=request_context, response=response)
        return response

    async def wrap_model_request(
        self,
        ctx: RunContext[AgentDepsT],

180:            wrap_validation_errors: If true, wrap the validation errors in a retry message.
183:            Result of either the validated result data (ok) or a retry message (Err).
489:class ToolOutputSchema(OutputSchema[OutputDataT]):
614:            wrap_validation_errors: If true, wrap the validation errors in a retry message.
617:            Either the validated output data (left) or a retry message (right).
866:    max_retries: int
943:        max_retries: int = 1,
948:        self.max_retries = max_retries
964:                max_retries=self.max_retries,

def __init__(
        self,
        output_function: TextOutputFunc[OutputDataT],
    ):
        self._function_schema = _function_schema.function_schema(output_function, GenerateToolJsonSchema)

        if (
            not (arguments_schema := self._function_schema.json_schema.get('properties', {}))
            or len(arguments_schema) != 1
            or not (argument_name := next(iter(arguments_schema.keys()), None))
            or arguments_schema.get(argument_name, {}).get('type') != 'string'
        ):

            raise UserError('TextOutput must take a function taking a single `str` argument')

        self._str_argument_name = argument_name

    async def process(
        self,
        data: str,
        *,
        run_context: RunContext[AgentDepsT],
        validation_context: Any | None = None,
        allow_partial: bool = False,
        wrap_validation_errors: bool 
…[6293 chars omitted]…
nce(output, NativeOutput):
            outputs_flat.extend(types_from_output_spec(output.outputs))  # pyright: ignore[reportUnknownMemberType,reportUnknownArgumentType]
        elif isinstance(output, PromptedOutput):
            outputs_flat.extend(types_from_output_spec(output.outputs))  # pyright: ignore[reportUnknownMemberType,reportUnknownArgumentType]
        elif isinstance(output, TextOutput):

pydantic_ai_slim/pydantic_ai/agent/__init__.py:432:            self._output_toolset.max_retries = self._max_result_retries
pydantic_ai_slim/pydantic_ai/agent/__init__.py:436:            max_retries=self._max_tool_retries,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:1022:                output_toolset.max_retries = self._max_result_retries
pydantic_ai_slim/pydantic_ai/agent/__init__.py:1113:            toolset, root_capability=run_capability, default_max_retries=self._max_tool_retries
pydantic_ai_slim/pydantic_ai/agent/__init__.py:2142:                max_retries=self._max_tool_retries,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:2327:        max_retries: int = 1,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:2333:        super().__init__(tools, max_retries=max_retries, timeout=timeout, id=id)
pydantic_ai_slim/pydantic_ai/durable_exec/prefect/_cache_policies.py:40:              
…[5529 chars omitted]…
 max_retries=tool.max_retries,
pydantic_ai_slim/pydantic_ai/_tool_manager.py:360:            max_retries = tool.max_retries if tool is not None else self.default_max_retries
pydantic_ai_slim/pydantic_ai/_tool_manager.py:361:            self._check_max_retries(name, max_retries, e)

pydantic_ai_slim/pydantic_ai/_tool_manager.py:464:            self._check_max_retries(name, validated.tool.max_retries, e)

run_id='...',
                    )
                ),
                CallToolsNode(
                    model_response=ModelResponse(
                        parts=[TextPart(content='The capital of France is Paris.')],
                        usage=RequestUsage(input_tokens=56, output_tokens=7),
                        model_name='gpt-5.2',
                        timestamp=datetime.datetime(...),
                        run_id='...',
                    )
                ),

                End(data=FinalResult(output='The capital of France is Paris.')),
            ]
            '''
            print(agent_run.result.output)
            #> The capital of France is Paris.
        ```

        Args:
            user_prompt: User input to start/continue the conversation.
            output_type: Custom output type to use for this run, `output_type` may only be 
…[7368 chars omitted]…
   return '\n\n'.join(parts).strip()

        graph_deps = _agent_graph.GraphAgentDeps[AgentDepsT, OutputDataT](
            user_deps=deps,
            prompt=user_prompt,
            new_message_index=len(message_history) if message_history else 0,
            resumed_request=None,
            model=model_used,
            get_model_settings=get_model_settings,
            usage_limits=usage_limits,

860:    def iter(
880:    def iter(
900:    async def iter(  # noqa: C901
1106:        toolset = self._get_toolset(
2093:    def _get_toolset(
2191:                    toolset = self._get_toolset()
2221:        self._get_toolset().apply(_set_sampling_model)

cls,
        path: Path | str,
        *,
        fmt: Literal['yaml', 'json'] | None = None,
        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
        model: models.Model | models.KnownModelName | str | None = None,
        output_type: OutputSpec[Any] = str,
        instructions: AgentInstructions[Any] = None,
        system_prompt: str | Sequence[str] = (),
        name: str | None = None,
        description: TemplateStr[Any] | str | None = None,
        model_settings: ModelSettings | None = None,

        retries: int | None = None,
        validation_context: Any = None,
        output_retries: int | None = None,
        tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
        builtin_tools: Sequence[AgentBuiltinTool[Any]] = (),
        prepare_tools: ToolsPrepareFunc[Any] | None = None,
        prepare_output_tools: ToolsPrepareFu
…[8629 chars omitted]…
              request=ModelRequest(
                        parts=[
                            UserPromptPart(
                                content='What is the capital of France?',
                                timestamp=datetime.datetime(...),
                            )
                        ],
                        timestamp=datetime.datetime(...),
                        run_id='...',

2348 pydantic_ai_slim/pydantic_ai/agent/__init__.py

from __future__ import annotations

import inspect
import json
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field, replace
from typing import TYPE_CHECKING, Any, Generic, Literal

from opentelemetry.trace import Tracer
from pydantic import ValidationError

from typing_extensions import deprecated

from . import messages as _messages
from ._instrumentation import InstrumentationNames
from ._run_context import AgentDepsT, RunContext
from .exceptions import ModelRetry, SkipToolExecution, SkipToolValidation, ToolRetryError, UnexpectedModelBehavior
from .messages import ToolCallPart
from .tools import ToolDefinition
from .toolsets.abstract import AbstractToolset, ToolsetTool
from .usage import RunUsage

if TYPE_CHECKING:
    from .capabilities.abstract import AbstractCapability

ParallelExecu
…[7983 chars omitted]…
       args_dict = validator.validate_python(
                raw_args or {}, allow_partial=pyd_allow_partial, context=ctx.validation_context
            )

        if tool.args_validator_func is not None:
            result = tool.args_validator_func(ctx, **args_dict)
            if inspect.isawaitable(result):
                await result

        return args_dict

    async def _run_validate_hooks(

229:    def build(  # noqa: C901
321:        toolset = OutputToolset.build(tool_outputs + other_outputs, name=name, description=description, strict=strict)
461:    def build_instructions(cls, template: str, object_def: OutputObjectDefinition) -> str:
870:    def build(
213:class OutputSchema(ABC, Generic[OutputDataT]):

assert validated.tool is not None
        assert validated.validated_args is not None

        if validated.tool.tool_def.kind == 'external':
            raise RuntimeError('External tools cannot be called')

        try:
            tool_result = await self._run_execute_hooks(validated, usage=usage)
        except SkipToolExecution as e:
            if usage is not None:
                usage.tool_calls += 1
            return e.result

        return tool_result

    async def _raw_execute(
        self,
        validated: ValidatedToolCall[AgentDepsT],
        *,
        usage: RunUsage | None = None,
    ) -> Any:
        """Execute a validated tool call without hooks or tracing."""
        assert validated.tool is not None
        assert validated.validated_args is not None

        name = validated.call.tool_name
        try:
            tool_result = await self.toolset.
…[4827 chars omitted]…
.add(name)

            if not wrap_validation_errors:
                raise

            validation_error = self._wrap_error_as_retry(name, call, e)

            return ValidatedToolCall(
                call=call,
                tool=tool,
                ctx=ctx,
                args_valid=False,
                validated_args=None,
                validation_error=validation_error,
            )

from __future__ import annotations as _annotations

import asyncio
import dataclasses
import inspect
import uuid
from asyncio import Task
from collections import defaultdict, deque
from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Sequence
from contextlib import asynccontextmanager, contextmanager
from contextvars import ContextVar
from copy import deepcopy

from dataclasses import field, replace
from typing import TYPE_CHECKING, Any, Generic, Literal, TypeGuard, cast

from opentelemetry.trace import Tracer
from typing_extensions import TypeVar, assert_never

from pydantic_ai._history_processor import HistoryProcessor
from pydantic_ai._instrumentation import DEFAULT_INSTRUMENTATION_VERSION
from pydantic_ai._tool_manager import ToolManager, ValidatedToolCall
from pydantic_ai._utils import dataclasses_no_defaults_repr, get_union_args, now_utc
from pydantic_ai.buil
…[5239 chars omitted]…
dles the user prompt and instructions."""

    user_prompt: str | Sequence[_messages.UserContent] | None

    _: dataclasses.KW_ONLY

    deferred_tool_results: DeferredToolResults | None = None

    instructions: str | None = None
    instructions_functions: list[_system_prompt.SystemPromptRunner[DepsT]] = dataclasses.field(
        default_factory=list[_system_prompt.SystemPromptRunner[DepsT]]
    )

def _emit_skipped_output_tool(
    call: _messages.ToolCallPart,
    message: str,
    output_parts: list[_messages.ModelRequestPart],
    *,
    args_valid: bool | None = None,
) -> Iterator[_messages.HandleResponseEvent]:
    """Yield events for an output tool call that was skipped, and append the result to output_parts."""
    yield _messages.FunctionToolCallEvent(call, args_valid=args_valid)
    part = _messages.ToolReturnPart(
        tool_name=call.tool_name,
        content=message,

        tool_call_id=call.tool_call_id,
    )
    output_parts.append(part)
    yield _messages.FunctionToolResultEvent(part)


async def process_tool_calls(  # noqa: C901
    tool_manager: ToolManager[DepsT],
    tool_calls: list[_messages.ToolCallPart],
    tool_call_results: dict[str, DeferredToolResult | Literal['skip']] | None,
    tool_call_metadata: dict[str, dict[str, Any]] | None,
    fina
…[8323 chars omitted]…
deferred_result, ToolApproved):
                yield _messages.FunctionToolCallEvent(call)
                continue
            try:
                if isinstance(deferred_result, ToolApproved):
                    validate_call = call
                    if deferred_result.override_args is not None:
                        validate_call = dataclasses.replace(call, args=deferred_result.override_args)

tests/test_agent.py:38:    PrefixedToolset,
tests/test_agent.py:6399:    toolset = CombinedToolset([server1, PrefixedToolset(server2, 'prefix')])
tests/test_agent.py:6419:    toolset = CombinedToolset([server1, PrefixedToolset(server2, 'prefix')])
tests/test_agent.py:6579:    toolset = CombinedToolset([server1, PrefixedToolset(server2, 'prefix')])
tests/test_agent.py:6633:    assert wrapper_agent.toolsets == agent.toolsets
tests/test_agent.py:6659:    with wrapper_agent.override(toolsets=[bar_toolset]):
tests/test_dbos.py:482:    # DBOS only wraps the MCP server toolsets. Other toolsets are not wrapped.
tests/test_dbos.py:504:    assert toolsets[3].wrapped == complex_agent.toolsets[2]
tests/test_dbos.py:506:    # Unwrapped 'external' toolset
tests/test_dbos.py:669:    # Since DBOS does not automatically wrap the tools in a workflow, and allows dynamic steps, we can pass in toolsets di
…[5883 chars omitted]…
/toolsets/wrapper.py:17:    See [toolset docs](../toolsets.md#wrapping-a-toolset) for more information.

pydantic_ai_slim/pydantic_ai/toolsets/__init__.py:11:from .prefixed import PrefixedToolset
pydantic_ai_slim/pydantic_ai/toolsets/__init__.py:29:    'PrefixedToolset',
pydantic_ai_slim/pydantic_ai/__init__.py:137:    PrefixedToolset,
pydantic_ai_slim/pydantic_ai/__init__.py:260:    'PrefixedToolset',

def process_first(output: OutputType) -> OutputType:
            output_tools_called.append('first')
            return output

        def process_second(output: OutputTypeWithCount) -> OutputTypeWithCount:  # pragma: no cover
            output_tools_called.append('second')
            return output

        def return_model(_: list[ModelMessage], info: AgentInfo) -> ModelResponse:
            assert info.output_tools is not None
            return ModelResponse(
                parts=[

                    ToolCallPart('first_output', {'value': 'valid'}),
                    ToolCallPart('second_output', {'value': 'invalid', 'count': 'not_an_int'}),
                ],
            )

        agent = Agent(
            FunctionModel(return_model),
            output_type=[
                ToolOutput(process_first, name='first_output'),
                ToolOutput(process_second
…[3023 chars omitted]…
= 'valid'

    # NOTE: When changing tests in this class:
    # 1. Follow the existing order
    # 2. Update tests in `tests/test_streaming.py::TestMultipleToolCallsStreaming` as well


async def test_model_settings_override() -> None:
    def return_settings(_: list[ModelMessage], info: AgentInfo) -> ModelResponse:
        return ModelResponse(parts=[TextPart(to_json(info.model_settings).decode())])

"""The output return JSON schema."""
        if output_type is None:
            output_type = self.output_type

        return_types = types_from_output_spec(output_spec=output_type)

        json_schemas: list[JsonSchema] = []
        for return_type in return_types:
            json_schema = TypeAdapter(return_type).json_schema(mode='serialization')
            if json_schema not in json_schemas:
                json_schemas.append(json_schema)

        if len(json_schemas) == 1:
            return json_schemas[0]
        else:
            json_schemas, all_defs = _utils.merge_json_schema_defs(json_schemas)
            json_schema: JsonSchema = {'anyOf': json_schemas}
            if all_defs:
                json_schema['$defs'] = all_defs
            return json_schema

    @overload
    async def run(
        self,
        user_prompt: str | Sequence[_messages.UserContent
…[3038 chars omitted]…
ethod builds an internal agent graph (using system prompts, tools and output schemas) and then
        runs the graph to completion. The result of the run is returned.

        Example:
        ```python
        from pydantic_ai import Agent

pydantic_ai_slim/pydantic_ai/agent/abstract.py:268:        async with self.iter(
pydantic_ai_slim/pydantic_ai/agent/abstract.py:542:        async with self.iter(

from __future__ import annotations

import asyncio
from collections.abc import Callable, Sequence
from contextlib import AsyncExitStack
from dataclasses import dataclass, field, replace
from typing import Any

from typing_extensions import Self

from .._run_context import AgentDepsT, RunContext
from ..exceptions import UserError

from .abstract import AbstractToolset, ToolsetTool


@dataclass(kw_only=True)
class _CombinedToolsetTool(ToolsetTool[AgentDepsT]):
    """A tool definition for a combined toolset tools that keeps track of the source toolset and tool."""

    source_toolset: AbstractToolset[AgentDepsT]
    source_tool: ToolsetTool[AgentDepsT]


@dataclass
class CombinedToolset(AbstractToolset[AgentDepsT]):
    """A toolset that combines multiple toolsets.

    See [toolset docs](../toolsets.md#combining-toolsets) for more information.
    """

    toolsets: Sequence[AbstractToo
…[6527 chars omitted]…
ll_tool(
        self, name: str, tool_args: dict[str, Any], ctx: RunContext[AgentDepsT], tool: ToolsetTool[AgentDepsT]
    ) -> Any:
        original_name = name.removeprefix(self.prefix + '_')
        ctx = replace(ctx, tool_name=original_name)
        tool = replace(tool, tool_def=replace(tool.tool_def, name=original_name))
        return await super().call_tool(original_name, tool_args, ctx, tool)

model_settings=input_settings,
            model_request_parameters=ModelRequestParameters(),
        ),
    )
    # Static settings are handled by get_model_settings, not before_model_request
    assert result.model_settings is input_settings


def test_abstract_capability_get_model_settings_default():
    """AbstractCapability.get_model_settings() returns None by default."""

    @dataclass

    class PlainCap(AbstractCapability[None]):
        pass

    cap = PlainCap()
    assert cap.get_model_settings() is None


def test_combined_capability_get_model_settings_merge():
    """CombinedCapability.get_model_settings() merges settings from all sub-capabilities."""
    caps = CombinedCapability(
        capabilities=[
            ModelSettings(settings=_ModelSettings(max_tokens=100)),
            ModelSettings(settings=_ModelSettings(temperature=0.5)),
        ]
    )
    m
…[6018 chars omitted]…
[Any]):
2176:        class AddSuffixCap(AbstractCapability[Any]):
2210:        class NodeObserverCap(AbstractCapability[Any]):
2226:        class NodeObserverCap(AbstractCapability[Any]):
2256:        class NodeObserverCap(AbstractCapability[Any]):
2278:        class NodeObserverCap(AbstractCapability[Any]):
2300:        class OrderedCap(AbstractCapability[Any]):
2377:class TestPrepareToolsCapability:

pydantic_ai_slim/pydantic_ai/AGENTS.md:8:- Add capability flags to model profile classes instead of inline `isinstance()` checks — prevents scattered feature detection logic — Centralizing feature support as boolean flags (e.g., `bedrock_supports_prompt_caching`) in profile classes makes capability detection maintainable and prevents duplicating provider/model type checks across usage sites.
docs/capabilities.md:81:A capability that provides tools can return a pre-built [toolset](toolsets.md) from [`get_toolset`][pydantic_ai.capabilities.AbstractCapability.get_toolset]:
docs/capabilities.md:110:    def get_toolset(self):
docs/capabilities.md:126:| [`get_toolset()`][pydantic_ai.capabilities.AbstractCapability.get_toolset] | [`AgentToolset`][pydantic_ai.toolsets.AgentToolset] ` \| None` | A [toolset](toolsets.md) to register with the agent |

if is_empty or is_thinking_only:
                    # No actionable output was returned by the model.

                    # Don't retry if the token limit was exceeded, possibly during thinking.
                    if self.model_response.finish_reason == 'length':
                        raise exceptions.UnexpectedModelBehavior(
                            f'Model token limit ({ctx.state.last_max_tokens or "provider default"}) exceeded before any response was generated. Increase the `max_tokens` model setting, or simplify the prompt to result in a shorter response that will fit within the limit.'
                        )

                    # Check for content filter on empty response
                    if is_empty and self.model_response.finish_reason == 'content_filter':
                        details = self.model_response.provider_details or {}

               
…[8628 chars omitted]…
          model_settings=model_settings,
            model_request_parameters=model_request_parameters,
        )
        self.last_request_context = request_context
        request_context = await ctx.deps.root_capability.before_model_request(
            run_context,
            request_context,
        )
        self.last_request_context = request_context
        messages = request_context.messages

timeout: failed to run command ‘curl’: No such file or directory

<class 'urllib.error.URLError'> <urlopen error [Errno -3] Temporary failure in name resolution>

from __future__ import annotations as _annotations

from collections.abc import AsyncIterator, Iterator, Sequence
from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager
from typing import Any, overload

from .. import (
    _instructions,
    _utils,
    messages as _messages,
    models,
    usage as _usage,

)
from .._json_schema import JsonSchema
from .._template import TemplateStr
from ..output import OutputDataT, OutputSpec
from ..run import AgentRun
from ..tools import (
    AgentBuiltinTool,
    AgentDepsT,
    DeferredToolResults,
    Tool,
    ToolFuncEither,
)
from ..toolsets import AbstractToolset
from .abstract import AbstractAgent, AgentMetadata, AgentModelSettings, EventStreamHandler, RunOutputDataT


class WrapperAgent(AbstractAgent[AgentDepsT, OutputDataT]):
    """Agent which wraps another agent.

    Does nothing on its own, used as a 
…[4142 chars omitted]…
| None = None,
        usage: _usage.RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        builtin_tools: Sequence[AgentBuiltinTool[AgentDepsT]] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:124:class Agent(AbstractAgent[AgentDepsT, OutputDataT]):

pydantic_ai_slim/pydantic_ai/toolsets/prepared.py:13:class PreparedToolset(WrapperToolset[AgentDepsT]):
from __future__ import annotations

from dataclasses import dataclass, replace

from .._run_context import AgentDepsT, RunContext
from ..exceptions import UserError
from ..tools import ToolsPrepareFunc
from .abstract import ToolsetTool
from .wrapper import WrapperToolset

@dataclass
class PreparedToolset(WrapperToolset[AgentDepsT]):
    """A toolset that prepares the tools it contains using a prepare function that takes the agent context and the original tool definitions.

    See [toolset docs](../toolsets.md#preparing-tool-definitions) for more information.
    """

    prepare_func: ToolsPrepareFunc[AgentDepsT]

    async def get_tools(self, ctx: RunContext[AgentDepsT]) -> dict[str, ToolsetTool[AgentDepsT]]:
        original_tools = await super().get_tools(ctx)
        original_
…[233 chars omitted]…
 len(prepared_tool_defs_by_name.keys() - original_tools.keys()) > 0:
            raise UserError(
                'Prepare function cannot add or rename tools. Use `FunctionToolset.add_function()` or `RenamedToolset` instead.'
            )

        return {
            name: replace(original_tools[name], tool_def=tool_def)
            for name, tool_def in prepared_tool_defs_by_name.items()
        }

from typing import Any

from pydantic_ai import Agent
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.models.test import TestModel


@dataclass
class KnowsCurrentTime(AbstractCapability[Any]):
    """Tells the agent what time it is."""

    def get_instructions(self):

        return f'The current date and time is {datetime.now().isoformat()}.'


agent = Agent(TestModel(), capabilities=[KnowsCurrentTime()])
result = agent.run_sync('What time is it?')
print(result.output)
#> success (no tool calls)
```

A capability that provides tools can return a pre-built [toolset](toolsets.md) from [`get_toolset`][pydantic_ai.capabilities.AbstractCapability.get_toolset]:

```python {title="custom_capability_tools.py"}
from dataclasses import dataclass
from typing import Any

from pydantic_ai import Agent
from pydantic_ai.capabilities import AbstractCapability
from pydantic_a
…[2992 chars omitted]…
.get_serialization_name()) is not None
}
"""Registry of all capability types that have a serialization name, mapping name to class."""

__all__ = [
    'AbstractCapability',
    'ModelRequestContext',
    'CAPABILITY_TYPES',
    'BuiltinTool',
    'Instructions',
    'HistoryProcessor',
    'ModelSettings',
    'PrepareTools',
    'Thinking',
    'Toolset',
    'WebSearch',
    'CombinedCapability',
]

approvals: list[ToolCallPart] = field(default_factory=list[ToolCallPart])
    """Tool calls that require human-in-the-loop approval."""
    metadata: dict[str, dict[str, Any]] = field(default_factory=dict[str, dict[str, Any]])
    """Metadata for deferred tool calls, keyed by `tool_call_id`."""


@dataclass(kw_only=True)
class ToolApproved:
    """Indicates that a tool call has been approved and that the tool function should be executed."""

    override_args: dict[str, Any] | None = None
    """Optional tool call arguments to use instead of the original arguments."""

    kind: Literal['tool-approved'] = 'tool-approved'


@dataclass
class ToolDenied:
    """Indicates that a tool call has been denied and that a denial message should be returned to the model."""

    message: str = 'The tool call was denied.'
    """The message to return to the model."""

    _: KW_ONLY

    kind: 
…[8629 chars omitted]…
set[AgentDepsT]:
        """Return the toolset to use for this run step.

        Called at the start of each run step. Override this to return a modified
        instance for per-step state transitions. If returning a new instance,
        you are responsible for managing any lifecycle transitions (exiting old
        inner toolsets, entering new ones). Default: return ``self`` (no per-step changes).

timestamp=IsNow(tz=timezone.utc),
                provider_name='test',
            ),
            ModelResponse(
                parts=[TextPart(content='The cat sat on the mat.')],
                usage=RequestUsage(input_tokens=51, output_tokens=7),
                model_name='test',
                timestamp=IsNow(tz=timezone.utc),
                provider_name='test',
            ),
            ModelResponse(
                parts=[TextPart(content='The cat sat on the mat.')],

                usage=RequestUsage(input_tokens=51, output_tokens=7),
                model_name='test',
                timestamp=IsNow(tz=timezone.utc),
                provider_name='test',
            ),
            ModelResponse(
                parts=[TextPart(content='The cat sat on the mat.')],
                usage=RequestUsage(input_tokens=51, output_tokens=7),
                mode
…[4798 chars omitted]…
ls]:
            assert info.output_tools is not None
            yield {1: DeltaToolCall('final_result', '{"bad_value": "first"}')}
            yield {2: DeltaToolCall('final_result', '{"value": "second"}')}

        agent = Agent(FunctionModel(stream_function=stream_function), output_type=OutputType, end_strategy='early')

        async with agent.run_stream('test multiple final results') as result:

Refused: `find /` walks the whole container FS. Scope under /repo.

./pydantic_ai_slim/pydantic_ai/_output.py:859:class OutputToolset(AbstractToolset[AgentDepsT]):

tests/test_agent.py:4293:            output_retries=0,  # No retries - model must succeed first try
tests/test_agent.py:4580:            output_retries=0,
tests/test_agent.py:4614:            output_retries=0,
tests/test_streaming.py:2060:            output_retries=0,  # No retries - model must succeed first try

]
        )

    def test_exhaustive_strategy_valid_first_invalid_second_output(self):
        """Test that exhaustive strategy uses the first valid output even when the second is invalid."""
        output_tools_called: list[str] = []

        def process_first(output: OutputType) -> OutputType:
            """Process first output - will be valid."""
            output_tools_called.append('first')
            return output

        def process_second(output: OutputType) -> OutputType:
            """Process second output - will be invalid."""
            output_tools_called.append('second')
            raise ModelRetry('Second output validation failed')

        def return_model(_: list[ModelMessage], info: AgentInfo) -> ModelResponse:
            assert info.output_tools is not None
            return ModelResponse(
                parts=[
                    ToolCallPar
…[3860 chars omitted]…
     timestamp=IsNow(tz=timezone.utc),
                    run_id=IsStr(),
                ),
                ModelResponse(
                    parts=[
                        ToolCallPart(tool_name='first_output', args='{"value": "valid"}', tool_call_id=IsStr()),
                        ToolCallPart(tool_name='second_output', args='{"value": "invalid"}', tool_call_id=IsStr()),
                    ],

Traceback (most recent call last):
  File "/repo/pydantic_ai_slim/pydantic_ai/_output.py", line 623, in process
    output = self.validate(data, allow_partial=allow_partial, validation_context=run_context.validation_context)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/_output.py", line 646, in validate
    return self.validator.validate_json(
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
pydantic_core._pydantic_core.ValidationError: 1 validation error for Output
  Invalid JSON: expected value at line 1 column 1 [type=json_invalid, input_value='done', input_type=str]
    For further information visit https://errors.pydantic.dev/2.14/v/json_invalid

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/repo/pydantic_ai_sli
…[4505 chars omitted]…
"/repo/pydantic_ai_slim/pydantic_ai/_agent_graph.py", line 916, in _run_stream
    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
  File "/repo/pydantic_ai_slim/pydantic_ai/_agent_graph.py", line 109, in increment_retries
    raise exceptions.UnexpectedModelBehavior(message) from error
pydantic_ai.exceptions.UnexpectedModelBehavior: Exceeded maximum retries (1) for output validation

run: (['plain_tool'], ['final_result'])
Traceback (most recent call last):
  File "/tmp/repro_cap.py", line 50, in <module>
    asyncio.run(main())
  File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
    return future.result()

           ^^^^^^^^^^^^^^^
  File "/tmp/repro_cap.py", line 46, in main
    async with agent.run_stream('hi') as result:
               ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/contextlib.py", line 210, in __aenter__
    return await anext(self.gen)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydanti
…[1532 chars omitted]…
sr/local/lib/python3.12/contextlib.py", line 210, in __aenter__
    return await anext(self.gen)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/models/function.py", line 178, in request_stream
    assert self.stream_function is not None, (
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: FunctionModel must receive a `stream_function` to support streamed requests

Traceback (most recent call last):
  File "/repo/pydantic_ai_slim/pydantic_ai/_tool_manager.py", line 457, in _raw_execute
    tool_result = await self.toolset.call_tool(
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/toolsets/combined.py", line 93, in call_tool
    return await tool.source_toolset.call_tool(name, tool_args, ctx, tool.source_tool)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/_output.py", line 973, in call_tool
    output = await self.processors[name].call(tool_args, ctx, wrap_validation_errors=False)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/_output.py", line 664, in call
    output = await execute_traced_output_function(

             ^^^^^^^^^^^^^^^
…[4655 chars omitted]…
ydantic_ai_slim/pydantic_ai/_agent_graph.py", line 1186, in process_tool_calls
    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
  File "/repo/pydantic_ai_slim/pydantic_ai/_agent_graph.py", line 109, in increment_retries
    raise exceptions.UnexpectedModelBehavior(message) from error
pydantic_ai.exceptions.UnexpectedModelBehavior: Exceeded maximum retries (2) for output validation

from __future__ import annotations as _annotations

import inspect
import json
from abc import ABC, abstractmethod
from collections.abc import Awaitable, Callable, Sequence
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Generic, Literal, cast, overload

from pydantic import Json, TypeAdapter, ValidationError
from pydantic_core import SchemaValidator, to_json
from typing_extensions import Self, TypedDict, TypeVar

from pydantic_ai._instrumentation import InstrumentationNames
from pydantic_ai._utils import get_function_type_hints

from . import _function_schema, _utils, messages as _messages
from ._run_context import AgentDepsT, RunContext
from .exceptions import ModelRetry, ToolRetryError, UserError
from .output import (
    DeferredToolRequests,
    NativeOutput,
    OutputDataT,
    OutputMode,
    OutputObjectDefinition,
    OutputSpec,
    OutputTyp
…[114 chars omitted]…
 ignore[reportPrivateUsage]
)
from .tools import GenerateToolJsonSchema, ObjectJsonSchema, ToolDefinition
from .toolsets.abstract import AbstractToolset, ToolsetTool

if TYPE_CHECKING:
    pass

T = TypeVar('T')
"""An invariant TypeVar."""
OutputDataT_inv = TypeVar('OutputDataT_inv', default=str)
"""
An invariant type variable for the result data of a model.

7:from dataclasses import dataclass, field

diff --git a/pydantic_ai_slim/pydantic_ai/capabilities/abstract.py b/pydantic_ai_slim/pydantic_ai/capabilities/abstract.py
index 510a977..1601e74 100644
--- a/pydantic_ai_slim/pydantic_ai/capabilities/abstract.py
+++ b/pydantic_ai_slim/pydantic_ai/capabilities/abstract.py
@@ -9,7 +9,7 @@ from pydantic_ai._instructions import AgentInstructions
 from pydantic_ai.messages import AgentStreamEvent, ModelResponse, ToolCallPart
 from pydantic_ai.models import ModelRequestContext
 from pydantic_ai.tools import AgentBuiltinTool, AgentDepsT, RunContext, ToolDefinition
-from pydantic_ai.toolsets import AgentToolset
+from pydantic_ai.toolsets import AbstractToolset, AgentToolset
 
 if TYPE_CHECKING:

     from pydantic_ai import _agent_graph
@@ -93,6 +93,20 @@ class AbstractCapability(ABC, Generic[AgentDepsT]):
         """Return a toolset to register with the agent, or None."""
         return No
…[1694 chars omitted]…
ities:
+            wrapped = capability.get_wrapper_toolset(toolset)
+            if wrapped is not None:
+                toolset = wrapped
+                wrapped_toolset = wrapped
+        return wrapped_toolset
+
     def get_builtin_tools(self) -> Sequence[AgentBuiltinTool[AgentDepsT]]:
         builtin_tools: list[AgentBuiltinTool[AgentDepsT]] = []
         for capability in self.capabilities:

diff --git a/pydantic_ai_slim/pydantic_ai/agent/__init__.py b/pydantic_ai_slim/pydantic_ai/agent/__init__.py
index da580bb..6f4829f 100644
--- a/pydantic_ai_slim/pydantic_ai/agent/__init__.py
+++ b/pydantic_ai_slim/pydantic_ai/agent/__init__.py
@@ -1107,6 +1107,7 @@ class Agent(AbstractAgent[AgentDepsT, OutputDataT]):
             output_toolset=output_toolset,
             additional_toolsets=toolsets,
             cap_toolsets=cap_toolsets,
+            root_capability=run_capability,
         )
         toolset = await toolset.for_run(initial_ctx)
         tool_manager = ToolManager[AgentDepsT](

@@ -2095,6 +2096,7 @@ class Agent(AbstractAgent[AgentDepsT, OutputDataT]):
         output_toolset: AbstractToolset[AgentDepsT] | None | _utils.Unset = _utils.UNSET,
         additional_toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
         cap_toolsets: Sequence[AgentTools
…[1110 chars omitted]…
pare_tools)
 
+        if root_capability is not None:
+            wrapped_toolset = root_capability.get_wrapper_toolset(toolset)
+            if wrapped_toolset is not None:
+                toolset = wrapped_toolset
+
         output_toolset = output_toolset if _utils.is_set(output_toolset) else self._output_toolset
         if output_toolset is not None:
             if self._prepare_output_tools:

diff --git a/pydantic_ai_slim/pydantic_ai/_output.py b/pydantic_ai_slim/pydantic_ai/_output.py
index 87f93c4..9ee4386 100644
--- a/pydantic_ai_slim/pydantic_ai/_output.py
+++ b/pydantic_ai_slim/pydantic_ai/_output.py
@@ -864,6 +864,7 @@ class OutputToolset(AbstractToolset[AgentDepsT]):
     processors: dict[str, ObjectOutputProcessor[Any]]
     """The processors for the output tools in this toolset."""
     max_retries: int
+    _tool_max_retries: dict[str, int | None]
     output_validators: list[OutputValidator[AgentDepsT, Any]]
 
     @classmethod

@@ -879,6 +880,7 @@ class OutputToolset(AbstractToolset[AgentDepsT]):
 
         processors: dict[str, ObjectOutputProcessor[Any]] = {}
         tool_defs: list[ToolDefinition] = []
+        tool_max_retries: dict[str, int | None] = {}
 
         default_name = name or DEFAULT_OUTPUT_TOOL_NAME
         default_description = description
@@
…[1825 chars omitted]…
          max_retries=self.max_retries,
+                max_retries=(
+                    self._tool_max_retries[tool_def.name]
+                    if self._tool_max_retries.get(tool_def.name) is not None
+                    else self.max_retries
+                ),
                 args_validator=self.processors[tool_def.name].validator,
             )
             for tool_def in self._tool_defs

@dataclasses.dataclass(kw_only=True)
class GraphAgentState:
    """State kept across the execution of the agent graph."""

    message_history: list[_messages.ModelMessage] = dataclasses.field(default_factory=list[_messages.ModelMessage])
    usage: _usage.RunUsage = dataclasses.field(default_factory=_usage.RunUsage)
    retries: int = 0
    output_retries: dict[str, int] = dataclasses.field(default_factory=dict[str, int])
    run_step: int = 0
    run_id: str = dataclasses.field(default_factory=lambda: str(uuid.uuid4()))
    metadata: dict[str, Any] | None = None
    last_max_tokens: int | None = None

    """Last-resolved `max_tokens` from model settings, used only in error messages."""

    def increment_retries(
        self,
        max_result_retries: int,
        error: BaseException | None = None,
    ) -> None:
        self.retries += 1
        if self.retries > max_result_r
…[1797 chars omitted]…
     error = error.__cause__
            raise exceptions.UnexpectedModelBehavior(message) from error
        else:
            raise exceptions.UnexpectedModelBehavior(message)


@dataclasses.dataclass(kw_only=True)
class GraphAgentDeps(Generic[DepsT, OutputDataT]):
    """Dependencies/config passed to the agent graph."""

    user_deps: DepsT

    prompt: str | Sequence[_messages.UserContent] | None

1144:    # First, we handle output tool calls
1145-    for call in tool_calls_by_kind['output']:
1146-        # `final_result` can be passed into `process_tool_calls` from `Agent.run_stream`
1147-        # when streaming and there's already a final result
1148-        if final_result and final_result.tool_call_id == call.tool_call_id:
1149-            part = _messages.ToolReturnPart(
1150-                tool_name=call.tool_name,
1151-                content='Final result processed.',
1152-                tool_call_id=call.tool_call_id,
1153-            )
1154-            output_parts.append(part)
1155-        # Early strategy is chosen and final result is already set

1156-        elif ctx.deps.end_strategy == 'early' and final_result:
1157-            for event in _emit_skipped_output_tool(
1158-                call, 'Output tool not used - a final result was already processed.', out
…[4956 chars omitted]…
ls_to_run_results: dict[str, DeferredToolResult] = {}
1249-    if tool_call_results is not None:
1250-        # Deferred tool calls are "run" as well, by reading their value from the tool call results
1251-        calls_to_run.extend(tool_calls_by_kind['external'])
1252-        calls_to_run.extend(tool_calls_by_kind['unapproved'])
1253-
1254-        result_tool_call_ids = set(tool_call_results.keys())

part = _messages.ToolReturnPart(
        tool_name=call.tool_name,
        content=message,
        tool_call_id=call.tool_call_id,
    )
    output_parts.append(part)
    yield _messages.FunctionToolResultEvent(part)


def _get_output_tool_max_retries(
    tool_manager: ToolManager[DepsT],
    call: _messages.ToolCallPart,

    default: int,
) -> int:
    if tool_manager.tools is not None:
        tool = tool_manager.tools.get(call.tool_name)
        if tool is not None:
            return tool.max_retries
    return default


async def process_tool_calls(  # noqa: C901
    tool_manager: ToolManager[DepsT],
    tool_calls: list[_messages.ToolCallPart],
    tool_call_results: dict[str, DeferredToolResult | Literal['skip']] | None,
    tool_call_metadata: dict[str, dict[str, Any]] | None,
    final_result: result.FinalResult[NodeRunEndT] | None,
    ctx: GraphRunContext[GraphAgentSt
…[4658 chars omitted]…

                )
                raise  # pragma: lax no cover
            except ToolRetryError as e:
                # If we already have a valid final result, don't increment retries for invalid output tools
                # This allows the run to succeed if at least one output tool returned a valid result
                if not final_result:
                    assert validated.tool is not None

output: Weather in Mexico City
retries: [0, 1, 2, 3, 4, 5]
max_retries: [5, 5, 5, 5, 5, 5]

run: (['b_a_plain_tool'], ['final_result'])
stream: (['b_a_plain_tool'], ['final_result'])

........................................................................ [ 53%]
...............................................................          [100%]
135 passed in 5.35s

13:	uv sync --frozen --all-extras --no-extra outlines-vllm-offline --all-packages --group lint --group docs
18:	UV_PROJECT_ENVIRONMENT=.venv310 uv sync --python 3.10 --frozen --all-extras --no-extra outlines-vllm-offline --all-packages --group lint --group docs
19:	UV_PROJECT_ENVIRONMENT=.venv311 uv sync --python 3.11 --frozen --all-extras --no-extra outlines-vllm-offline --all-packages --group lint --group docs
20:	UV_PROJECT_ENVIRONMENT=.venv312 uv sync --python 3.12 --frozen --all-extras --no-extra outlines-vllm-offline --all-packages --group lint --group docs
21:	UV_PROJECT_ENVIRONMENT=.venv313 uv sync --python 3.13 --frozen --all-extras --no-extra outlines-vllm-offline --all-packages --group lint --group docs
25:	uv sync --all-extras --no-extra outlines-vllm-offline --all-packages --group lint --group docs
27:.PHONY: format
28:format: ## Format the code
29:	uv run ruff format
30:
…[133 chars omitted]…
:.PHONY: typecheck-pyright

38:typecheck-pyright:
41:	PYRIGHT_PYTHON_IGNORE_WARNINGS=1 uv run pyright $(if $(PYRIGHT_PYTHON),--pythonversion $(PYRIGHT_PYTHON))
48:typecheck: typecheck-pyright ## Run static type checking
51:typecheck-both: typecheck-pyright typecheck-mypy
103:all: format lint typecheck testcov ## Run code formatting, linting, static type checks, and tests with coverage report generation

..............................x..xx..................................... [ 93%]
.....                                                                    [100%]
74 passed, 3 xfailed in 5.87s

..................                                                       [100%]
18 passed in 3.85s

self._tool_max_retries = tool_max_retries or {}
        self.output_validators = output_validators or []

    @property
    def id(self) -> str | None:
        return '<output>'  # pragma: no cover

    @property
    def label(self) -> str:
        return "the agent's output tools"

    def _get_max_retries(self, tool_name: str) -> int:

        tool_max_retries = self._tool_max_retries.get(tool_name)
        return self.max_retries if tool_max_retries is None else tool_max_retries

    async def get_tools(self, ctx: RunContext[AgentDepsT]) -> dict[str, ToolsetTool[AgentDepsT]]:
        return {
            tool_def.name: ToolsetTool(
                toolset=self,
                tool_def=tool_def,
                max_retries=self._get_max_retries(tool_def.name),
                args_validator=self.processors[tool_def.name].validator,
            )
            for tool_def in self._tool_defs
        }

    async def call_tool(
        self, name: str, tool_args: dict[str, Any], ctx: RunContext[AgentDepsT], tool: ToolsetTool[AgentDepsT]
    ) -> Any:
        output = await self.processors[name].call(tool_args, ctx, wrap_validation_errors=False)
        for validator in self.output_validators:

output: ok
calls: [('first', 0, 1), ('second', 0, 5), ('second', 1, 5), ('second', 2, 5), ('second', 3, 5), ('second', 4, 5), ('second', 5, 5)]

UnexpectedModelBehavior Exceeded maximum retries (5) for output validation

49:from pydantic_ai.exceptions import ApprovalRequired, CallDeferred, ModelRetry
1967:            raise ModelRetry('First output validation failed')
2046:            raise ModelRetry('Second output validation failed')
2060:            output_retries=0,  # No retries - model must succeed first try
2122:            """Process second output - will raise ModelRetry."""
2124:            raise ModelRetry('Second output validation failed')
2138:            output_retries=1,  # Allow 1 retry so ToolRetryError is raised
3389:            raise ModelRetry('Validation failed: x must be positive')

content='Final result processed.',
                            tool_call_id=IsStr(),
                            timestamp=IsNow(tz=timezone.utc),
                        ),
                        ToolReturnPart(
                            tool_name='second_output',
                            content='Output tool not used - output function execution failed.',
                            tool_call_id=IsStr(),
                            timestamp=IsNow(tz=timezone.utc),
                        ),
                    ],
                    timestamp=IsNow(tz=timezone.utc),

                    run_id=IsStr(),
                ),
            ]
        )

    async def test_exhaustive_strategy_with_tool_retry_and_final_result(self):
        """Test that exhaustive strategy doesn't increment retries when `final_result` exists and `ToolRetryError` occurs."""
   
…[4351 chars omitted]…
ts == snapshot(
        [
            PartStartEvent(
                index=0,
                part=ToolCallPart(tool_name='add_numbers', args={'x': 0, 'y': 0}, tool_call_id=IsStr()),
            ),
            PartEndEvent(
                index=0,
                part=ToolCallPart(tool_name='add_numbers', args={'x': 0, 'y': 0}, tool_call_id=IsStr()),
            ),
            FunctionToolCallEvent(

pydantic_ai_slim/pydantic_ai/_agent_graph.py       | 92 ++++++++++++++++------
 pydantic_ai_slim/pydantic_ai/_output.py            | 15 +++-
 pydantic_ai_slim/pydantic_ai/agent/__init__.py     |  9 +++
 .../pydantic_ai/capabilities/abstract.py           | 16 +++-
 .../pydantic_ai/capabilities/combined.py           |  9 +++
 5 files changed, 114 insertions(+), 27 deletions(-)
diff --git a/pydantic_ai_slim/pydantic_ai/_agent_graph.py b/pydantic_ai_slim/pydantic_ai/_agent_graph.py
index ae9801c..d1dc557 100644
--- a/pydantic_ai_slim/pydantic_ai/_agent_graph.py
+++ b/pydantic_ai_slim/pydantic_ai/_agent_graph.py
@@ -76,6 +76,7 @@ class GraphAgentState:
     message_history: list[_messages.ModelMessage] = dataclasses.field(default_factory=list[_messages.ModelMessage])

     usage: _usage.RunUsage = dataclasses.field(default_factory=_usage.RunUsage)
     retries: int = 0
+    output_retries:
…[8629 chars omitted]…
ities:
+            wrapped = capability.get_wrapper_toolset(toolset)
+            if wrapped is not None:
+                toolset = wrapped
+                wrapped_toolset = wrapped
+        return wrapped_toolset
+
     def get_builtin_tools(self) -> Sequence[AgentBuiltinTool[AgentDepsT]]:
         builtin_tools: list[AgentBuiltinTool[AgentDepsT]] = []
         for capability in self.capabilities:

tests/test_streaming.py:2430:    state = GraphAgentState(metadata=metadata)
pydantic_ai_slim/pydantic_ai/agent/__init__.py:1030:        state = _agent_graph.GraphAgentState(
grep: tests/__pycache__/test_agent.cpython-312-pytest-9.2.0.pyc: binary file matches
grep: tests/__pycache__/test_capabilities.cpython-312-pytest-9.2.0.pyc: binary file matches
grep: tests/__pycache__/test_streaming.cpython-312-pytest-9.2.0.pyc: binary file matches
grep: pydantic_ai_slim/pydantic_ai/agent/__pycache__/spec.cpython-312.pyc: binary file matches
grep: pydantic_ai_slim/pydantic_ai/agent/__pycache__/__init__.cpython-312.pyc: binary file matches
grep: pydantic_ai_slim/pydantic_ai/__pycache__/_agent_graph.cpython-312.pyc: binary file matches
tests/test_agent.py:382:    agent = Agent(FunctionModel(return_model), output_type=Foo, output_retries=target_retries)
tests/test_agent.py:419:    agent = Agent(Funct
…[3215 chars omitted]…
       ctx.state.increment_output_retries(

pydantic_ai_slim/pydantic_ai/_agent_graph.py:1207:                ctx.state.increment_output_retries(
pydantic_ai_slim/pydantic_ai/_agent_graph.py:1225:                ctx.state.increment_output_retries(
pydantic_ai_slim/pydantic_ai/_agent_graph.py:1236:                    ctx.state.increment_output_retries(call.tool_name, validated.tool.max_retries, error=e)

tool_call_id=IsStr(),
                        timestamp=IsNow(tz=timezone.utc),
                    )
                ],
                timestamp=IsNow(tz=timezone.utc),
                run_id=IsStr(),
            ),
            ModelResponse(
                parts=[ToolCallPart(tool_name='foobar', args='{}', tool_call_id=IsStr())],
                usage=RequestUsage(input_tokens=65, output_tokens=4),
                model_name='function:empty:',
                timestamp=IsNow(tz=timezone.utc),

                run_id=IsStr(),
            ),
            ModelRequest(
                parts=[
                    RetryPromptPart(
                        tool_name='foobar',
                        content="Unknown tool name: 'foobar'. No tools available.",
                        tool_call_id=IsStr(),
                        timestamp=IsNow(tz=timezone.utc),
     
…[1765 chars omitted]…
guments\.',
    ):
        agent.run_sync('Hello')


def test_tool_exceeds_token_limit_but_complete_args():
    def return_complete_tool_but_hit_limit(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
        if len(messages) == 1:
            resp = ModelResponse(parts=[ToolCallPart('dummy_tool', args='{"foo": "bar"}')])
            resp.finish_reason = 'length'
            return resp

test_a2a.py
test_ag_ui.py
test_agent.py
test_agent_output_schemas.py
test_builtin_tools.py
test_capabilities.py
test_cassette_utils.py
test_cli.py
test_concurrency.py
test_dbos.py
test_deps.py
test_direct.py

test_embeddings.py
test_examples.py
test_exceptions.py
test_fastmcp.py
test_format_as_xml.py
test_function_schema.py
test_history_processor.py
test_json_body_serializer.py
test_json_schema.py
test_logfire.py
test_mcp.py
test_messages.py
test_native_output_schema.py
test_parts_manager.py
test_prefect.py
test_settings.py
test_spec.py
test_ssrf.py
test_streaming.py
test_tavily.py
test_template.py
test_temporal.py
test_tenacity.py
test_thinking_part.py
test_tools.py
test_toolsets.py
test_ui.py
test_ui_web.py
test_usage_limits.py
test_utils.py
test_validation_context.py
test_vercel_ai.py

ssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssssss [ 66%]
sssssssssssssssssssssssssssssssssssss                                    [100%]
109 skipped in 3.35s

OK

Traceback (most recent call last):
  File "/tmp/repro_prepare.py", line 40, in <module>
    asyncio.run(main())
  File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^

  File "/tmp/repro_prepare.py", line 37, in main
    await agent.run('hi')
  File "/repo/pydantic_ai_slim/pydantic_ai/agent/abstract.py", line 293, in run
    node = await agent_run.next(node)  # pyright: ignore[reportArgumentType]
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/run.py", line 304, in next
    re
…[3131 chars omitted]…
ed.get_tools(ctx)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/toolsets/prepared.py", line 25, in get_tools
    tool_def.name: tool_def for tool_def in (await self.prepare_func(ctx, original_tool_defs) or [])
                                             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
TypeError: object list can't be used in 'await' expression

26-    'ToolFuncEither',
27-    'ToolParams',
28-    'ToolPrepareFunc',
29:    'ToolsPrepareFunc',
30-    'AgentBuiltinTool',
31-    'BuiltinToolFunc',
32-    'Tool',
33-    'ObjectJsonSchema',
34-    'ToolDefinition',
--
110-Usage `ToolPrepareFunc[AgentDepsT]`.
111-"""

112-
113:ToolsPrepareFunc: TypeAlias = Callable[
114-    [RunContext[AgentDepsT], list['ToolDefinition']], Awaitable['list[ToolDefinition] | None']
115-]
116-"""Definition of a function that can prepare the tool definition of all tools for each step.
117-This is useful if you want to customize the definition of multiple tools or you want to register
118-a subset of tools for a given step.
119-
120:Example — here `turn_on_strict_if_openai` is valid as a `ToolsPrepareFunc`:
121-
122-```python {noqa="I001"}
123-from dataclasses import replace
124-
125-from pydantic_ai import Agent, RunContext
--
136-agent = Agent('openai:gpt-5.2', prepare_tools=turn_on_strict_if_openai)
137-```
138-

['p_keep']

19:334-    [`get_output`][pydantic_ai.result.StreamedRunResult.get_output] completes.
213:528-    async def get_output(self) -> OutputDataT:
220:535-            output = await self._stream_response.get_output()
374:736-    def get_output(self) -> OutputDataT:
376:738-        return _utils.get_event_loop().run_until_complete(self._streamed_run_result.get_output())
419:781-        [`get_output`][pydantic_ai.result.StreamedRunResultSync.get_output] completes.
176:    async def get_output(self) -> OutputDataT:
177-        """Stream the whole response, validate the output and return it."""
178-        if self._cached_output is not None:
179-            return deepcopy(self._cached_output)
180-
181-        # Iterate through any stream events

182-        async for _ in self:
183-            pass
184-
185-        # Final validation with `allow_partial=False` (default)
186-        self._cached
…[7259 chars omitted]…
llow_partial: bool = False
598-    ) -> OutputDataT:
599-        return await self.validate_response_output(message, allow_partial=allow_partial)
600-
601-    async def validate_response_output(
602-        self, message: _messages.ModelResponse, *, allow_partial: bool = False
603-    ) -> OutputDataT:
604-        """Validate a structured result message."""
605-        if self._run_result is not None:

toolsets=toolsets,
            builtin_tools=builtin_tools,
        ) as agent_run:
            # Handle wrap_run short-circuit: result is already available
            if agent_run.result is not None:
                graph_ctx = agent_run.ctx
                yield StreamedRunResult(
                    graph_ctx.state.message_history,
                    graph_ctx.deps.new_message_index,
                    run_result=agent_run.result,
                )
                yielded = True

            first_node = agent_run.next_node  # start with the first node
            assert isinstance(first_node, _agent_graph.UserPromptNode)  # the first node should be a user prompt node
            node: _agent_graph.AgentNode[Any, Any] = first_node
            while not yielded:
                graph_ctx = agent_run.ctx
                if self.is_model_request_node(node):
            
…[6287 chars omitted]…
StreamedRunResult before getting here'
                    )
                node = cast(_agent_graph.AgentNode[Any, Any], next_node)

        if not yielded:
            raise exceptions.AgentRunError('Agent run finished without producing a final result')  # pragma: no cover

    @overload
    def run_stream_sync(
        self,
        user_prompt: str | Sequence[_messages.UserContent] | None = None,

........................................................................ [ 12%]
.................s...................................................... [ 25%]
................sss...s................................................. [ 38%]
..............................x..xx..................................... [ 50%]
........................................................................ [ 63%]
........................................................................ [ 76%]
........................................................................ [ 89%]
....................................ss........................           [100%]
556 passed, 7 skipped, 3 xfailed in 10.21s

M pydantic_ai_slim/pydantic_ai/_agent_graph.py
 M pydantic_ai_slim/pydantic_ai/_output.py
 M pydantic_ai_slim/pydantic_ai/agent/__init__.py
 M pydantic_ai_slim/pydantic_ai/capabilities/abstract.py
 M pydantic_ai_slim/pydantic_ai/capabilities/combined.py

Weather in Mexico City [(0, 5), (1, 5), (2, 5), (3, 5), (4, 5), (5, 5)]

48:from ..capabilities import AbstractCapability, CombinedCapability
49:from ..capabilities.builtin_tool import BuiltinTool as BuiltinToolCap
50:from ..capabilities.history_processor import HistoryProcessor as HistoryProcessorCap
229:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
261:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
291:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
369:            capabilities: Optional list of [capabilities](https://ai.pydantic.dev/capabilities/) to configure the agent with.
370:                Custom capabilities can be created by subclassing
371:                [`AbstractCapability`][pydantic_ai.capabilities.AbstractCapability].
384:        capabilities = list(capabilities or [])
386:            capabilities.append(HistoryProcessorCap(history_processor))
388:

…[738 chars omitted]…
s.append(capability)
663:        if capabilities:
664:            all_capabilities.extend(capabilities)
694:            capabilities=all_capabilities,
728:        capabilities: Sequence[AbstractCapability[Any]] | None = None,
763:        capabilities: Sequence[AbstractCapability[Any]] | None = None,
1092:            # Capability settings (e.g. from Thinking, ModelSettings capabilities), cached at init

48:from ..capabilities import AbstractCapability, CombinedCapability
49:from ..capabilities.builtin_tool import BuiltinTool as BuiltinToolCap
50:from ..capabilities.history_processor import HistoryProcessor as HistoryProcessorCap
229:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
261:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
291:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
369:            capabilities: Optional list of [capabilities](https://ai.pydantic.dev/capabilities/) to configure the agent with.
370:                Custom capabilities can be created by subclassing
371:                [`AbstractCapability`][pydantic_ai.capabilities.AbstractCapability].
384:        capabilities = list(capabilities or [])
386:            capabilities.append(HistoryProcessorCap(history_processor))
388:

…[7292 chars omitted]…
s search on DeepWiki: https://deepwiki.com/search/what-is-this-repository-about_5104a64d-2f5e-4461-80d8-eb0892242441\n"}]} }
docs/capabilities.md:504:By default, a capability instance is shared across all runs of an agent. If your capability accumulates mutable state that should not leak between runs, override [`for_run`][pydantic_ai.capabilities.AbstractCapability.for_run] to return a fresh instance:

def override(  # noqa: C901
        self,
        *,
        name: str | _utils.Unset = _utils.UNSET,
        deps: AgentDepsT | _utils.Unset = _utils.UNSET,
        model: models.Model | models.KnownModelName | str | _utils.Unset = _utils.UNSET,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset = _utils.UNSET,
        tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] | _utils.Unset = _utils.UNSET,
        instructions: AgentInstructions[AgentDepsT] | _utils.Unset = _utils.UNSET,
        metadata: AgentMetadata[AgentDepsT] | _utils.Unset = _utils.UNSET,
        model_settings: AgentModelSettings[AgentDepsT] | _utils.Unset = _utils.UNSET,
    ) -> Iterator[None]:

        """Context manager to temporarily override agent name, dependencies, model, toolsets, tools, or instructions.

        This is particularly useful when testing.
        You ca
…[4513 chars omitted]…
ntext`][pydantic_ai.tools.RunContext] as its only argument.
        Can decorate a sync or async functions.

        The decorator can be used bare (`agent.instructions`).

        Overloads for every possible signature of `instructions` are included so the decorator doesn't obscure
        the type of the function.

        Example:
        ```python
        from pydantic_ai import Agent, RunContext

grep: pydantic_ai_slim/pydantic_ai/agent/__pycache__/__init__.cpython-312.pyc: binary file matches
pydantic_ai_slim/pydantic_ai/agent/__init__.py:390:        self._root_capability = CombinedCapability(capabilities)
pydantic_ai_slim/pydantic_ai/agent/__init__.py:411:        self._cap_instructions = _instructions.normalize_instructions(self._root_capability.get_instructions())
pydantic_ai_slim/pydantic_ai/agent/__init__.py:423:        self._cap_builtin_tools = list(self._root_capability.get_builtin_tools())
pydantic_ai_slim/pydantic_ai/agent/__init__.py:425:        self._cap_model_settings = self._root_capability.get_model_settings()
pydantic_ai_slim/pydantic_ai/agent/__init__.py:451:        cap_toolset = self._root_capability.get_toolset()
pydantic_ai_slim/pydantic_ai/agent/__init__.py:1065:        run_capability = await self._root_capability.for_run(initial_ctx)
pydantic_ai_slim/pydantic_ai/agent/__init__.py:1067:        if run_capability is not self._root_capability:
pydantic_ai_slim/pydantic_ai/capabilities/combined.py:25:class CombinedCapability(AbstractCapability[AgentDepsT]):

grep: docs/api/capabilities.md: No such file or directory
docs/capabilities.md:51:To build your own capability, subclass [`AbstractCapability`][pydantic_ai.capabilities.AbstractCapability] and override the methods you need. There are two categories: **configuration methods** that are called once at agent construction, and **lifecycle hooks** that fire during each run.
docs/capabilities.md:185:| [`before_run`][pydantic_ai.capabilities.AbstractCapability.before_run] | `(ctx: RunContext) -> None` | Observe-only notification that a run is starting |
docs/capabilities.md:186:| [`after_run`][pydantic_ai.capabilities.AbstractCapability.after_run] | `(ctx: RunContext, *, result: AgentRunResult) -> AgentRunResult` | Modify the final result |
docs/capabilities.md:187:| [`wrap_run`][pydantic_ai.capabilities.AbstractCapability.wrap_run] | `(ctx: RunContext, *, handler: () -> AgentRunResult) -> Ag
…[988 chars omitted]…
ream: AsyncIterable[AgentStreamEvent]) -> AsyncIterable[AgentStreamEvent]` | Observe, filter, or transform streamed events |
docs/capabilities.md:504:By default, a capability instance is shared across all runs of an agent. If your capability accumulates mutable state that should not leak between runs, override [`for_run`][pydantic_ai.capabilities.AbstractCapability.for_run] to return a fresh instance:

==================================== ERRORS ====================================
______________________ ERROR collecting tests/test_a2a.py ______________________
ImportError while importing test module '/repo/tests/test_a2a.py'.
Hint: make sure your test modules/packages have valid Python names.
Traceback:
/usr/local/lib/python3.12/importlib/__init__.py:90: in import_module
    return _bootstrap._gcd_import(name[level:], package, level)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
tests/test_a2a.py:8: in <module>
    from asgi_lifespan import LifespanManager
E   ModuleNotFoundError: No module named 'asgi_lifespan'
=========================== short test summary info ============================

ERROR tests/test_a2a.py
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
1 error in 7.77s

else:
            # Validate and execute the output tool call — unlike deferred tools,
            # output tools track retries and can be skipped if a final_result exists.
            try:
                validated = await tool_manager.validate_tool_call(call)
            except exceptions.UnexpectedModelBehavior as e:
                # If we already have a valid final result, don't fail the entire run
                # This allows exhaustive strategy to complete successfully when at least one output tool is valid
                if final_result:
                    # If output tool fails when we already have a final result, skip it without retrying
                    for event in _emit_skipped_output_tool(
                        call, 'Output tool not used - output failed validation.', output_parts, args_valid=False

                    ):
                        yield even
…[7458 chars omitted]…
elf.processors[tool_def.name].validator,
            )
            for tool_def in self._tool_defs
        }

    async def call_tool(
        self, name: str, tool_args: dict[str, Any], ctx: RunContext[AgentDepsT], tool: ToolsetTool[AgentDepsT]
    ) -> Any:
        output = await self.processors[name].call(tool_args, ctx, wrap_validation_errors=False)
        for validator in self.output_validators:

Exceeded maximum retries (5) for output validation
[(0, 5), (1, 5), (2, 5), (3, 5), (4, 5), (5, 5)]

48:from ..capabilities import AbstractCapability, CombinedCapability
229:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
261:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
291:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
371:                [`AbstractCapability`][pydantic_ai.capabilities.AbstractCapability].
390:        self._root_capability = CombinedCapability(capabilities)
487:        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
511:        capabilities: Sequence[AbstractCapability[Any]] | None = None,
521:        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
545:        capabilities: Sequence[AbstractCapability[Any]] | None = None,
554:        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
578:        capabilities:

…[200 chars omitted]…
y[Any]] = []
704:        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
728:        capabilities: Sequence[AbstractCapability[Any]] | None = None,
739:        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
763:        capabilities: Sequence[AbstractCapability[Any]] | None = None,
2099:        root_capability: AbstractCapability[AgentDepsT] | None = None,

agent = Agent(
            FunctionModel(stream_function=stream_function),
            output_type=[
                ToolOutput(process_first, name='first_output'),
                ToolOutput(process_second, name='second_output'),
            ],
            end_strategy='exhaustive',
        )

        async with agent.run_stream('test exhaustive output tools') as result:
            response = await result.get_output()

        # Verify the result came from the first output tool
        assert isinstance(response, OutputType)
        assert response.value == 'first'

        # Verify both output tools were called
        assert output_tools_called == ['first', 'second']

        # Verify we got tool returns in the correct order
        assert result.all_messages() == snapshot(
            [
                ModelRequest(
                    parts=[UserPromptPart(content='test
…[6384 chars omitted]…
e result came from the first output tool (second was invalid, but we ignore it)
        assert isinstance(response, OutputType)
        assert response.value == snapshot('valid')

        # Verify both output tools were called
        assert output_tools_called == snapshot(['first', 'second'])

        # Verify we got appropriate messages
        assert result.all_messages() == snapshot(
            [

Weather in Mexico City model calls: 6

['ax_tool']

PrefixedToolset b a
None
None

ToolRetryError nope
[(0, 5)]

{'a': 5, 'b': 2}

Exceeded maximum retries (5) for output validation

Exceeded maximum retries (1) for output validation

........................................................................ [ 12%]
.................s...................................................... [ 24%]
................sss...s................................................. [ 36%]
..............................x..xx..................................... [ 49%]
........................................................................ [ 61%]
........................................................................ [ 73%]
........................................................................ [ 86%]
....................................ss.................................. [ 98%]
........                                                                 [100%]
574 passed, 7 skipped, 3 xfailed in 10.55s

* master
7c4cd2c base

pydantic_ai_slim/pydantic_ai/_agent_graph.py       | 93 ++++++++++++++++------
 pydantic_ai_slim/pydantic_ai/_output.py            | 16 +++-
 pydantic_ai_slim/pydantic_ai/agent/__init__.py     |  9 +++
 .../pydantic_ai/capabilities/abstract.py           | 16 +++-
 .../pydantic_ai/capabilities/combined.py           |  9 +++
 5 files changed, 116 insertions(+), 27 deletions(-)
 M pydantic_ai_slim/pydantic_ai/_agent_graph.py
 M pydantic_ai_slim/pydantic_ai/_output.py
 M pydantic_ai_slim/pydantic_ai/agent/__init__.py
 M pydantic_ai_slim/pydantic_ai/capabilities/abstract.py
 M pydantic_ai_slim/pydantic_ai/capabilities/combined.py

diff --git a/pydantic_ai_slim/pydantic_ai/_agent_graph.py b/pydantic_ai_slim/pydantic_ai/_agent_graph.py
index ae9801c..6369918 100644
--- a/pydantic_ai_slim/pydantic_ai/_agent_graph.py
+++ b/pydantic_ai_slim/pydantic_ai/_agent_graph.py
@@ -76,6 +76,8 @@ class GraphAgentState:
     message_history: list[_messages.ModelMessage] = dataclasses.field(default_factory=list[_messages.ModelMessage])
     usage: _usage.RunUsage = dataclasses.field(default_factory=_usage.RunUsage)
     retries: int = 0
+    output_retries: dict[str, int] = dataclasses.field(default_factory=dict[str, int])
+    """Per-output-tool retry counts, tracked independently of each other and of `retries`."""
     run_step: int = 0
     run_id: str = dataclasses.field(default_factory=lambda: str(uuid.uuid4()))

     metadata: dict[str, Any] | None = None
@@ -89,26 +91,45 @@ class GraphAgentState:
     ) -> None:
         s
…[8629 chars omitted]…
ities:
+            wrapped = capability.get_wrapper_toolset(toolset)
+            if wrapped is not None:
+                toolset = wrapped
+                wrapped_toolset = wrapped
+        return wrapped_toolset
+
     def get_builtin_tools(self) -> Sequence[AgentBuiltinTool[AgentDepsT]]:
         builtin_tools: list[AgentBuiltinTool[AgentDepsT]] = []
         for capability in self.capabilities:

ok [(0, 2), (1, 2), (2, 2)]

…[414 chars omitted]…
n_stream, prepare_tools ordering, explicit/default retry limits, multiple output tools, and output validators. Affected test suite: 574 passed, 7 skipped, 3 xfailed."}
submitted.

/repo
total 1988
drwxr-xr-x 1 root root    4096 Nov  4 00:37 .
drwxr-xr-x 1 root root    4096 Nov  4 00:38 ..
drwxr-xr-x 1 root root    4096 Nov  4 00:36 .claude
drwxr-xr-x 1 root root    4096 Nov  4 00:35 .gemini
drwxr-xr-x 1 root root    4096 Nov  4 00:38 .git
drwxr-xr-x 1 root root    4096 Nov  4 00:36 .github
-rw-r--r-- 1 root root     544 Nov  4 00:36 .gitignore
-rw-r--r-- 1 root root    2082 Nov  4 00:36 .pre-commit-config.yaml
-rw-r--r-- 1 root root   10131 Nov  4 00:36 AGENTS.md
-rw-r--r-- 1 root root    1044 Nov  4 00:36 BRIEF_BUILTIN_TOOLS_CAP.md

-rw-r--r-- 1 root root    1555 Nov  4 00:36 BRIEF_FOR_RUN_STEP.md
-rw-r--r-- 1 root root    2216 Nov  4 00:36 BRIEF_OTEL_SPANS.md
-rw-r--r-- 1 root root     865 Nov  4 00:36 BRIEF_TEMPORAL_TOOLSET.md
lrwxrwxrwx 1 root root       9 Nov  4 00:35 CLAUDE.md -> AGENTS.md
-rw-r--r-- 1 root root    1100 Nov  4 00:35 LICENSE
-rw-r--r-- 1 ro
…[2721 chars omitted]…
./scripts/scrub_cassette.py
./scripts/upload_test_files.py
./scripts/verify_bedrock_access.py
./scripts/verify_vertex_gcs.py
./scripts/verify_vertex_gcs_all_types.py
./scripts/verify_vertex_gcs_tool_result.py
./tests/AGENTS.md
./tests/cassette_utils.py
./tests/conftest.py
./tests/import_examples.py
./tests/json_body_serializer.py
./tests/mcp_server.py
./tests/parts_from_messages.py
./tests/test_a2a.py

pydantic_ai_slim/LICENSE
pydantic_ai_slim/pydantic_ai/agent/abstract.py
pydantic_ai_slim/pydantic_ai/agent/spec.py
pydantic_ai_slim/pydantic_ai/agent/wrapper.py
pydantic_ai_slim/pydantic_ai/agent/__init__.py
pydantic_ai_slim/pydantic_ai/AGENTS.md
pydantic_ai_slim/pydantic_ai/ag_ui.py
pydantic_ai_slim/pydantic_ai/builtin_tools.py
pydantic_ai_slim/pydantic_ai/capabilities/abstract.py
pydantic_ai_slim/pydantic_ai/capabilities/builtin_tool.py
pydantic_ai_slim/pydantic_ai/capabilities/combined.py
pydantic_ai_slim/pydantic_ai/capabilities/history_processor.py

pydantic_ai_slim/pydantic_ai/capabilities/instructions.py
pydantic_ai_slim/pydantic_ai/capabilities/model_settings.py
pydantic_ai_slim/pydantic_ai/capabilities/prepare_tools.py
pydantic_ai_slim/pydantic_ai/capabilities/thinking.py
pydantic_ai_slim/pydantic_ai/capabilities/toolset.py
pydantic_ai_slim/pydantic_ai/capabilities/web_search.
…[3619 chars omitted]…
oviders/gateway.py
pydantic_ai_slim/pydantic_ai/providers/github.py
pydantic_ai_slim/pydantic_ai/providers/google.py
pydantic_ai_slim/pydantic_ai/providers/google_gla.py
pydantic_ai_slim/pydantic_ai/providers/google_vertex.py
pydantic_ai_slim/pydantic_ai/providers/grok.py
pydantic_ai_slim/pydantic_ai/providers/groq.py
pydantic_ai_slim/pydantic_ai/providers/heroku.py
bash: line 8: rg: command not found

pydantic_ai_slim/pydantic_ai/agent/__init__.py:186:    _prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = dataclasses.field(repr=False)
pydantic_ai_slim/pydantic_ai/agent/__init__.py:218:        prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:250:        prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:280:        prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:325:            prepare_tools: Custom function to prepare the tool definition of all tools for each step, except output tools.
pydantic_ai_slim/pydantic_ai/agent/__init__.py:427:        self._prepare_tools = prepare_tools
pydantic_ai_slim/pydantic_ai/agent/__init__.py:432:            self._output_toolset.max_retries = self._max_result_retries
pydantic_ai_
…[8628 chars omitted]…
 max_retries=tool.max_retries,
pydantic_ai_slim/pydantic_ai/_tool_manager.py:360:            max_retries = tool.max_retries if tool is not None else self.default_max_retries
pydantic_ai_slim/pydantic_ai/_tool_manager.py:361:            self._check_max_retries(name, max_retries, e)

pydantic_ai_slim/pydantic_ai/_tool_manager.py:464:            self._check_max_retries(name, validated.tool.max_retries, e)

tests/AGENTS.md:26:- Test both positive and negative cases for optional capabilities (model features, server features, streaming) — ensures features work when supported AND fail gracefully when absent — Prevents false confidence from tests that only check unsupported cases, catching both implementation bugs and missing error handling
tests/cassettes/test_dbos/test_fastmcp_toolset.yaml:22:        capabilities: {}
tests/cassettes/test_dbos/test_fastmcp_toolset.yaml:32:        data: {"jsonrpc":"2.0","id":0,"result":{"protocolVersion":"2025-03-26","capabilities":{"tools":{"listChanged":true}},"serverInfo":{"name":"DeepWiki","version":"0.0.1"}}}
tests/cassettes/test_dbos/test_fastmcp_toolset.yaml:419:        capabilities: {}
tests/cassettes/test_dbos/test_fastmcp_toolset.yaml:429:        data: {"jsonrpc":"2.0","id":0,"result":{"protocolVersion":"2025-03-26","capabilities":{"tools":{"listCh
…[8630 chars omitted]…
ent = Agent(FunctionModel(simple_model_function), capabilities=[cap])
tests/test_capabilities.py:2286:        agent = Agent(FunctionModel(simple_model_function), capabilities=[cap])
tests/test_capabilities.py:2296:        """Multiple capabilities compose wrap_node_run as nested middleware."""
tests/test_capabilities.py:2311:            capabilities=[OrderedCap(name='outer'), OrderedCap(name='inner')],

<!-- braindump: rules extracted from PR review patterns -->

# Coding Guidelines

## Code Style

<!-- rule:409 -->
- Keep PRs focused on their stated purpose — exclude unrelated changes even if conceptually related — Simplifies review, prevents unintended side effects, and makes rollbacks cleaner when each PR has a single clear objective
<!-- rule:910 -->
- Wrap code identifiers in backticks in user-facing messages (errors, warnings, logs) — Improves readability and clearly distinguishes code elements from prose, making error messages easier to parse and debug
<!-- rule:193 -->
- Centralize validation at one layer — removes redundancy and establishes single source of truth — Prevents validation drift when requirements change and reduces maintenance burden by avoiding duplicate validation logic across the call chain

<!-- rule:2 -->
- Extract duplicated logic into shared helpers after 2
…[8630 chars omitted]…
   ) -> Any:
        """Modify result after execution."""
        return result

    async def wrap_tool_execute(
        self,
        ctx: RunContext[AgentDepsT],
        *,
        call: ToolCallPart,
        args: dict[str, Any],
        handler: Callable[[dict[str, Any]], Awaitable[Any]],
    ) -> Any:
        """Wraps tool execution. handler() runs the tool."""
        return await handler(args)

from __future__ import annotations

from collections.abc import AsyncIterable, Awaitable, Callable, Sequence
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING, Any

from pydantic_ai import _system_prompt
from pydantic_ai._instructions import AgentInstructions, normalize_instructions
from pydantic_ai.messages import AgentStreamEvent, ModelResponse, ToolCallPart
from pydantic_ai.settings import ModelSettings, merge_model_settings
from pydantic_ai.tools import AgentBuiltinTool, AgentDepsT, RunContext, ToolDefinition
from pydantic_ai.toolsets import AbstractToolset, AgentToolset, CombinedToolset

from pydantic_ai.toolsets._dynamic import DynamicToolset

from .abstract import AbstractCapability, ModelRequestContext

if TYPE_CHECKING:
    from pydantic_ai import _agent_graph
    from pydantic_ai.result import FinalResult
    from pydantic_ai.run import AgentRunResul
…[8630 chars omitted]…
nore[reportUnknownArgumentType]
                else:
                    toolsets.append(DynamicToolset(cap_ts))

        return toolsets

    @overload
    def _prepare_output_schema(self, output_type: None) -> _output.OutputSchema[OutputDataT]: ...

    @overload
    def _prepare_output_schema(
        self, output_type: OutputSpec[RunOutputDataT]
    ) -> _output.OutputSchema[RunOutputDataT]: ...

48:from ..capabilities import AbstractCapability, CombinedCapability
49:from ..capabilities.builtin_tool import BuiltinTool as BuiltinToolCap
50:from ..capabilities.history_processor import HistoryProcessor as HistoryProcessorCap
184:    _output_toolset: OutputToolset[AgentDepsT] | None = dataclasses.field(repr=False)
188:    _max_result_retries: int = dataclasses.field(repr=False)
229:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
261:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
291:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
369:            capabilities: Optional list of [capabilities](https://ai.pydantic.dev/capabilities/) to configure the agent with.
370:                Custom capabilities can be created by subclassing
371:                [`AbstractCapability`][pydantic_ai.capabilitie
…[8629 chars omitted]…

ets.extend(self._user_toolsets)
            toolsets.extend(self._dynamic_toolsets)
            for cap_ts in cap_toolsets if cap_toolsets is not None else self._cap_toolsets:
                if isinstance(cap_ts, AbstractToolset):
                    toolsets.append(cap_ts)  # pyright: ignore[reportUnknownArgumentType]
                else:
                    toolsets.append(DynamicToolset(cap_ts))

from __future__ import annotations

from abc import ABC, abstractmethod
from collections.abc import Callable
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Generic, Literal, Protocol

from pydantic_core import SchemaValidator
from typing_extensions import Self

from .._run_context import AgentDepsT, RunContext
from ..tools import ToolDefinition, ToolsPrepareFunc

if TYPE_CHECKING:
    from .approval_required import ApprovalRequiredToolset
    from .filtered import FilteredToolset
    from .prefixed import PrefixedToolset
    from .prepared import PreparedToolset
    from .renamed import RenamedToolset


class SchemaValidatorProt(Protocol):
    """Protocol for a Pydantic Core `SchemaValidator` or `PluggableSchemaValidator` (which is private but API-compatible)."""

    def validate_json(
        self,
        input: str | bytes | bytearray,
        *,
        
…[8630 chars omitted]…
ation_settings.include_content
                        and run_span.is_recording()
                        and final_result is not None
                    ):
                        run_span.set_attribute(
                            'final_result',
                            (
                                final_result.output
                                if isinstance(final_result.output, str)

__init__.py
_dynamic.py
abstract.py
approval_required.py
combined.py
external.py
fastmcp.py
filtered.py
function.py
prefixed.py
prepared.py
renamed.py

wrapper.py
pydantic_ai_slim/pydantic_ai/toolsets/wrapper.py:14:class WrapperToolset(AbstractToolset[AgentDepsT]):
pydantic_ai_slim/pydantic_ai/toolsets/prepared.py:13:class PreparedToolset(WrapperToolset[AgentDepsT]):
pydantic_ai_slim/pydantic_ai/toolsets/prefixed.py:12:class PrefixedToolset(WrapperToolset[AgentDepsT]):
pydantic_ai_slim/pydantic_ai/toolsets/abstract.py:39:class ToolsetTool(Generic[AgentDepsT]):
pydantic_ai_slim/pydantic_ai/toolsets/abstract.py:70:class AbstractToolset(ABC, Generic[AgentDepsT]):
pydantic_ai_slim/pydantic_ai/toolsets/approval_required.py:16:class ApprovalRequiredToolset(WrapperToolset[AgentDepsT]):
pydantic_ai_slim/pydantic_ai/toolsets/combined.py:17:class _CombinedToolsetTool(ToolsetTool[AgentDepsT]):
py
…[5886 chars omitted]…
ll_tool(
        self, name: str, tool_args: dict[str, Any], ctx: RunContext[AgentDepsT], tool: ToolsetTool[AgentDepsT]
    ) -> Any:
        original_name = name.removeprefix(self.prefix + '_')
        ctx = replace(ctx, tool_name=original_name)
        tool = replace(tool, tool_def=replace(tool.tool_def, name=original_name))
        return await super().call_tool(original_name, tool_args, ctx, tool)

from __future__ import annotations as _annotations

import asyncio
import dataclasses
import inspect
import json
import warnings
from asyncio import Lock
from collections.abc import AsyncIterator, Awaitable, Callable, Iterator, Sequence
from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager, contextmanager
from contextvars import ContextVar
from pathlib import Path

from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast, overload

from opentelemetry.trace import NoOpTracer, use_span
from pydantic.json_schema import GenerateJsonSchema
from typing_extensions import Self, TypeVar, deprecated

from pydantic_ai._instrumentation import DEFAULT_INSTRUMENTATION_VERSION, InstrumentationNames
from pydantic_ai._spec import load_from_registry

from .. import (
    _agent_graph,
    _instructions,
    _output,
    _system_prompt,
    _utils,
    concurrency 
…[8630 chars omitted]…
settings,
        )

        user_prompt_node = _agent_graph.UserPromptNode[AgentDepsT](
            user_prompt=user_prompt,
            deferred_tool_results=deferred_tool_results,
            instructions=instructions_literal,
            instructions_functions=instructions_functions,
            system_prompts=self._system_prompts,
            system_prompt_functions=self._system_prompt_functions,

pydantic_ai_slim/pydantic_ai/toolsets/__init__.py:16:AgentToolset = Union[AbstractToolset[AgentDepsT], ToolsetFunc[AgentDepsT]]  # noqa: UP007 — Union needed at runtime (no future annotations)
pydantic_ai_slim/pydantic_ai/toolsets/__init__.py:21:    'AgentToolset',
from __future__ import annotations as _annotations

import dataclasses
from collections.abc import Iterator, Sequence
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import field
from typing import TYPE_CHECKING, Any, Generic

from opentelemetry.trace import NoOpTracer, Tracer

from typing_extensions import TypeVar

from pydantic_ai._instrumentation import DEFAULT_INSTRUMENTATION_VERSION

from . import _utils, messages as _messages

if TYPE_CHECKING:
    from .models import Model
    from .result import RunUsage
    from .settings import ModelSettings

# TODO (v2): Change the default
…[8630 chars omitted]…
etries(ctx.deps.max_result_retries, error=e)
1170:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=validated.validation_error)
1186:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1192:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1225:        ctx.state.increment_retries(ctx.deps.max_result_retries)

@property
    def name(self) -> str | None:
        """The name of the agent, used for logging.

        If `None`, we try to infer the agent name from the call frame when the agent is first run.
        """
        name_ = self._override_name.get()
        return name_.value if name_ else self._name

    @name.setter
    def name(self, value: str | None) -> None:
        """Set the name of the agent, used for logging."""

        self._name = value

    @property
    def description(self) -> str | None:
        """A human-readable description of the agent.

        If the description is a TemplateStr, returns the raw template source.
        The rendered description is available at runtime via OTel span attributes.
        """
        if self._description is None:
            return None
        return str(self._description)

    @description.setter
    def description(self, valu
…[8630 chars omitted]…
sult_tool_call_ids = set(tool_call_results.keys())
        tool_call_ids_to_run = {call.tool_call_id for call in calls_to_run}
        if tool_call_ids_to_run != result_tool_call_ids:
            raise exceptions.UserError(
                'Tool call results need to be provided for all deferred tool calls. '
                f'Expected: {tool_call_ids_to_run}, got: {result_tool_call_ids}'
            )

Callables are called before each model request, allowing dynamic per-step settings.
            usage_limits: Optional limits on model request count or token usage.
            usage: Optional usage to start with, useful for resuming a conversation or agents used in tools.
            metadata: Optional metadata to attach to this run. Accepts a dictionary or a callable taking
                [`RunContext`][pydantic_ai.tools.RunContext]; merged with the agent's configured metadata.
            infer_name: Whether to try to infer the agent name from the call frame if it's not set.
            toolsets: Optional additional toolsets for this run.
            builtin_tools: Optional additional builtin tools for this run.

        Returns:
            The result of the run.
        """

        if infer_name and self.name is None:
            self._infer_name(inspect.currentf
…[8630 chars omitted]…
      args_override: str | dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Validate tool arguments using Pydantic schema and custom args_validator_func.

        Returns:
            The validated arguments as a dictionary.

        Raises:
            ValidationError: If argument validation fails.
            ModelRetry: If argument validation fails with a retry request.
        """

pydantic_ai_slim/pydantic_ai/agent/__init__.py:1106:        toolset = self._get_toolset(
pydantic_ai_slim/pydantic_ai/agent/__init__.py:2093:    def _get_toolset(
pydantic_ai_slim/pydantic_ai/agent/__init__.py:2191:                    toolset = self._get_toolset()
pydantic_ai_slim/pydantic_ai/agent/__init__.py:2221:        self._get_toolset().apply(_set_sampling_model)

return toolsets

    @overload
    def _prepare_output_schema(self, output_type: None) -> _output.OutputSchema[OutputDataT]: ...

    @overload
    def _prepare_output_schema(
        self, output_type: OutputSpec[RunOutputDataT]
    ) -> _output.OutputSchema[RunOutputDataT]: ...

    def _prepare_output_schema(self, output_type: OutputSpec[Any] | None) -> _output.OutputSchema[Any]:
        if output_type is not None:

            if self._output_validators:
                raise exceptions.UserError('Cannot set a custom run `output_type` when the agent has output validators')
            schema = _output.OutputSchema.build(output_type)
        else:
            schema = self._output_schema

        return schema

    async def __aenter__(self) -> Self:
        """Enter the agent context.

        This will start all [`MCPServerStdio`s][pydantic_ai.mcp.MCPServerStdio] register
…[1978 chars omitted]…
 a web chat UI for this agent.

        This method returns a pre-configured Starlette application that provides a web-based
        chat interface for interacting with the agent. By default, the UI is fetched from a
        CDN and cached on first use.

        The returned Starlette application can be mounted into a FastAPI app or run directly
        with any ASGI server (uvicorn, hypercorn, etc.).

pydantic_ai_slim/pydantic_ai/agent/abstract.py:162:    async def run(
pydantic_ai_slim/pydantic_ai/agent/abstract.py:183:    async def run(
pydantic_ai_slim/pydantic_ai/agent/abstract.py:203:    async def run(
pydantic_ai_slim/pydantic_ai/agent/abstract.py:257:            event_stream_handler: Optional handler for events from the model's streaming response and the agent's execution of tools to use for this run.
pydantic_ai_slim/pydantic_ai/agent/abstract.py:289:                    async with node.stream(agent_run.ctx) as stream:
pydantic_ai_slim/pydantic_ai/agent/abstract.py:291:                        wrapped = await agent_run.ctx.deps.root_capability.wrap_run_event_stream(run_ctx, stream=stream)
pydantic_ai_slim/pydantic_ai/agent/abstract.py:292:                        await event_stream_handler(run_ctx, wrapped)
pydantic_ai_slim/pydantic_ai/agent/abstract.py:299:    def run_sync(
p
…[3257 chars omitted]…
vents('What is the capital of France?'):
pydantic_ai_slim/pydantic_ai/agent/abstract.py:944:        return self._run_stream_events(

pydantic_ai_slim/pydantic_ai/agent/abstract.py:960:    async def _run_stream_events(
pydantic_ai_slim/pydantic_ai/agent/abstract.py:987:        async def run_agent() -> AgentRunResult[Any]:
pydantic_ai_slim/pydantic_ai/agent/__init__.py:2299:    async def run_mcp_servers(

"""The output return JSON schema."""
        if output_type is None:
            output_type = self.output_type

        return_types = types_from_output_spec(output_spec=output_type)

        json_schemas: list[JsonSchema] = []
        for return_type in return_types:
            json_schema = TypeAdapter(return_type).json_schema(mode='serialization')
            if json_schema not in json_schemas:
                json_schemas.append(json_schema)

        if len(json_schemas) == 1:
            return json_schemas[0]
        else:
            json_schemas, all_defs = _utils.merge_json_schema_defs(json_schemas)
            json_schema: JsonSchema = {'anyOf': json_schemas}
            if all_defs:
                json_schema['$defs'] = all_defs
            return json_schema

    @overload
    async def run(
        self,
        user_prompt: str | Sequence[_messages.UserContent
…[8630 chars omitted]…
lset or a factory function that creates a toolset from a run context."""

__all__ = (
    'AbstractToolset',
    'AgentToolset',
    'ToolsetFunc',
    'ToolsetTool',
    'CombinedToolset',
    'ExternalToolset',
    'DeferredToolset',
    'FilteredToolset',
    'FunctionToolset',
    'PrefixedToolset',
    'RenamedToolset',
    'PreparedToolset',
    'WrapperToolset',
    'ApprovalRequiredToolset',
)

pydantic_ai_slim/pydantic_ai/agent/__init__.py:694:            capabilities=all_capabilities,
pydantic_ai_slim/pydantic_ai/capabilities/combined.py:34:        return replace(self, capabilities=new_caps)
pydantic_ai_slim/pydantic_ai/capabilities/prepare_tools.py:28:    agent = Agent('openai:gpt-5', capabilities=[PrepareTools(hide_admin_tools)])
48:from ..capabilities import AbstractCapability, CombinedCapability
49:from ..capabilities.builtin_tool import BuiltinTool as BuiltinToolCap
50:from ..capabilities.history_processor import HistoryProcessor as HistoryProcessorCap
87:from .spec import AgentSpec, get_capability_registry
229:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
261:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
291:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
369:            cap
…[8629 chars omitted]…

entDepsT, ...]]]
        ] = ContextVar('_override_tools', default=None)
        self._override_instructions: ContextVar[
            _utils.Option[list[str | _system_prompt.SystemPromptFunc[AgentDepsT]]]
        ] = ContextVar('_override_instructions', default=None)
        self._override_metadata: ContextVar[_utils.Option[AgentMetadata[AgentDepsT]]] = ContextVar(
1427:    def override(  # noqa: C901

pydantic_ai_slim/pydantic_ai/agent/__init__.py:48:from ..capabilities import AbstractCapability, CombinedCapability
pydantic_ai_slim/pydantic_ai/agent/__init__.py:49:from ..capabilities.builtin_tool import BuiltinTool as BuiltinToolCap
pydantic_ai_slim/pydantic_ai/agent/__init__.py:50:from ..capabilities.history_processor import HistoryProcessor as HistoryProcessorCap
pydantic_ai_slim/pydantic_ai/agent/__init__.py:229:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:261:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:291:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:369:            capabilities: Optional list of [capabilities](https://ai.pydantic.dev/capabi
…[8630 chars omitted]…
ire agent run. handler() executes the run."""
        return await handler()

    async def wrap_node_run(

        self,
        ctx: RunContext[AgentDepsT],
        *,
        node: _agent_graph.AgentNode[AgentDepsT, Any],
        handler: Callable[
            [_agent_graph.AgentNode[AgentDepsT, Any]],
            Awaitable[_agent_graph.AgentNode[AgentDepsT, Any] | End[FinalResult[Any]]],
        ],

tests/test_agent.py:371:    """Test that ctx.retry and ctx.max_retries are correctly tracked in RunContext for output validators."""
tests/test_agent.py:382:    agent = Agent(FunctionModel(return_model), output_type=Foo, output_retries=target_retries)
tests/test_agent.py:386:        retries_log.append(ctx.retry)
tests/test_agent.py:387:        max_retries_log.append(ctx.max_retries)
tests/test_agent.py:389:        if ctx.retry == target_retries:
tests/test_agent.py:392:            raise ModelRetry(f'Retry {ctx.retry}')
tests/test_agent.py:403:    """Test that ctx.retry and ctx.max_retries are correctly tracked in RunContext for output functions."""
tests/test_agent.py:409:        retries_log.append(ctx.retry)
tests/test_agent.py:410:        max_retries_log.append(ctx.max_retries)
tests/test_agent.py:411:        if ctx.retry == target_retries:
tests/test_agent.py:414:            raise 
…[4197 chars omitted]…

repare_retries.append(ctx.retry)
tests/test_tools.py:1304:        prepare_max_retries.append(ctx.max_retries)
tests/test_tools.py:1311:        call_retries.append(ctx.retry)
tests/test_tools.py:1312:        call_max_retries.append(ctx.max_retries)
tests/test_tools.py:3544:        retry_values.append(ctx.retry)
tests/test_tools.py:3573:        validator_calls.append((ctx.retry, ctx.tool_call_approved))

usage=RequestUsage(input_tokens=51, output_tokens=7),
                model_name='function:return_model:',
                timestamp=IsNow(tz=timezone.utc),
                run_id=IsStr(),
            ),
            ModelRequest(
                parts=[
                    RetryPromptPart(
                        content='"a" should be 42',
                        tool_name='final_result',
                        tool_call_id=IsStr(),
                        timestamp=IsNow(tz=timezone.utc),

                    )
                ],
                timestamp=IsNow(tz=timezone.utc),
                run_id=IsStr(),
            ),
            ModelResponse(
                parts=[ToolCallPart(tool_name='final_result', args='{"a": 42, "b": "foo"}', tool_call_id=IsStr())],
                usage=RequestUsage(input_tokens=63, output_tokens=14),
                model_name='func
…[8629 chars omitted]…
etries(ctx.deps.max_result_retries, error=e)
1170:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=validated.validation_error)
1186:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1192:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1225:        ctx.state.increment_retries(ctx.deps.max_result_retries)

def __init__(self, *, allows_deferred_tools: bool):
        super().__init__(allows_deferred_tools=allows_deferred_tools, allows_image=True)

    @property
    def mode(self) -> OutputMode:
        return 'image'


@dataclass(init=False)
class StructuredTextOutputSchema(OutputSchema[OutputDataT], ABC):
    processor: BaseObjectOutputProcessor[OutputDataT]
    template: str | Literal[False] | None

    def __init__(
        self,
        *,
        template: str | Literal[False] | None = None,
        processor: BaseObjectOutputProcessor[OutputDataT],
        allows_deferred_tools: bool,
        allows_image: bool,
    ):
        super().__init__(
            text_processor=processor,
            object_def=processor.object_def,
            allows_deferred_tools=allows_deferred_tools,
            allows_image=allows_image,
        )
        self.processor = processor
        self.t
…[8630 chars omitted]…
t] = {}
    if tool_call_results is not None:
        # Deferred tool calls are "run" as well, by reading their value from the tool call results
        calls_to_run.extend(tool_calls_by_kind['external'])
        calls_to_run.extend(tool_calls_by_kind['unapproved'])

        result_tool_call_ids = set(tool_call_results.keys())
        tool_call_ids_to_run = {call.tool_call_id for call in calls_to_run}

160:class OutputValidator(Generic[AgentDepsT, OutputDataT_inv]):
213:class OutputSchema(ABC, Generic[OutputDataT]):
512:    async def process(
530:class ObjectOutputProcessor(BaseObjectOutputProcessor[OutputDataT]):
600:    async def process(
768:    async def process(
804:    async def process(
837:    async def process(
859:class OutputToolset(AbstractToolset[AgentDepsT]):
    @property
    def mode(self) -> OutputMode:
        raise NotImplementedError()

    @property
    def allows_text(self) -> bool:
        return self.text_processor is not None

    @classmethod
    def build(  # noqa: C901
        cls,
        output_spec: OutputSpec[OutputDataT],
        *,
        name: str | None = None,
        description: str | None = None,
        strict: bool | None = None,
    ) -> OutputSchema[OutputDataT]:
        """Build an OutputSchema dataclass from an output type."""
        o
…[8629 chars omitted]…
    return output


@overload
def _flatten_output_spec(
    output_spec: OutputTypeOrFunction[T] | Sequence[OutputTypeOrFunction[T]],
) -> Sequence[OutputTypeOrFunction[T]]: ...


@overload
def _flatten_output_spec(output_spec: OutputSpec[T]) -> Sequence[_OutputSpecItem[T]]: ...


def _flatten_output_spec(output_spec: OutputSpec[T]) -> Sequence[_OutputSpecItem[T]]:
    outputs: Sequence[OutputSpec[T]]

'HistoryProcessor',
)


T = TypeVar('T')
S = TypeVar('S')
NoneType = type(None)
EndStrategy = Literal['early', 'exhaustive']
DepsT = TypeVar('DepsT')
OutputT = TypeVar('OutputT')

@dataclasses.dataclass(kw_only=True)
class GraphAgentState:
    """State kept across the execution of the agent graph."""

    message_history: list[_messages.ModelMessage] = dataclasses.field(default_factory=list[_messages.ModelMessage])
    usage: _usage.RunUsage = dataclasses.field(default_factory=_usage.RunUsage)
    retries: int = 0
    run_step: int = 0
    run_id: str = dataclasses.field(default_factory=lambda: str(uuid.uuid4()))
    metadata: dict[str, Any] | None = None
    last_max_tokens: int | None = None
    """Last-resolved `max_tokens` from model settings, used only in error messages."""

    def increment_retries(
        self,
        max_result_retries: int,
        error: BaseExceptio
…[8629 chars omitted]…
return self.__class__(
            toolset=toolset,
            root_capability=self.root_capability,
            ctx=ctx,
            tools=await toolset.get_tools(ctx),
            default_max_retries=self.default_max_retries,
        )

    @property
    def tool_defs(self) -> list[ToolDefinition]:
        """The tool definitions for the tools in this tool manager."""
        if self.tools is None:

from __future__ import annotations

import asyncio
from collections.abc import Callable, Sequence
from contextlib import AsyncExitStack
from dataclasses import dataclass, field, replace
from typing import Any

from typing_extensions import Self

from .._run_context import AgentDepsT, RunContext
from ..exceptions import UserError

from .abstract import AbstractToolset, ToolsetTool


@dataclass(kw_only=True)
class _CombinedToolsetTool(ToolsetTool[AgentDepsT]):
    """A tool definition for a combined toolset tools that keeps track of the source toolset and tool."""

    source_toolset: AbstractToolset[AgentDepsT]
    source_tool: ToolsetTool[AgentDepsT]


@dataclass
class CombinedToolset(AbstractToolset[AgentDepsT]):
    """A toolset that combines multiple toolsets.

    See [toolset docs](../toolsets.md#combining-toolsets) for more information.
    """

    toolsets: Sequence[AbstractToo
…[8629 chars omitted]…
l_defs(self) -> list[ToolDefinition]:
        """The tool definitions for the tools in this tool manager."""
        if self.tools is None:
            raise ValueError('ToolManager has not been prepared for a run step yet')  # pragma: no cover

        return [tool.tool_def for tool in self.tools.values()]

    def get_parallel_execution_mode(self, calls: list[ToolCallPart]) -> ParallelExecutionMode:

pydantic_ai_slim/pydantic_ai/_agent_graph.py:653:        ctx.deps.tool_manager = await ctx.deps.tool_manager.for_run_step(run_context)
pydantic_ai_slim/pydantic_ai/_agent_graph.py:936:        ctx.deps.tool_manager = await ctx.deps.tool_manager.for_run_step(run_context)
pydantic_ai_slim/pydantic_ai/agent/__init__.py:2192:                    await exit_stack.enter_async_context(toolset)

__repr__ = dataclasses_no_defaults_repr


@dataclasses.dataclass
class SetFinalResult(AgentNode[DepsT, NodeRunEndT]):
    """A node that immediately ends the graph run after a streaming response produced a final result."""

    final_result: result.FinalResult[NodeRunEndT]

    async def run(
        self, ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, NodeRunEndT]]
    ) -> End[result.FinalResult[NodeRunEndT]]:

        return End(self.final_result)


def build_run_context(ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, Any]]) -> RunContext[DepsT]:
    """Build a `RunContext` object from the current agent graph run context."""
    run_context = RunContext[DepsT](
        deps=ctx.deps.user_deps,
        model=ctx.deps.model,
        usage=ctx.state.usage,
        prompt=ctx.deps.prompt,
        messages=ctx.state.message_history,
        validation_context
…[8629 chars omitted]…
            output_parts.append(
1318:                    if call in tool_calls_by_kind['external']:
1328:                        output_parts.append(e.tool_retry)
1353:    output_parts: list[_messages.ModelRequestPart],
1448:    output_parts.extend([tool_parts_by_index[k] for k in sorted(tool_parts_by_index)])
1449:    output_parts.extend([user_parts_by_index[k] for k in sorted(user_parts_by_index)])

if _messages.BinaryImage in flattened_outputs:
                raise UserError(  # pragma: no cover
                    '`NativeOutput` cannot contain `BinaryImage`. Include it alongside the native output marker instead: `output_type=[NativeOutput(...), BinaryImage]`'
                )

            return NativeOutputSchema(
                template=output.template,
                processor=cls._build_processor(
                    flattened_outputs,
                    name=output.name,
                    description=output.description,
                    strict=output.strict,

                ),
                allows_deferred_tools=allows_deferred_tools,
                allows_image=allows_image,
            )
        elif output := next((output for output in outputs if isinstance(output, PromptedOutput)), None):  # pyright: ignore[reportUnknownVariableType,reportUnkn
…[8630 chars omitted]…
o, session state) — Verifies actual data flow and integration behavior rather than just testing mock interfaces, catching real-world issues that mocks would miss

## General

<!-- rule:463 -->
- Remove stale test docstrings, comments, and historical provider bug notes when behavior changes — Outdated test documentation misleads developers about what's actually being tested and why

<!-- /braindump -->

'final_result': {'type': 'object'},
                    },
                }
            ),
        }

    @contextmanager
    def override(  # noqa: C901
        self,
        *,
        name: str | _utils.Unset = _utils.UNSET,
        deps: AgentDepsT | _utils.Unset = _utils.UNSET,

        model: models.Model | models.KnownModelName | str | _utils.Unset = _utils.UNSET,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset = _utils.UNSET,
        tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] | _utils.Unset = _utils.UNSET,
        instructions: AgentInstructions[AgentDepsT] | _utils.Unset = _utils.UNSET,
        metadata: AgentMetadata[AgentDepsT] | _utils.Unset = _utils.UNSET,
        model_settings: AgentModelSettings[AgentDepsT] | _utils.Unset = _utils.UNSET,
    ) -> Iterator[None]:
        """Context manager to tempor
…[8684 chars omitted]…
ts/test_capabilities.py:988:        async def for_run(self, ctx: RunContext[None]) -> AbstractCapability[None]:
tests/test_capabilities.py:1019:        async def for_run(self, ctx: RunContext[None]) -> AbstractCapability[None]:
tests/test_capabilities.py:1696:    async def test_wrap_run_fires_for_run_stream(self):
tests/test_capabilities.py:1707:    async def test_after_run_fires_for_run_stream(self):

span.set_attribute(
                instrumentation_names.tool_result_attr,
                output if isinstance(output, str) else json.dumps(InstrumentedModel.serialize_any(output)),
            )

        return output


@dataclass
class OutputValidator(Generic[AgentDepsT, OutputDataT_inv]):
    function: OutputValidatorFunc[AgentDepsT, OutputDataT_inv]
    _takes_ctx: bool = field(init=False)

    _is_async: bool = field(init=False)

    def __post_init__(self):
        self._takes_ctx = len(inspect.signature(self.function).parameters) > 1
        self._is_async = _utils.is_async_callable(self.function)

    async def validate(
        self,
        result: T,
        run_context: RunContext[AgentDepsT],
        wrap_validation_errors: bool = True,
    ) -> T:
        """Validate a result but calling the function.

        Args:
            result: The result data after
…[8629 chars omitted]…
ted.validation_error)
pydantic_ai_slim/pydantic_ai/_agent_graph.py:1186:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
pydantic_ai_slim/pydantic_ai/_agent_graph.py:1192:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
pydantic_ai_slim/pydantic_ai/_agent_graph.py:1225:        ctx.state.increment_retries(ctx.deps.max_result_retries)

pydantic_ai_slim/pydantic_ai/_agent_graph.py:398:    all_tool_defs = list(ctx.deps.tool_manager.tool_defs)
pydantic_ai_slim/pydantic_ai/_agent_graph.py:402:    all_tool_defs = await ctx.deps.root_capability.prepare_tools(run_context, all_tool_defs)
pydantic_ai_slim/pydantic_ai/_agent_graph.py:406:    for tool_def in all_tool_defs:
pydantic_ai_slim/pydantic_ai/_agent_graph.py:653:        ctx.deps.tool_manager = await ctx.deps.tool_manager.for_run_step(run_context)
pydantic_ai_slim/pydantic_ai/_agent_graph.py:892:                        # skips to CallToolsNode, bypassing for_run_step); in that case we
pydantic_ai_slim/pydantic_ai/_agent_graph.py:936:        ctx.deps.tool_manager = await ctx.deps.tool_manager.for_run_step(run_context)
pydantic_ai_slim/pydantic_ai/_tool_manager.py:100:    async def for_run_step(self, ctx: RunContext[AgentDepsT]) -> ToolManager[AgentDepsT]:
pydantic_ai_slim/pydantic_ai/_tool_manager.py:112:        toolset = await self.toolset.for_run_step(ctx)
pydantic_ai_slim/pydantic_ai/_tool_manager.py:123:    def tool_defs(self) -> list[ToolDefinition]:

async def _sys_parts(self, run_context: RunContext[DepsT]) -> list[_messages.ModelRequestPart]:
        """Build the initial messages for the conversation."""
        messages: list[_messages.ModelRequestPart] = [_messages.SystemPromptPart(p) for p in self.system_prompts]
        for sys_prompt_runner in self.system_prompt_functions:
            prompt = await sys_prompt_runner.run(run_context)
            if sys_prompt_runner.dynamic:
                # To enable dynamic system prompt refs in future runs, use a placeholder string
                messages.append(
                    _messages.SystemPromptPart(prompt or '', dynamic_ref=sys_prompt_runner.function.__qualname__)
                )
            elif prompt:
                # omit empty system prompts

                messages.append(_messages.SystemPromptPart(prompt))
        return messages

    __repr__ = dataclasses_no_
…[8629 chars omitted]…
ed_args is not None

        cap = self.root_capability
        call = validated.call
        ctx = validated.ctx

        async def do_execute(args: dict[str, Any]) -> Any:
            # Execute with potentially modified args
            modified_validated = replace(validated, validated_args=args)
            return await self._raw_execute(modified_validated, usage=usage)

        if cap is not None:

from __future__ import annotations as _annotations

import asyncio
import inspect
from abc import ABC, abstractmethod
from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable, Iterator, Mapping, Sequence
from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager
from types import FrameType
from typing import TYPE_CHECKING, Any, Generic, TypeAlias, cast, overload

import anyio
from pydantic import TypeAdapter

from typing_extensions import Self, TypeIs, TypeVar, deprecated

from pydantic_graph import End

from .. import (
    _agent_graph,
    _instructions,
    _tool_manager,
    _utils,
    exceptions,
    messages as _messages,
    models,
    result,
    usage as _usage,
)
from .._json_schema import JsonSchema
from .._output import types_from_output_spec
from .._template import TemplateStr
from .._tool_manager import ToolManager
from
…[8629 chars omitted]…
pydantic_ai/agent/__init__.py:2136:        """Build the list of toolsets, optionally with per-run capability toolsets."""
pydantic_ai_slim/pydantic_ai/run.py:188:        `wrap_node_run`. Use `next()` for capability-hooked iteration, or use
pydantic_ai_slim/pydantic_ai/run.py:304:        return await self.ctx.deps.root_capability.wrap_node_run(run_context, node=node, handler=_step_handler)
333f09c base

997:            retry=ctx.state.retries,
998:            max_retries=ctx.deps.max_result_retries,
pydantic_ai_slim/pydantic_ai/_agent_graph.py:997:            retry=ctx.state.retries,
85:    def increment_retries(
844:                        ctx.state.increment_retries(ctx.deps.max_result_retries)
916:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1158:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1170:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=validated.validation_error)
1186:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1192:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1225:        ctx.state.increment_retries(ctx.deps.max_result_retries)

tests/models/test_google.py:4292:    by using the OutputToolset approach rather than treating nested models as separate tools.
tests/models/test_google.py:4314:    # This should work WITHOUT NativeOutput - the agent should use OutputToolset
pydantic_ai_slim/pydantic_ai/agent/__init__.py:44:from .._output import OutputToolset
pydantic_ai_slim/pydantic_ai/agent/__init__.py:184:    _output_toolset: OutputToolset[AgentDepsT] | None = dataclasses.field(repr=False)
pydantic_ai_slim/pydantic_ai/_output.py:215:    toolset: OutputToolset[Any] | None = None
pydantic_ai_slim/pydantic_ai/_output.py:321:        toolset = OutputToolset.build(tool_outputs + other_outputs, name=name, description=description, strict=strict)
pydantic_ai_slim/pydantic_ai/_output.py:388:        toolset: OutputToolset[Any] | None,
pydantic_ai_slim/pydantic_ai/_output.py:493:        toolset: OutputToolset[Any] | None,
pydantic_ai_slim/pydantic_ai/_output.py:859:class OutputToolset(AbstractToolset[AgentDepsT]):
75:async def execute_traced_output_function(
654:    async def call(
664:            output = await execute_traced_output_function(

847:        data = await execute_traced_output_function(self._function_schema, run_context, args, wrap_validation_errors)

else:
                raise

        output = await self.call(output, run_context, wrap_validation_errors)

        return output

    def validate(
        self,
        data: str | dict[str, Any] | None,
        *,
        allow_partial: bool = False,

        validation_context: Any | None = None,
    ) -> dict[str, Any]:
        pyd_allow_partial: Literal['off', 'trailing-strings'] = 'trailing-strings' if allow_partial else 'off'
        if isinstance(data, str):
            return self.validator.validate_json(
                data or '{}', allow_partial=pyd_allow_partial, context=validation_context
            )
        else:
            return self.validator.validate_python(
                data or {}, allow_partial=pyd_allow_partial, context=validation_context
            )

    async def call(
        self,
        output: dict[str, Any],
        run_context: RunCon
…[8629 chars omitted]…
not None else self.timeout
544-        if timeout is not None:
545-            try:
546-                with anyio.fail_after(timeout):
547-                    return await tool.call_func(tool_args, ctx)
548-            except TimeoutError:
549-                raise ModelRetry(f'Timed out after {timeout} seconds.') from None
550-        else:
551-            return await tool.call_func(tool_args, ctx)

tool_call_id=IsStr(),
                        timestamp=IsNow(tz=timezone.utc),
                    )
                ],
                timestamp=IsNow(tz=timezone.utc),
                run_id=IsStr(),
            ),
            ModelResponse(
                parts=[ToolCallPart(tool_name='foobar', args='{}', tool_call_id=IsStr())],
                usage=RequestUsage(input_tokens=65, output_tokens=4),
                model_name='function:empty:',
                timestamp=IsNow(tz=timezone.utc),

                run_id=IsStr(),
            ),
            ModelRequest(
                parts=[
                    RetryPromptPart(
                        tool_name='foobar',
                        content="Unknown tool name: 'foobar'. No tools available.",
                        tool_call_id=IsStr(),
                        timestamp=IsNow(tz=timezone.utc),
     
…[8629 chars omitted]…
of 1"):
3227:        UnexpectedModelBehavior,
3233:        UnexpectedModelBehavior,
3248:        UnexpectedModelBehavior,
3254:        UnexpectedModelBehavior,
4426:        """Test that exhaustive strategy raises `UnexpectedModelBehavior` when all outputs have validation errors."""
4449:        with pytest.raises(UnexpectedModelBehavior, match='Exceeded maximum retries \\(1\\) for output validation'):

ERROR: UnexpectedModelBehavior Exceeded maximum retries (2) for output validation
SEEN: [(0, 2), (1, 2), (2, 2)]

213:class AgentInfo:
214-    """Information about an agent.
215-
216-    This is passed as the second to functions used within [`FunctionModel`][pydantic_ai.models.function.FunctionModel].
217-    """
218-
219-    function_tools: list[ToolDefinition]
220-    """The function tools available on this agent.
221-
222-    These are the tools registered via the [`tool`][pydantic_ai.agent.Agent.tool] and
223-    [`tool_plain`][pydantic_ai.agent.Agent.tool_plain] decorators.
224-    """

225-    allow_text_output: bool
226-    """Whether a plain text output is allowed."""
227-    output_tools: list[ToolDefinition]
228-    """The tools that can called to produce the final output of the run."""
229-    model_settings: ModelSettings | None
230-    """The model settings passed to the run call."""
231-    model_request_parameters: ModelRequestParameters
232-    """The model request parameters passe
…[6884 chars omitted]…
    async def test_wrap_run_fires_for_iter(self):
        cap = LoggingCapability()
        agent = Agent(FunctionModel(simple_model_function), capabilities=[cap])
        async with agent.iter('hello') as agent_run:
            async for _node in agent_run:
                pass
        assert 'wrap_run:before' in cap.log
        assert 'wrap_run:after' in cap.log
        assert 'after_run' in cap.log

run_id=IsStr(),
            ),
        ]
    )


def test_empty_response():
    def llm(messages: list[ModelMessage], _info: AgentInfo) -> ModelResponse:
        if len(messages) == 1:
            return ModelResponse(parts=[])
        else:
            return ModelResponse(parts=[TextPart('ok here is text')])

    agent = Agent(FunctionModel(llm))

    result = agent.run_sync('Hello')

    assert result.all_messages() == snapshot(
        [
            ModelRequest(
                parts=[
                    UserPromptPart(
                        content='Hello',
                        timestamp=IsDatetime(),
                    )
                ],
                timestamp=IsNow(tz=timezone.utc),
                run_id=IsStr(),
            ),
            ModelResponse(
                parts=[],
                usage=RequestUsage(input_tokens=51),
                
…[8629 chars omitted]…
= 'valid'

    # NOTE: When changing tests in this class:
    # 1. Follow the existing order
    # 2. Update tests in `tests/test_streaming.py::TestMultipleToolCallsStreaming` as well


async def test_model_settings_override() -> None:
    def return_settings(_: list[ModelMessage], info: AgentInfo) -> ModelResponse:
        return ModelResponse(parts=[TextPart(to_json(info.model_settings).decode())])

188:    _max_result_retries: int = dataclasses.field(repr=False)
189:    _max_tool_retries: int = dataclasses.field(repr=False)
213:        retries: int = 1,
215:        output_retries: int | None = None,
245:        retries: int = 1,
247:        output_retries: int | None = None,
275:        retries: int = 1,
277:        output_retries: int | None = None,
317:            retries: The default number of retries to allow for tool calls and output validation, before raising an error.
320:            output_retries: The maximum number of retries to allow for output validation, defaults to `retries`.
417:        self._max_result_retries = output_retries if output_retries is not None else retries
432:            self._output_toolset.max_retries = self._max_result_retries

495:        retries: int | None = None,
497:        output_retries: int | None = None,
529:        retries: int | None = 
…[8629 chars omitted]…
 a directory
grep: docs/models: Is a directory
grep: docs/ui: Is a directory
grep: pydantic_ai_slim/pydantic_ai: Is a directory
docs/capabilities.md:746:| `output_retries` | `int \| None` | Output validation retries |
docs/changelog.md:54:See [#2441](https://github.com/pydantic/pydantic-ai/pull/2441) - The `result_retries` argument was also removed from the `Agent` class. Use `output_retries` instead.

run_span = tracer.start_span(
            instrumentation_names.get_agent_run_span_name(agent_name),
            attributes=span_attributes,
        )

        run_metadata: dict[str, Any] | None = None
        try:
            async with (
                _concurrency.get_concurrency_context(self._concurrency_limiter, f'agent:{agent_name}'),
                graph.iter(
                    inputs=user_prompt_node,
                    state=state,

                    deps=graph_deps,
                    span=use_span(run_span) if run_span.is_recording() else None,
                    infer_name=False,
                ) as graph_run,
            ):
                async with toolset:
                    agent_run = AgentRun(graph_run)
                    run_metadata = self._resolve_and_store_metadata(agent_run.ctx, metadata)

                    # Build RunContext for run life
…[8629 chars omitted]…
ts=deferred_tool_results,
                model=model,
                deps=deps,
                model_settings=model_settings,
                usage_limits=usage_limits,
                usage=usage,
46:from .._tool_manager import ParallelExecutionMode, ToolManager
1112:        tool_manager = ToolManager[AgentDepsT](
1148:            tool_manager=tool_manager,
1198:                async with toolset:

),
            ModelResponse(
                parts=[TextPart(content='ok here is text')],
                usage=RequestUsage(input_tokens=51, output_tokens=4),
                model_name='function:llm:',
                timestamp=IsDatetime(),
                run_id=IsStr(),
            ),
        ]
    )

def test_empty_response_without_recovery():
    def llm(messages: list[ModelMessage], _info: AgentInfo) -> ModelResponse:
        return ModelResponse(parts=[])

    agent = Agent(FunctionModel(llm), output_type=tuple[str, int])

    with capture_run_messages() as messages:
        with pytest.raises(UnexpectedModelBehavior, match=r'Exceeded maximum retries \(1\) for output validation'):
            agent.run_sync('Hello')

    assert messages == snapshot(
        [
            ModelRequest(
                parts=[
                    UserPromptPart(
                  
…[8629 chars omitted]…
81:    with pytest.raises(UnexpectedModelBehavior, match=r'Exceeded maximum retries \(1\) for output validation'):
tests/test_streaming.py:773:        with pytest.raises(UnexpectedModelBehavior, match=r'Exceeded maximum retries \(0\) for output validation'):
tests/test_streaming.py:2215:        with pytest.raises(UnexpectedModelBehavior, match='Exceeded maximum retries \\(1\\) for output validation'):

run_context = build_run_context(ctx)
        assert self.last_request_context is not None, 'last_request_context must be set before _finish_handling'
        request_context = self.last_request_context
        run_context.model_settings = request_context.model_settings
        response = await ctx.deps.root_capability.after_model_request(
            run_context, request_context=request_context, response=response
        )

        # Update usage
        ctx.state.usage.incr(response.usage)
        if ctx.deps.usage_limits:  # pragma: no branch
            ctx.deps.usage_limits.check_tokens(ctx.state.usage)

        # Append the model response to state.message_history
        ctx.state.message_history.append(response)

        # Set the `_result` attribute since we can't use `return` in an async iterator
        self._result = CallToolsNode(response)

        return self._res
…[8629 chars omitted]…
llPart],
    ) -> AsyncIterator[_messages.HandleResponseEvent]:
        run_context = build_run_context(ctx)

        # This will raise errors for any tool name conflicts
        ctx.deps.tool_manager = await ctx.deps.tool_manager.for_run_step(run_context)

        output_parts: list[_messages.ModelRequestPart] = []
        output_final_result: deque[result.FinalResult[NodeRunEndT]] = deque(maxlen=1)

from __future__ import annotations as _annotations

from collections.abc import AsyncIterator, Iterator, Sequence
from contextlib import AbstractAsyncContextManager, asynccontextmanager, contextmanager
from typing import Any, overload

from .. import (
    _instructions,
    _utils,
    messages as _messages,
    models,
    usage as _usage,

)
from .._json_schema import JsonSchema
from .._template import TemplateStr
from ..output import OutputDataT, OutputSpec
from ..run import AgentRun
from ..tools import (
    AgentBuiltinTool,
    AgentDepsT,
    DeferredToolResults,
    Tool,
    ToolFuncEither,
)
from ..toolsets import AbstractToolset
from .abstract import AbstractAgent, AgentMetadata, AgentModelSettings, EventStreamHandler, RunOutputDataT


class WrapperAgent(AbstractAgent[AgentDepsT, OutputDataT]):
    """Agent which wraps another agent.

    Does nothing on its own, used as a 
…[8630 chars omitted]…
nother AsyncBaseTransport and adds retry capabilities using the tenacity library.
pydantic_ai_slim/pydantic_ai/_agent_graph.py:24:from pydantic_ai.capabilities.abstract import AbstractCapability
pydantic_ai_slim/pydantic_ai/_agent_graph.py:400:    # Let capabilities filter/modify tool definitions
pydantic_ai_slim/pydantic_ai/_tool_manager.py:25:    from .capabilities.abstract import AbstractCapability

if isinstance(part, _messages.TextPart):
                        text += part.content
                    elif isinstance(part, _messages.ToolCallPart):
                        tool_calls.append(part)
                    elif isinstance(part, _messages.FilePart):
                        files.append(part.content)
                    elif isinstance(part, _messages.BuiltinToolCallPart):
                        # Text parts before a built-in tool call are essentially thoughts,
                        # not part of the final result output, so we reset the accumulated text
                        text = ''
                        yield _messages.BuiltinToolCallEvent(part)  # pyright: ignore[reportDeprecated]
                    elif isinstance(part, _messages.BuiltinToolReturnPart):

                        yield _messages.BuiltinToolResultEvent(part)  # pyright: ignore
…[8629 chars omitted]…
 well, by reading their value from the tool call results
        calls_to_run.extend(tool_calls_by_kind['external'])
        calls_to_run.extend(tool_calls_by_kind['unapproved'])

        result_tool_call_ids = set(tool_call_results.keys())
        tool_call_ids_to_run = {call.tool_call_id for call in calls_to_run}
./pydantic_ai_slim/pydantic_ai/_agent_graph.py:997:            retry=ctx.state.retries,

async def run_stream(  # noqa: C901
        self,
        user_prompt: str | Sequence[_messages.UserContent] | None = None,
        *,
        output_type: OutputSpec[RunOutputDataT] | None = None,
        message_history: Sequence[_messages.ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        model: models.Model | models.KnownModelName | str | None = None,
        instructions: _instructions.AgentInstructions[AgentDepsT] = None,
        deps: AgentDepsT = None,
        model_settings: AgentModelSettings[AgentDepsT] | None = None,
        usage_limits: _usage.UsageLimits | None = None,

        usage: _usage.RunUsage | None = None,
        metadata: AgentMetadata[AgentDepsT] | None = None,
        infer_name: bool = True,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        builtin_tools: Sequence[AgentBuil
…[8629 chars omitted]…
      )

        task = asyncio.create_task(run_agent())

        try:
            async with receive_stream:
                async for message in receive_stream:
                    yield message

            result = await task

        except asyncio.CancelledError as e:
            task.cancel(msg=e.args[0] if len(e.args) != 0 else None)
            raise

        yield AgentRunResultEvent(result)

async def model_handler(req_ctx: ModelRequestContext) -> _messages.ModelResponse:
            with set_current_run_context(run_context):
                return await ctx.deps.model.request(
                    req_ctx.messages, req_ctx.model_settings, req_ctx.model_request_parameters
                )

        model_response = await ctx.deps.root_capability.wrap_model_request(
            run_context,
            request_context=ModelRequestContext(
                messages=message_history,
                model_settings=model_settings,
                model_request_parameters=model_request_parameters,

            ),
            handler=model_handler,
        )
        ctx.state.usage.requests += 1

        return await self._finish_handling(ctx, model_response)

    async def _prepare_request(
        self, ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, NodeRunE
…[8628 chars omitted]…
.add(name)

            if not wrap_validation_errors:
                raise

            validation_error = self._wrap_error_as_retry(name, call, e)

            return ValidatedToolCall(
                call=call,
                tool=tool,
                ctx=ctx,
                args_valid=False,
                validated_args=None,
                validation_error=validation_error,
            )

==== 2810
                parts=[],
                usage=RequestUsage(input_tokens=51),
                model_name='function:llm:',
                timestamp=IsDatetime(),
                run_id=IsStr(),
            ),
            ModelRequest(
                parts=[],
                timestamp=IsNow(tz=timezone.utc),
                run_id=IsStr(),
            ),

            ModelResponse(
                parts=[TextPart(content='ok here is text')],
                usage=RequestUsage(input_tokens=51, output_tokens=4),
                model_name='function:llm:',
                timestamp=IsDatetime(),
                run_id=IsStr(),
            ),
        ]
    )


def test_empty_response_without_recovery():
    def llm(messages: list[ModelMessage], _info: AgentInfo) -> ModelResponse:
        return ModelResponse(parts=[])

    agent = Agent(FunctionModel(llm), output_type=tuple[str
…[8629 chars omitted]…
  content=[
                                ErrorDetails(
                                    type='missing',
                                    loc=('value',),
                                    msg='Field required',
                                    input={'bad_value': 'first'},
                                )
                            ],
                            tool_name='final_result',

# Capabilities

A capability is a reusable, composable unit of agent behavior. Instead of threading multiple arguments through your `Agent` constructor — instructions here, model settings there, a toolset somewhere else, a history processor on yet another parameter — you can bundle related behavior into a single capability and pass it via the [`capabilities`][pydantic_ai.agent.Agent.__init__] parameter.

Capabilities can provide any combination of:

* **Instructions** — static or dynamic system prompt additions
* **Model settings** — static or per-step configuration
* **Tools** — via toolsets or builtin tools
* **Lifecycle hooks** — intercept and modify model requests, tool calls, and the overall run

This makes them the primary extension point for Pydantic AI. Whether you're building a memory system, a guardrail, a cost tracker, or an approval workflow, a capability is the right abs

…[8630 chars omitted]…
NMENT=.venv$(subst .,,$(PYTEST_PYTHON))) uv run $(if $(PYTEST_PYTHON),--python $(PYTEST_PYTHON)) coverage run -m pytest -n auto --dist=loadgroup --durations=20
78:	uv run -m pytest --update-examples tests/test_examples.py
82:	uv run -m pytest --record-mode=rewrite tests
103:all: format lint typecheck testcov ## Run code formatting, linting, static type checks, and tests with coverage report generation

diff --git a/pydantic_ai_slim/pydantic_ai/agent/__init__.py b/pydantic_ai_slim/pydantic_ai/agent/__init__.py
index da580bb..b9f7ab2 100644
--- a/pydantic_ai_slim/pydantic_ai/agent/__init__.py
+++ b/pydantic_ai_slim/pydantic_ai/agent/__init__.py
@@ -1107,6 +1107,7 @@ class Agent(AbstractAgent[AgentDepsT, OutputDataT]):
             output_toolset=output_toolset,
             additional_toolsets=toolsets,
             cap_toolsets=cap_toolsets,
+            root_capability=run_capability,
         )
         toolset = await toolset.for_run(initial_ctx)
         tool_manager = ToolManager[AgentDepsT](

@@ -2095,6 +2096,7 @@ class Agent(AbstractAgent[AgentDepsT, OutputDataT]):
         output_toolset: AbstractToolset[AgentDepsT] | None | _utils.Unset = _utils.UNSET,
         additional_toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
         cap_toolsets: Sequence[AgentTools
…[4275 chars omitted]…
ities:
+            wrapper = capability.get_wrapper_toolset(toolset)
+            if wrapper is not None:
+                toolset = wrapper
+                wrapped_toolset = wrapper
+        return wrapped_toolset
+
     def get_builtin_tools(self) -> Sequence[AgentBuiltinTool[AgentDepsT]]:
         builtin_tools: list[AgentBuiltinTool[AgentDepsT]] = []
         for capability in self.capabilities:

844:                        ctx.state.increment_retries(ctx.deps.max_result_retries)
916:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1184:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1190:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1223:        ctx.state.increment_retries(ctx.deps.max_result_retries)
    tool_calls_by_kind: dict[ToolKind | Literal['unknown'], list[_messages.ToolCallPart]] = defaultdict(list)
    for call in tool_calls:
        tool_def = tool_manager.get_tool_def(call.tool_name)
        if tool_def:
            kind = tool_def.kind
        else:
            kind = 'unknown'

        tool_calls_by_kind[kind].append(call)

    # First, we handle output tool calls
    for call in tool_calls_by_kind['output']:
        # `final_result` can be passed into 
…[8628 chars omitted]…
   async def call_tool(
        self, name: str, tool_args: dict[str, Any], ctx: RunContext[AgentDepsT], tool: ToolsetTool[AgentDepsT]
    ) -> Any:
        output = await self.processors[name].call(tool_args, ctx, wrap_validation_errors=False)
        for validator in self.output_validators:
            output = await validator.validate(output, ctx, wrap_validation_errors=False)
        return output

args_valid: bool
    """Whether argument validation (schema + custom validator) passed."""
    validated_args: dict[str, Any] | None = None
    """The validated arguments if validation passed, None otherwise."""
    validation_error: ToolRetryError | None = None
    """The validation error if validation failed, None otherwise."""


@dataclass
class ToolManager(Generic[AgentDepsT]):
    """Manages tools for an agent run step. It caches the agent run's toolset's tool definitions and handles calling tools and retries."""

    toolset: AbstractToolset[AgentDepsT]
    """The toolset that provides the tools for this run step."""
    root_capability: AbstractCapability[AgentDepsT] | None = None
    """The root capability for hook invocation."""
    ctx: RunContext[AgentDepsT] | None = None
    """The agent run context for a specific run step."""
    tools: dict[str, ToolsetTool[AgentDeps
…[5795 chars omitted]…
Any:
        """Execute a validated function tool call within a trace span.

        See <https://opentelemetry.io/docs/specs/semconv/gen-ai/gen-ai-spans/#execute-tool-span>.
        """
70:    failed_tools: set[str] = field(default_factory=set[str])
108:                for failed_tool_name in self.failed_tools
365:                self.failed_tools.add(name)
465:            self.failed_tools.add(name)

844:                        ctx.state.increment_retries(ctx.deps.max_result_retries)
916:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1184:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1190:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1223:        ctx.state.increment_retries(ctx.deps.max_result_retries)
 pydantic_ai_slim/pydantic_ai/_agent_graph.py          |  2 --
 pydantic_ai_slim/pydantic_ai/_output.py               | 13 +++++++++++--
 pydantic_ai_slim/pydantic_ai/agent/__init__.py        | 16 ++++++++++++----
 pydantic_ai_slim/pydantic_ai/capabilities/abstract.py | 12 +++++++++++-
 pydantic_ai_slim/pydantic_ai/capabilities/combined.py |  9 +++++++++
 5 files changed, 43 insertions(+), 9 deletions(-)

yield event
        # Early strategy is chosen and final result is not yet set
        # Or exhaustive strategy is chosen
        else:
            # Validate and execute the output tool call — unlike deferred tools,
            # output tools track retries and can be skipped if a final_result exists.
            try:
                validated = await tool_manager.validate_tool_call(call)
            except exceptions.UnexpectedModelBehavior as e:
                # If we already have a valid final result, don't fail the entire run
                # This allows exhaustive strategy to complete successfully when at least one output tool is valid
                if final_result:

                    # If output tool fails when we already have a final result, skip it without retrying
                    for event in _emit_skipped_output_tool(
                        call, 'O
…[1913 chars omitted]…
eld _messages.FunctionToolCallEvent(call, args_valid=True)
                output_parts.append(e.tool_retry)
                yield _messages.FunctionToolResultEvent(e.tool_retry)
                continue

            part = _messages.ToolReturnPart(
                tool_name=call.tool_name,
                content='Final result processed.',
                tool_call_id=call.tool_call_id,
            )

)
                yielded = True

            first_node = agent_run.next_node  # start with the first node
            assert isinstance(first_node, _agent_graph.UserPromptNode)  # the first node should be a user prompt node
            node: _agent_graph.AgentNode[Any, Any] = first_node
            while not yielded:
                graph_ctx = agent_run.ctx
                if self.is_model_request_node(node):
                    async with node.stream(graph_ctx) as stream:
                        final_result_event = None

                        async def stream_to_final(
                            stream: AgentStream,
                        ) -> AsyncIterator[_messages.ModelResponseStreamEvent]:
                            nonlocal final_result_event
                            async for event in stream:
                                yield event
              
…[6581 chars omitted]…
ly(self, visitor: Callable[[AbstractToolset[AgentDepsT]], None]) -> None:
        """Run a visitor function on all "leaf" toolsets (i.e. those that implement their own tool listing and calling)."""
        visitor(self)

1106:        toolset = self._get_toolset(
2094:    def _get_toolset(
2199:                    toolset = self._get_toolset()
2229:        self._get_toolset().apply(_set_sampling_model)

Traceback (most recent call last):
  File "/repo/pydantic_ai_slim/pydantic_ai/_tool_manager.py", line 457, in _raw_execute
    tool_result = await self.toolset.call_tool(
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/toolsets/combined.py", line 93, in call_tool
    return await tool.source_toolset.call_tool(name, tool_args, ctx, tool.source_tool)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/_output.py", line 982, in call_tool
    output = await self.processors[name].call(tool_args, ctx, wrap_validation_errors=False)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/_output.py", line 664, in call
    output = await execute_traced_output_function(

             ^^^^^^^^^^^^^^^
…[5982 chars omitted]…
ydantic_ai_slim/pydantic_ai/_agent_graph.py", line 1190, in process_tool_calls
    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
  File "/repo/pydantic_ai_slim/pydantic_ai/_agent_graph.py", line 109, in increment_retries
    raise exceptions.UnexpectedModelBehavior(message) from error
pydantic_ai.exceptions.UnexpectedModelBehavior: Exceeded maximum retries (2) for output validation

844:                        ctx.state.increment_retries(ctx.deps.max_result_retries)
916:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1184:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1190:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1223:        ctx.state.increment_retries(ctx.deps.max_result_retries)
                # This allows exhaustive strategy to complete successfully when at least one output tool is valid
                if final_result:
                    # If output tool fails when we already have a final result, skip it without retrying
                    for event in _emit_skipped_output_tool(
                        call, 'Output tool not used - output failed validation.', output_parts, args_valid=False
                    ):
                        yi

…[3080 chars omitted]…
ValidationError or ModelRetry to a ToolRetryError with a RetryPromptPart."""
        if isinstance(error, ValidationError):
            content: list[Any] | str = error.errors(include_url=False, include_context=False)
        else:
            content = error.message
        m = _messages.RetryPromptPart(tool_name=name, content=content, tool_call_id=call.tool_call_id)
        return ToolRetryError(m)

Traceback (most recent call last):
  File "<stdin>", line 30, in <module>
AssertionError
844:                        ctx.state.increment_retries(ctx.deps.max_result_retries)
916:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1184:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1190:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1223:        ctx.state.increment_retries(ctx.deps.max_result_retries)
                # This allows exhaustive strategy to complete successfully when at least one output tool is valid
                if final_result:
                    # If output tool fails when we already have a final result, skip it without retrying
                    for event in _emit_skipped_output_tool(

                        call, 'Output tool not used - output failed valid
…[1872 chars omitted]…
eld _messages.FunctionToolCallEvent(call, args_valid=True)
                output_parts.append(e.tool_retry)
                yield _messages.FunctionToolResultEvent(e.tool_retry)
                continue

            part = _messages.ToolReturnPart(
                tool_name=call.tool_name,
                content='Final result processed.',
                tool_call_id=call.tool_call_id,
            )

3006:def test_unknown_tool():
3052:def test_unknown_tool_fix():
3100:def test_unknown_tool_multiple_retries():
3167:def test_unknown_tool_per_tool_retries_exceeded():

    result = agent.run_sync('Hello')
    assert result.output == 'success'
    assert result.all_messages() == snapshot(
        [
            ModelRequest(
                parts=[UserPromptPart(content='Hello', timestamp=IsNow(tz=timezone.utc))],
                timestamp=IsNow(tz=timezone.utc),

                run_id=IsStr(),
            ),
            ModelResponse(
                parts=[ToolCallPart(tool_name='foobar', args='{}', tool_call_id=IsStr())],
                usage=RequestUsage(input_tokens=51, output_tokens=2),
                model_name='function:empty:',
                timestamp=IsNow(tz=timezone.utc),
                run_id=IsStr(),
            ),
            ModelRequest(
                parts=[
   
…[8629 chars omitted]…
, 0) == max_retries:
+            if self.tools is not None and (tool := self.tools.get(name)) is not None and tool.tool_def.kind == 'output':
+                raise UnexpectedModelBehavior(f'Exceeded maximum retries ({max_retries}) for output validation') from error
             raise UnexpectedModelBehavior(f'Tool {name!r} exceeded max retries count of {max_retries}') from error
 
     @staticmethod

844:                        ctx.state.increment_retries(ctx.deps.max_result_retries)
916:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1184:                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1190:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1223:        ctx.state.increment_retries(ctx.deps.max_result_retries)
Traceback (most recent call last):
  File "/repo/pydantic_ai_slim/pydantic_ai/_tool_manager.py", line 459, in _raw_execute
    tool_result = await self.toolset.call_tool(
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/toolsets/combined.py", line 93, in call_tool
    return await tool.source_toolset.call_tool(name, tool_args, ctx, tool.source_tool)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

…[6405 chars omitted]…
ydantic_ai_slim/pydantic_ai/_agent_graph.py", line 1190, in process_tool_calls
    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
  File "/repo/pydantic_ai_slim/pydantic_ai/_agent_graph.py", line 109, in increment_retries
    raise exceptions.UnexpectedModelBehavior(message) from error
pydantic_ai.exceptions.UnexpectedModelBehavior: Exceeded maximum retries (2) for output validation

Removing:                 ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
Removing:                     ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
                yield event
        # Early strategy is chosen and final result is not yet set
        # Or exhaustive strategy is chosen
        else:
            # Validate and execute the output tool call — unlike deferred tools,
            # output tools track retries and can be skipped if a final_result exists.
            try:
                validated = await tool_manager.validate_tool_call(call)
            except exceptions.UnexpectedModelBehavior as e:
                # If we already have a valid final result, don't fail the entire run

                # This allows exhaustive strategy to complete successfully when at least one output tool is valid
                if final_result:
         
…[2221 chars omitted]…
l_name,
                content='Final result processed.',
                tool_call_id=call.tool_call_id,
            )
            output_parts.append(part)

844:                        ctx.state.increment_retries(ctx.deps.max_result_retries)
916:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1221:        ctx.state.increment_retries(ctx.deps.max_result_retries)

1175 '            try:'
1176 '                result_data = await tool_manager.execute_tool_call(validated)'
1177 '            except exceptions.UnexpectedModelBehavior as e:'
1178 '                if final_result:'
1179 '                    for event in _emit_skipped_output_tool('
1180 "                        call, 'Output tool not used - output function execution failed.', output_parts, args_valid=True"
1181 '                    ):'
1182 '                        yield event'
1183 '                    continue'
1184 '                raise  # pragma: lax no cover'
1185 '            except ToolRetryError as e:'
1186 "                # If we already have a valid final result, don't increment retries for invalid output tools"

1187 '                # This allows the run to succeed if at least one output tool returned a valid result'
1188 '                if not final_result:'
1189 '                    yield _messages.FunctionToolCallEvent(call, args_valid=True)'
1190 '                output_parts.append(e.tool_retry)'
1191 '                yield _messages.FunctionToolResultEvent(e.tool_retry)'
1192 '                continue'
1193 ''
1194 '            part = _messages.ToolReturnPart('
1195 '                tool_name=call.tool_name,'

844:                        ctx.state.increment_retries(ctx.deps.max_result_retries)
916:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1221:        ctx.state.increment_retries(ctx.deps.max_result_retries)
            except exceptions.UnexpectedModelBehavior as e:
                # If we already have a valid final result, don't fail the entire run
                # This allows exhaustive strategy to complete successfully when at least one output tool is valid
                if final_result:
                    # If output tool fails when we already have a final result, skip it without retrying
                    for event in _emit_skipped_output_tool(
                        call, 'Output tool not used - output failed validation.', output_parts, args_valid=False
                    ):
                        yield event

                    con
…[3621 chars omitted]…
antic_ai_slim/pydantic_ai/_output.py               | 13 +++++++++++--
 pydantic_ai_slim/pydantic_ai/_tool_manager.py         |  2 ++
 pydantic_ai_slim/pydantic_ai/agent/__init__.py        | 16 ++++++++++++----
 pydantic_ai_slim/pydantic_ai/capabilities/abstract.py | 12 +++++++++++-
 pydantic_ai_slim/pydantic_ai/capabilities/combined.py |  9 +++++++++
 6 files changed, 46 insertions(+), 12 deletions(-)

OUTPUT RETRIES: Paris [(0, 5), (1, 5), (2, 5), (3, 5), (4, 5), (5, 5)]
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']

TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_regular_tool'] OUTPUTS: ['final_result']
TOOLS: ['b_a_reg
…[5360 chars omitted]…
_slim/pydantic_ai/_agent_graph.py", line 710, in _prepare_request
    ctx.deps.usage_limits.check_before_request(usage)
  File "/repo/pydantic_ai_slim/pydantic_ai/usage.py", line 378, in check_before_request
    raise UsageLimitExceeded(f'The next request would exceed the request_limit of {request_limit}')
pydantic_ai.exceptions.UsageLimitExceeded: The next request would exceed the request_limit of 50

19-    'AgentDepsT',
20-    'ArgsValidatorFunc',
21-    'DocstringFormat',
22-    'RunContext',
23-    'SystemPromptFunc',
24-    'ToolFuncContext',
25-    'ToolFuncPlain',
26-    'ToolFuncEither',
27-    'ToolParams',
28-    'ToolPrepareFunc',
29:    'ToolsPrepareFunc',
30-    'AgentBuiltinTool',

31-    'BuiltinToolFunc',
32-    'Tool',
33-    'ObjectJsonSchema',
34-    'ToolDefinition',
35-    'DeferredToolRequests',
36-    'DeferredToolResults',
37-    'ToolApproved',
38-    'ToolDenied',
39-)
40-
41-
42-ToolParams = ParamSpec('ToolParams', default=...)
43-"""Retrieval function param spec."""
44-
45-SystemPromptFunc: TypeAlias = (
46-    Callable[[RunContext[AgentDepsT]], str | None]
47-    | Callable[[RunContext[AgentDepsT]], Awaitable[str | None]]
48-    | Callable[[], str | None]
49-    | Callable[[], Awaitable[str | None]]
50-)
51-"""A function that may or maybe not take `RunCo
…[3267 chars omitted]…
put_type` and will be used as the output of the agent run if the model called any deferred tools.
172-
173-    Results can be passed to the next agent run using a [`DeferredToolResults`][pydantic_ai.tools.DeferredToolResults] object with the same tool call IDs.
174-
pydantic_ai_slim/pydantic_ai/tools.py:468:    async def prepare_tool_def(self, ctx: RunContext[ToolAgentDepsT]) -> ToolDefinition | None:

......................x..xx..................                            [100%]
42 passed, 32 deselected, 3 xfailed in 3.01s

11:* **Lifecycle hooks** — intercept and modify model requests, tool calls, and the overall run
15:## Using built-in capabilities
49:## Building a custom capability
51:To build your own capability, subclass [`AbstractCapability`][pydantic_ai.capabilities.AbstractCapability] and override the methods you need. There are two categories: **configuration methods** that are called once at agent construction, and **lifecycle hooks** that fire during each run.
53:### Providing configuration
129:### Dynamic model settings
163:## Lifecycle hooks
165:Capabilities can hook into four lifecycle points, each with up to three variants:
171:### Short-circuit exceptions
173:Three exceptions allow hooks to short-circuit the normal flow with a replacement value:
181:### Run hooks
191:### Node hooks

197:The [`wrap_node_run`][pydantic_ai.capabilities.AbstractCapability.wrap_node_run] hook fires for every n
…[8630 chars omitted]…
pabilities.md:650:Override [`from_spec`][pydantic_ai.capabilities.AbstractCapability.from_spec] when the constructor takes types that can't be represented in YAML/JSON. The spec fields should mirror the dataclass fields, but with serializable types:
docs/capabilities.md:658:from pydantic_ai.capabilities import AbstractCapability
docs/capabilities.md:663:class ConditionalTools(AbstractCapability[Any]):

diff --git a/pydantic_ai_slim/pydantic_ai/_agent_graph.py b/pydantic_ai_slim/pydantic_ai/_agent_graph.py
index ae9801c..5e559e9 100644
--- a/pydantic_ai_slim/pydantic_ai/_agent_graph.py
+++ b/pydantic_ai_slim/pydantic_ai/_agent_graph.py
@@ -1155,7 +1155,6 @@ async def process_tool_calls(  # noqa: C901
                     ):
                         yield event
                     continue
-                ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
                 raise  # pragma: lax no cover
 
             if not validated.args_valid:

@@ -1167,7 +1166,6 @@ async def process_tool_calls(  # noqa: C901
                         yield event
                     continue
 
-                ctx.state.increment_retries(ctx.deps.max_result_retries, error=validated.validation_error)
                 yield _messages.FunctionToolCallEvent(call, args_valid=False)
       
…[8629 chars omitted]…
ities:
+            wrapper = capability.get_wrapper_toolset(toolset)
+            if wrapper is not None:
+                toolset = wrapper
+                wrapped_toolset = wrapper
+        return wrapped_toolset
+
     def get_builtin_tools(self) -> Sequence[AgentBuiltinTool[AgentDepsT]]:
         builtin_tools: list[AgentBuiltinTool[AgentDepsT]] = []
         for capability in self.capabilities:

## Building a custom capability

To build your own capability, subclass [`AbstractCapability`][pydantic_ai.capabilities.AbstractCapability] and override the methods you need. There are two categories: **configuration methods** that are called once at agent construction, and **lifecycle hooks** that fire during each run.

### Providing configuration

The simplest capabilities just provide static configuration. Here's a `KnowsCurrentTime` capability that injects the current time into the system prompt:

```python {title="custom_capability_config.py"}
from dataclasses import dataclass
from datetime import datetime
from typing import Any

from pydantic_ai import Agent
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.models.test import TestModel


@dataclass
class KnowsCurrentTime(AbstractCapability[Any]):
    """Tells the agent what time it is."""

    def get_inst
…[8629 chars omitted]…
tion: TemplateStr[Any] | str | None = None,
        model_settings: ModelSettings | None = None,
        retries: int | None = None,
        validation_context: Any = None,
        output_retries: int | None = None,
        tools: Sequence[Tool[Any] | ToolFuncEither[Any, ...]] = (),
        builtin_tools: Sequence[AgentBuiltinTool[Any]] = (),
        prepare_tools: ToolsPrepareFunc[Any] | None = None,

43:class FunctionModel(Model):
44-    """A model controlled by a local function.
45-
46-    Apart from `__init__`, all methods are private or match those of the base class.
47-    """
48-
49-    function: FunctionDef | None
50-    stream_function: StreamFunctionDef | None
51-
52-    _model_name: str = field(repr=False)
53-    _system: str = field(default='function', repr=False)
54-

55-    @overload
56-    def __init__(
57-        self,
58-        function: FunctionDef,
59-        *,
60-        model_name: str | None = None,
61-        profile: ModelProfileSpec | None = None,
62-        settings: ModelSettings | None = None,
63-    ) -> None: ...
64-
65-    @overload
66-    def __init__(
67-        self,
68-        *,
69-        stream_function: StreamFunctionDef,
70-        model_name: str | None = None,
71-        profile: ModelProfileSpec | None = None,
72-        settings: ModelSet
…[8628 chars omitted]…
r
835:                            except ToolRetryError:  # pragma: no cover
914:                    raise ToolRetryError(m)
915:                except ToolRetryError as e:
1185:            except ToolRetryError as e:
1320:                    except ToolRetryError as e:
1496:            raise ToolRetryError(m)
1500:            raise ToolRetryError(tool_call_result)
1503:    except ToolRetryError as e:

async def _handle_tool_calls(
        self,
        ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, NodeRunEndT]],
        tool_calls: list[_messages.ToolCallPart],
    ) -> AsyncIterator[_messages.HandleResponseEvent]:
        run_context = build_run_context(ctx)

        # This will raise errors for any tool name conflicts
        ctx.deps.tool_manager = await ctx.deps.tool_manager.for_run_step(run_context)

        output_parts: list[_messages.ModelRequestPart] = []
        output_final_result: deque[result.FinalResult[NodeRunEndT]] = deque(maxlen=1)

        async for event in process_tool_calls(
            tool_manager=ctx.deps.tool_manager,
            tool_calls=tool_calls,
            tool_call_results=self.tool_call_results,
            tool_call_metadata=self.tool_call_metadata,
            final_result=None,
            ctx=ctx,
            output_parts=out
…[6762 chars omitted]…
lve this, add `DeferredToolRequests` to the list of output types for this agent.'
            )
        deferred_tool_requests = _output.DeferredToolRequests(
            calls=deferred_calls['external'],
            approvals=deferred_calls['unapproved'],
            metadata=deferred_metadata,
        )

        final_result = result.FinalResult(cast(NodeRunEndT, deferred_tool_requests), None, None)

<!-- braindump: rules extracted from PR review patterns -->

# docs/ Guidelines

## Documentation

<!-- rule:232 -->
- Link all concepts, features, and API elements to their docs/reference pages using anchor fragments (`#section-name`) for specific sections — Improves discoverability and reduces user friction by providing direct navigation to relevant documentation context
<!-- rule:66 -->
- Use reference-style links for API elements: `[ElementName][module.path.ElementName]` — enables hover docs and navigation in mkdocs — Provides interactive documentation features like tooltips and jump-to-definition that plain backticks cannot support
<!-- rule:714 -->
- Omit deprecated features from user-facing docs — document only current approaches — Prevents users from learning outdated patterns and reduces confusion about the recommended way forward

<!-- rule:82 -->
- Write project name as `Pyd
…[8629 chars omitted]…
def.name].validator,
975-            )
976-            for tool_def in self._tool_defs
977-        }
 M pydantic_ai_slim/pydantic_ai/_agent_graph.py
 M pydantic_ai_slim/pydantic_ai/_output.py
 M pydantic_ai_slim/pydantic_ai/_tool_manager.py
 M pydantic_ai_slim/pydantic_ai/agent/__init__.py
 M pydantic_ai_slim/pydantic_ai/capabilities/abstract.py
 M pydantic_ai_slim/pydantic_ai/capabilities/combined.py

242:If desired, this marker class can be used alongside one or more [`ToolOutput`](#tool-output) marker classes (or unmarked types or functions) in a list provided to `output_type`.
322:If you'd like to change the name of the output tool, pass a custom description to aid the model, or turn on or off strict mode, you can wrap the type(s) in the [`ToolOutput`][pydantic_ai.output.ToolOutput] marker class and provide the appropriate arguments. Note that by default, the description is taken from the docstring specified on a Pydantic model or output function, so specifying it using the marker class is typically not necessary.
329:from pydantic_ai import Agent, ToolOutput
345:        ToolOutput(Fruit, name='return_fruit'),
346:        ToolOutput(Vehicle, name='return_vehicle'),
354:1. If we were passing just `Fruit` and `Vehicle` without custom tool names, we could have used a union: `output
…[8087 chars omitted]…
rview.md)), capabilities can observe or transform the event stream:

| Hook | Signature | Purpose |
|---|---|---|
| [`wrap_run_event_stream`][pydantic_ai.capabilities.AbstractCapability.wrap_run_event_stream] | `(ctx, *, stream: AsyncIterable[AgentStreamEvent]) -> AsyncIterable[AgentStreamEvent]` | Observe, filter, or transform streamed events |

```python {title="event_stream_example.py" test="skip"}

async def main():
    async with agent.run_stream('Create a record with name "test" and value 42') as result:
        async for output in result.stream_output(debounce_by=None):
            print(output)
            #> name='test' value=None
            #> name='test' value=42
```

_(This example is complete, it can be run "as is" — you'll need to add `asyncio.run(main())` to run `main`)_

### Output modes

Pydantic AI implements three different methods to get a model to output structured data:

1. [Tool Output](#tool-output), where tool calls are used to produce the output.
2. [Native Output](#native-output), where the model is required to produce text content compliant with a provided JSON schema.
3. [Prompted Output](#prompted-output), where a prompt is injected into the model instructions including the desired JSON schema, and we attempt to parse the model's plain-text response as
…[8629 chars omitted]…
etries: int | None
116-    """The maximum number of retries for the tool."""
117-    strict: bool | None
118-    """Whether to use strict mode for the tool."""
119-
--
124-        name: str | None = None,
125-        description: str | None = None,
126:        max_retries: int | None = None,
127-        strict: bool | None = None,
128-    ):
129-        self.output = type_
130-        self.name = name

..............                                                           [100%]
14 passed, 202 deselected in 2.51s

SYNC RESULT: finished
STREAM RESULT: finished

part = _messages.ToolReturnPart(
                tool_name=call.tool_name,
                content='Final result processed.',
                tool_call_id=call.tool_call_id,
            )
            output_parts.append(part)

            # In both `early` and `exhaustive` modes, use the first output tool's result as the final result
            if not final_result:
                final_result = result.FinalResult(result_data, call.tool_name, call.tool_call_id)

    # Then, we handle function tool calls

    calls_to_run: list[_messages.ToolCallPart] = []
    if final_result and ctx.deps.end_strategy == 'early':
        for call in tool_calls_by_kind['function']:
            output_parts.append(
                _messages.ToolReturnPart(
                    tool_name=call.tool_name,
                    content='Tool not executed - a final result was already processed.',
  
…[8628 chars omitted]…
 output_validators
1024-
1025-        # Build the graph
--
1139-            model=model_used,
1140-            get_model_settings=get_model_settings,
1141-            usage_limits=usage_limits,
1142:            max_result_retries=self._max_result_retries,
1143-            end_strategy=self.end_strategy,
1144-            output_schema=output_schema,
1145-            output_validators=output_validators,

call, 'Output tool not used - a final result was already processed.', output_parts, args_valid=None
            ):
                yield event
        # Early strategy is chosen and final result is not yet set
        # Or exhaustive strategy is chosen
        else:
            # Validate and execute the output tool call — unlike deferred tools,
            # output tools track retries and can be skipped if a final_result exists.
            try:
                validated = await tool_manager.validate_tool_call(call)
            except exceptions.UnexpectedModelBehavior:
                # If we already have a valid final result, don't fail the entire run

                # This allows exhaustive strategy to complete successfully when at least one output tool is valid
                if final_result:
                    # If output tool fails when we already have a final
…[1620 chars omitted]…
eld _messages.FunctionToolCallEvent(call, args_valid=True)
                output_parts.append(e.tool_retry)
                yield _messages.FunctionToolResultEvent(e.tool_retry)
                continue

            part = _messages.ToolReturnPart(
                tool_name=call.tool_name,
                content='Final result processed.',
                tool_call_id=call.tool_call_id,
            )

/usr/local/bin/pytest
.DEFAULT_GOAL := all

.PHONY: .uv
.uv: ## Check that uv is installed
	@uv --version || echo 'Please install uv: https://docs.astral.sh/uv/getting-started/installation/'

.PHONY: .pre-commit
.pre-commit: ## Check that pre-commit is installed
	@pre-commit -V || echo 'Please install pre-commit: https://pre-commit.com/'

.PHONY: install

install: .uv .pre-commit ## Install the package, dependencies, and pre-commit for local development
	uv sync --frozen --all-extras --no-extra outlines-vllm-offline --all-packages --group lint --group docs
	pre-commit install --install-hooks

.PHONY: install-all-python
install-all-python: ## Install and synchronize an interpreter for every python version
	UV_PROJECT_ENVIRONMENT=.venv310 uv sync --python 3.10 --frozen --all-extras --no-extra outlines-vllm-offline --all-packages --group lint --group docs
	UV_PROJECT_ENVIRONMENT=.venv311 
…[3611 chars omitted]…
 without fixing all warnings
.PHONY: docs-serve
docs-serve: ## Build and serve the documentation
	uv run mkdocs serve --no-strict

.PHONY: cf-pages-build
cf-pages-build: ## Install uv, install dependencies and build the docs, used on CloudFlare Pages
	curl -LsSf https://astral.sh/uv/install.sh | sh
	uv python install 3.12
	uv sync --python 3.12 --frozen --group docs
	uv pip freeze
	uv run mkdocs build

diff --git a/docs/capabilities.md b/docs/capabilities.md
index d80d762..2636e61 100644
--- a/docs/capabilities.md
+++ b/docs/capabilities.md
@@ -126,6 +126,41 @@ The full set of configuration methods:
 | [`get_toolset()`][pydantic_ai.capabilities.AbstractCapability.get_toolset] | [`AgentToolset`][pydantic_ai.toolsets.AgentToolset] ` \| None` | A [toolset](toolsets.md) to register with the agent |
 | [`get_builtin_tools()`][pydantic_ai.capabilities.AbstractCapability.get_builtin_tools] | `Sequence[AbstractBuiltinTool]` | [Builtin tools](builtin-tools.md) to register |
 
+### Wrapping function toolsets
+
+To decorate the tools assembled for a run, override [`get_wrapper_toolset()`][pydantic_ai.capabilities.AbstractCapability.get_wrapper_toolset]. It receives the function toolset after agent-level `prepare_tools` has been applied; output tools are kept separate and are never passed throu
…[8629 chars omitted]…

ities:
+            wrapper = capability.get_wrapper_toolset(toolset)
+            if wrapper is not None:
+                toolset = wrapper
+                wrapped_toolset = wrapper
+        return wrapped_toolset
+
     def get_builtin_tools(self) -> Sequence[AgentBuiltinTool[AgentDepsT]]:
         builtin_tools: list[AgentBuiltinTool[AgentDepsT]] = []
         for capability in self.capabilities:

```

The full set of configuration methods:

| Method | Return type | Purpose |
|---|---|---|
| [`get_instructions()`][pydantic_ai.capabilities.AbstractCapability.get_instructions] | [`AgentInstructions`][pydantic_ai._instructions.AgentInstructions] ` \| None` | System prompt additions (static strings, [template strings](#template-strings), or callables) |
| [`get_model_settings()`][pydantic_ai.capabilities.AbstractCapability.get_model_settings] | [`AgentModelSettings`][pydantic_ai.agent.abstract.AgentModelSettings] ` \| None` | Model settings dict, or a callable for [per-step settings](#dynamic-model-settings) |
| [`get_toolset()`][pydantic_ai.capabilities.AbstractCapability.get_toolset] | [`AgentToolset`][pydantic_ai.toolsets.AgentToolset] ` \| None` | A [toolset](toolsets.md) to register with the agent |
| [`get_builtin_tools()`][pydantic_ai.capabilities.AbstractCapability.get_buil
…[7530 chars omitted]…
f.tools is not None and (tool := self.tools.get(name)) is not None and tool.tool_def.kind == 'output':

+                raise UnexpectedModelBehavior(
+                    f'Exceeded maximum retries ({max_retries}) for output validation'
+                ) from error
             raise UnexpectedModelBehavior(f'Tool {name!r} exceeded max retries count of {max_retries}') from error
 
     @staticmethod

<stdin>:4: SyntaxWarning: invalid escape sequence '\|'
<stdin>:7: SyntaxWarning: invalid escape sequence '\|'

| [`get_instructions()`][pydantic_ai.capabilities.AbstractCapability.get_instructions] | [`AgentInstructions`][pydantic_ai._instructions.AgentInstructions] ` \| None` | System prompt additions (static strings, [template strings](#template-strings), or callables) |
| [`get_model_settings()`][pydantic_ai.capabilities.AbstractCapability.get_model_settings] | [`AgentModelSettings`][pydantic_ai.agent.abstract.AgentModelSettings] ` \| None` | Model settings dict, or a callable for [per-step settings](#dynamic-model-settings) |
| [`get_toolset()`][pydantic_ai.capabilities.AbstractCapability.get_toolset] | [`AgentToolset`][pydantic_ai.toolsets.AgentToolset] ` \| None` | A [toolset](toolsets.md) to register with the agent |
| [`get_wrapper_toolset()`][pydantic_ai.capabilities.AbstractCapability.get_wrapper_toolset] | [`AbstractToolset`][pydantic_ai.toolsets.AbstractToolset] ` \| None` | Wrap t
…[393 chars omitted]…
toolset]. It receives the function toolset after agent-level `prepare_tools` has been applied; output tools are kept separate and are never passed through this hook. When multiple capabilities return wrappers, they are applied in capability order, so the first capability is the innermost wrapper.

```python {title="prefix_toolset_capability.py"}
from dataclasses import dataclass
from typing import Any

pydantic_ai_slim/pydantic_ai/_agent_graph.py 103 293                         f'Model token limit ({self.last_max_tokens or "provider default"}) exceeded while generating a tool call, resulting in incomplete arguments. Increase the `max_tokens` model setting, or simplify the prompt to result in a shorter response that will fit within the limit.'
pydantic_ai_slim/pydantic_ai/_agent_graph.py 248 122                         # If there's no new prompt or instructions, skip ModelRequestNode and go directly to CallToolsNode
pydantic_ai_slim/pydantic_ai/_agent_graph.py 346 132         """Reevaluate any `SystemPromptPart` with dynamic_ref in the provided messages by running the associated runner function."""
pydantic_ai_slim/pydantic_ai/_agent_graph.py 689 139         # `ctx.state.message_history` is the same list used by `capture_run_messages`, so we should replace its contents, not the refer
…[8629 chars omitted]…
ling_model()`.'
pydantic_ai_slim/pydantic_ai/agent/__init__.py 2313 151         If you need to set a sampling model on all MCP servers, use [`agent.set_mcp_sampling_model()`][pydantic_ai.agent.Agent.set_mcp_sampling_model].
pydantic_ai_slim/pydantic_ai/agent/__init__.py 2354 124                 'To use tools that require approval, add `DeferredToolRequests` to the list of output types for this agent.'

pydantic_ai_slim/pydantic_ai/_agent_graph.py:90:        self.retries += 1
pydantic_ai_slim/pydantic_ai/_agent_graph.py:91:        if self.retries > max_result_retries:

ruff not installed
pyright not installed
mypy not installed
black not installed

........................................................................ [ 53%]
...............................................................          [100%]
135 passed in 3.35s

pydantic_ai_slim/pydantic_ai/_agent_graph.py 103 293                         f'Model token limit ({self.last_max_tokens or "provider default"}) exceeded while generating a tool call, resulting in incomplete arguments. Increase the `max_tokens` model setting, or simplify the prompt to result in a shorter response that will fit within the limit.'
pydantic_ai_slim/pydantic_ai/_output.py 897 133                 # do we need to error on conflicts here? (DavidM): If this is internal maybe doesn't matter, if public, use overloads
pydantic_ai_slim/pydantic_ai/_output.py 909 146             processor = ObjectOutputProcessor(output=output, description=description, strict=strict)  # pyright: ignore[reportUnknownArgumentType]
pydantic_ai_slim/pydantic_ai/_tool_manager.py 84 131                 - 'parallel_ordered_events': Run tool calls in parallel, but events are emitted in order, after all call
…[211 chars omitted]…
pped
pydantic_ai_slim/pydantic_ai/agent/__init__.py 152 152     - `'exhaustive'`: Output tools are executed first, then all function tools are executed. The first valid output tool result becomes the final output
pydantic_ai_slim/pydantic_ai/agent/__init__.py 2134 135         """All toolsets registered on the agent, including a function toolset holding tools that were registered on the agent directly.

106 def get_wrapper_toolset(self, toolset: AbstractToolset[AgentDepsT]) -> AbstractToolset[AgentDepsT] | None:
120             if self.tools is not None and (tool := self.tools.get(name)) is not None and tool.tool_def.kind == 'output':
87                 max_retries=self.tool_max_retries.get(tool_def.name, self.max_retries),
        When the callable is invoked, `ctx.model_settings` contains the merged
        result of all layers resolved before this capability (model defaults and
        agent-level settings). The returned dict is merged on top of that.
        """
        return None

    def get_toolset(self) -> AgentToolset[AgentDepsT] | None:
        """Return a toolset to register with the agent, or None."""
        return None

    def get_wrapper_toolset(
        self, toolset: AbstractToolset[AgentDepsT]
    ) -> AbstractToolset[AgentDepsT] | None:
        """Return a wrappe
…[37 chars omitted]…
, or None.

        Called once per run after the agent-level `prepare_tools` function has been applied.
        Output tools are not included in the toolset passed to this hook.
        """
        return None

    def get_builtin_tools(self) -> Sequence[AgentBuiltinTool[AgentDepsT]]:
        """Return builtin tools to register with the agent."""
        return []

    # --- Tool preparation hook ---

137:class NativeOutput(Generic[OutputDataT]):
138-    """Marker class to use the model's native structured outputs functionality for outputs and optionally customize the name and description.
139-
140-    Example:
141-    ```python {title="native_output.py" requires="tool_output.py"}
142-    from pydantic_ai import Agent, NativeOutput
143-
144-    from tool_output import Fruit, Vehicle
145-
146-    agent = Agent(
147-        'openai:gpt-5.2',
148-        output_type=NativeOutput(

149-            [Fruit, Vehicle],
150-            name='Fruit or vehicle',
151-            description='Return a fruit or vehicle.'
152-        ),
153-    )
154-    result = agent.run_sync('What is a Ford Explorer?')
155-    print(repr(result.output))
156-    #> Vehicle(name='Ford Explorer', wheels=4)
157-    ```
158-    """
159-
160-    outputs: OutputTypeOrFunction[OutputDataT] | Sequence[OutputTypeOrFuncti
…[8628 chars omitted]…
etTool[AgentDepsT]
    ) -> Any:
        output = await self.processors[name].call(tool_args, ctx, wrap_validation_errors=False)
        for validator in self.output_validators:
            output = await validator.validate(output, ctx, wrap_validation_errors=False)
        return output


@overload
def _flatten_output_spec(
    output_spec: OutputTypeOrFunction[T] | Sequence[OutputTypeOrFunction[T]],

59:class TestModel(Model):
60-    """A model specifically for testing purposes.
61-
62-    This will (by default) call all tools in the agent, then return a tool response if possible,
63-    otherwise a plain response.
64-
65-    How useful this model is will vary significantly.
66-
67-    Apart from `__init__` derived by the `dataclass` decorator, all methods are private or match those
68-    of the base class.
69-    """
70-

71-    # NOTE: Avoid test discovery by pytest.
72-    __test__ = False
73-
74-    call_tools: list[str] | Literal['all'] = 'all'
75-    """List of tools to call. If `'all'`, all tools will be called."""
76-    custom_output_text: str | None = None
77-    """If set, this text is returned as the final output."""
78-    custom_output_args: Any | None = None
79-    """If set, these args will be passed to the output tool."""
80-    seed: int = 0
81-    """Seed for ge
…[5665 chars omitted]…
       messages: list[ModelMessage],
204-        model_settings: ModelSettings | None,
205-        model_request_parameters: ModelRequestParameters,
206-    ) -> ModelResponse:
207-        if model_request_parameters.builtin_tools:
208-            raise UserError('TestModel does not support built-in tools')
209-
docs/capabilities.md:141:from pydantic_ai.toolsets import AbstractToolset, PrefixedToolset

first [(0, 2), (1, 2), (2, 2)] [(0, 4), (1, 4), (2, 4)]

168:def test_result_pydantic_model_retry():
1172:def test_output_type_function_with_retry():
1261:def test_output_type_text_output_function_with_retry():
2172:def test_prompted_output_function_with_retry():
3167:def test_unknown_tool_per_tool_retries_exceeded():
3175:    with pytest.raises(UnexpectedModelBehavior, match=r"Tool 'foobar' exceeded max retries count of 1"):
4344:    def test_exhaustive_strategy_with_tool_retry_and_final_result(self):
4552:    def test_exhaustive_strategy_second_output_max_retries_exceeded(self):
4553:        """Test exhaustive strategy when first output succeeds and second exceeds max retries."""
4583:        result = agent.run_sync('test exhaustive with max retries exceeded')
4589:    def test_early_strategy_second_output_max_retries_exceeded(self):
4590:        """Test early strategy when first output succeeds and second exceeds max retries."""

4617:   
…[2395 chars omitted]…
iption
132-        self.max_retries = max_retries
133-        self.strict = strict
134-
135-
136-@dataclass(init=False)
137-class NativeOutput(Generic[OutputDataT]):
138-    """Marker class to use the model's native structured outputs functionality for outputs and optionally customize the name and description.
139-
140-    Example:
141-    ```python {title="native_output.py" requires="tool_output.py"}

52:from pydantic_ai.output import PromptedOutput, TextOutput, ToolOutput
932:        agent = Agent(FunctionModel(stream_function=sf), output_type=ToolOutput(process_foo, name='my_output'))
955:        agent = Agent(FunctionModel(stream_function=sf), output_type=ToolOutput(process_foo, name='my_output'))
988:        agent = Agent(FunctionModel(stream_function=sf), output_type=ToolOutput(process_foo, name='my_output'))
1034:        agent = Agent(FunctionModel(stream_function=sf), output_type=ToolOutput(process_foo, name='my_output'))
1098:        agent = Agent(FunctionModel(stream_function=sf), output_type=ToolOutput(process_foo, name='my_output'))
1121:        agent = Agent(FunctionModel(stream_function=sf), output_type=ToolOutput(process_foo, name='my_output'))
1266:                ToolOutput(process_first, name='first_output'),
1267:                ToolOutput(process_second, name='se
…[8629 chars omitted]…
e got appropriate messages
        assert result.all_messages() == snapshot(

            [
                ModelRequest(
                    parts=[
                        UserPromptPart(
                            content='test exhaustive with tool retry', timestamp=IsNow(tz=datetime.timezone.utc)
                        )
                    ],
                    timestamp=IsNow(tz=timezone.utc),

tests/test_agent_output_schemas.py
tests/test_native_output_schema.py
 M docs/capabilities.md
 M docs/output.md
 M pydantic_ai_slim/pydantic_ai/_agent_graph.py
 M pydantic_ai_slim/pydantic_ai/_output.py
 M pydantic_ai_slim/pydantic_ai/_tool_manager.py
 M pydantic_ai_slim/pydantic_ai/agent/__init__.py
 M pydantic_ai_slim/pydantic_ai/capabilities/abstract.py
 M pydantic_ai_slim/pydantic_ai/capabilities/combined.py
 docs/capabilities.md                               | 36 ++++++++++++++++++++++
 docs/output.md                                     |  2 +-

 pydantic_ai_slim/pydantic_ai/_agent_graph.py       | 11 ++-----
 pydantic_ai_slim/pydantic_ai/_output.py            | 13 ++++++--
 pydantic_ai_slim/pydantic_ai/_tool_manager.py      |  4 +++
 pydantic_ai_slim/pydantic_ai/agent/__init__.py     | 16 +++++++---
 .../pydantic_ai/capabilities/abstract.py           | 12 +++++++-
 .../pydantic_ai/capabilities/combined.py           |  9 ++++++
 8 files changed, 86 insertions(+), 17 deletions(-)

498:class ToolDefinition:
499-    """Definition of a tool passed to a model.
500-
501-    This is used for both function tools and output tools.
502-    """
503-
504-    name: str
505-    """The name of the tool."""
506-
507-    parameters_json_schema: ObjectJsonSchema = field(default_factory=lambda: {'type': 'object', 'properties': {}})
508-    """The JSON schema for the tool's parameters."""
509-

510-    description: str | None = None
511-    """The description of the tool."""
512-
513-    outer_typed_dict_key: str | None = None
514-    """The key in the outer [TypedDict] that wraps an output tool.
515-
516-    This will only be set for output tools which don't have an `object` JSON schema.
517-    """
518-
519-    strict: bool | None = None
520-    """Whether to enforce (vendor-specific) strict JSON schema validation for tool calls.
521-
522-    Setting this to `True` while using a
…[2074 chars omitted]…
ty
559-    def defer(self) -> bool:
560-        """Whether calls to this tool will be deferred.
561-
562-        See the [tools documentation](../deferred-tools.md#deferred-tools) for more info.
563-        """
564-        return self.kind in ('external', 'unapproved')
565-
566-    __repr__ = _utils.dataclasses_no_defaults_repr
pydantic_ai_slim/pydantic_ai/_output.py:937:                kind='output',

AGENTS.md
BRIEF_BUILTIN_TOOLS_CAP.md
BRIEF_FOR_RUN_STEP.md
BRIEF_OTEL_SPANS.md
BRIEF_TEMPORAL_TOOLSET.md
CLAUDE.md
LICENSE
Makefile
README.md
agent_docs
clai
docs

docs-site
examples
mkdocs.yml
pydantic_ai_slim
pydantic_evals
pydantic_graph
pyproject.toml
scripts
tests
uv.lock
184b7ca base
.
./.claude
./.claude/skills
./.gemini
./.git
./.git/branches
./.git/hooks
./.git/info
./.git/logs
./.git/objects
./.git/refs
./.github
./.github/ISSUE_TEMPLATE
./.github/workflows
./agent_docs
./clai
./clai/clai
./docs
./docs/.cursor
./docs/.hooks
./docs/.overrides
./docs/.partials
./docs/api
./docs/durable_execution
./docs/evals
./docs/examples
./docs/extra
./docs/graph
./docs/img
./docs/javascripts
./docs/mcp
./docs/models
./docs/ui
./docs-site
./docs-site/src
./examples
./examples/pydantic_ai_examples
./pydantic_ai_slim
./pydantic_ai_slim/pydantic_ai
./pydantic_evals
./pydantic_evals/pydantic_evals
./pydantic_graph
./pydantic_graph/pydantic_graph
./scripts
./tests
./tests/assets
./tests/cassettes
./tests/evals
./tests/example_modules
./tests/ext

pydantic_ai_slim/pydantic_ai/tools.py:276:    max_retries: int | None
pydantic_ai_slim/pydantic_ai/tools.py:300:        max_retries: int | None = None,
pydantic_ai_slim/pydantic_ai/tools.py:353:            max_retries: Maximum number of retries allowed for this tool, set to the agent default if `None`.
pydantic_ai_slim/pydantic_ai/tools.py:388:        self.max_retries = max_retries
pydantic_ai_slim/pydantic_ai/_agent_graph.py:998:            max_retries=ctx.deps.max_result_retries,

pydantic_ai_slim/pydantic_ai/agent/spec.py:19:    from pydantic_ai.capabilities.abstract import AbstractCapability
pydantic_ai_slim/pydantic_ai/agent/spec.py:120:        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
pydantic_ai_slim/pydantic_ai/agent/spec.py:175:        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
pydantic_ai_slim/pydantic_ai/agent/spec.py:221:        custom_capability_types: Sequence[type[AbstractCapability[Any]]] = (),
pydantic_ai_slim/pydantic_ai/agent/spec.py:251:    custom_types: Sequence[type[AbstractCapability[Any]]] = (),
pydantic_ai_slim/pydantic_ai/agent/spec.py:252:) -> Mapping[str, type[AbstractCapability[Any]]]:
pydantic_ai_slim/pydantic_ai/agent/spec.py:255:    from pydantic_ai.capabilities.abstract import AbstractCapability
pydantic_ai_slim/pydantic_ai/agent/spec.py:257:    def _validate_capability(cls:
…[8629 chars omitted]…
       max_retries: int = 1,
pydantic_ai_slim/pydantic_ai/mcp.py:1073:            max_retries: The maximum number of times to retry a tool call.
pydantic_ai_slim/pydantic_ai/mcp.py:1108:            max_retries=max_retries,

pydantic_ai_slim/pydantic_ai/models/gemini.py:242:                    'Gemini does not support `NativeOutput` and tools at the same time. Use `output_type=ToolOutput(...)` instead.'

from __future__ import annotations

from abc import ABC
from collections.abc import AsyncIterable, Awaitable, Callable, Sequence
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Generic

from pydantic_ai._instructions import AgentInstructions
from pydantic_ai.messages import AgentStreamEvent, ModelResponse, ToolCallPart
from pydantic_ai.models import ModelRequestContext
from pydantic_ai.tools import AgentBuiltinTool, AgentDepsT, RunContext, ToolDefinition
from pydantic_ai.toolsets import AgentToolset

if TYPE_CHECKING:
    from pydantic_ai import _agent_graph
    from pydantic_ai.agent.abstract import AgentModelSettings
    from pydantic_ai.result import FinalResult
    from pydantic_ai.run import AgentRunResult
    from pydantic_graph import End


@dataclass
class AbstractCapability(ABC, Generic[AgentDepsT]):
    """Abstract base class for agent capabilities.

…[8629 chars omitted]…
psT]

    @classmethod
    def get_serialization_name(cls) -> str | None:
        return None  # Not spec-serializable (takes a callable)

    async def prepare_tools(self, ctx: RunContext[AgentDepsT], tool_defs: list[ToolDefinition]) -> list[ToolDefinition]:
        result: list[ToolDefinition] | None = await self.prepare_func(ctx, tool_defs)
        return result if result is not None else tool_defs

# --- Tool preparation hook ---

    async def prepare_tools(
        self,
        ctx: RunContext[AgentDepsT],
        tool_defs: list[ToolDefinition],
    ) -> list[ToolDefinition]:
        """Filter or modify tool definitions visible to the model for this step.

        The list contains all tool kinds (function, output, unapproved) distinguished
        by [`tool_def.kind`][pydantic_ai.tools.ToolDefinition.kind]. Return a filtered
        or modified list. Called after the agent-level

        [`prepare_tools`][pydantic_ai.tools.ToolsPrepareFunc] has already run.
        """
        return tool_defs

    # --- Run lifecycle hooks ---

    async def before_run(
        self,
        ctx: RunContext[AgentDepsT],
    ) -> None:
        """Called before the agent run starts. Observe-only; use wrap_run for modification."""

    async def after_run(
        self,
        ctx: RunCon
…[8629 chars omitted]…
           cap_model_settings = run_capability.get_model_settings()
1071:            cap_ts = run_capability.get_toolset()
1072:            cap_toolsets = [cap_ts] if cap_ts is not None else []
1077:            cap_toolsets = None
1079:        # Build model settings resolver using per-run capability
1092:            # Capability settings (e.g. from Thinking, ModelSettings capabilities), cached at init

The result of the run.
        """
        if infer_name and self.name is None:
            self._infer_name(inspect.currentframe())

        model_used = self._get_model(model)
        del model

        deps = self._get_deps(deps)
        output_schema = self._prepare_output_schema(output_type)

        output_type_ = output_type or self.output_type

        # We consider it a user error if a user tries to restrict the result type while having an output validator that
        # may change the result type from the restricted type to something else. Therefore, we consider the following
        # typecast reasonable, even though it is possible to violate it with otherwise-type-checked code.
        output_validators = self._output_validators

        output_toolset = self._output_toolset
        if output_schema != self._output_schema or output_validators:
            outpu
…[7616 chars omitted]…
       state=state,
                    deps=graph_deps,
                    span=use_span(run_span) if run_span.is_recording() else None,
                    infer_name=False,
                ) as graph_run,
            ):
                async with toolset:
                    agent_run = AgentRun(graph_run)
                    run_metadata = self._resolve_and_store_metadata(agent_run.ctx, metadata)

pydantic_ai_slim/pydantic_ai/agent/__init__.py:184:    _output_toolset: OutputToolset[AgentDepsT] | None = dataclasses.field(repr=False)
pydantic_ai_slim/pydantic_ai/agent/__init__.py:186:    _prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = dataclasses.field(repr=False)
pydantic_ai_slim/pydantic_ai/agent/__init__.py:218:        prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:250:        prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:280:        prepare_tools: ToolsPrepareFunc[AgentDepsT] | None = None,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:325:            prepare_tools: Custom function to prepare the tool definition of all tools for each step, except output tools.
pydantic_ai_slim/pydantic_ai/agent/__init__.py:427:        self._prepare_tools = prepare_tool
…[5715 chars omitted]…
n_strict_if_openai)
pydantic_ai_slim/pydantic_ai/tools.py:276:    max_retries: int | None
pydantic_ai_slim/pydantic_ai/tools.py:300:        max_retries: int | None = None,
pydantic_ai_slim/pydantic_ai/tools.py:353:            max_retries: Maximum number of retries allowed for this tool, set to the agent default if `None`.

pydantic_ai_slim/pydantic_ai/tools.py:388:        self.max_retries = max_retries

for instruction in instructions:
            if isinstance(instruction, str):
                literal_parts.append(instruction)
            else:
                # TemplateStr instances land here too: they are callable with a
                # RunContext parameter, so SystemPromptRunner handles them like
                # any other system prompt function.
                functions.append(_system_prompt.SystemPromptRunner[AgentDepsT](instruction))

        literal = '\n'.join(literal_parts).strip() or None
        return literal, functions

    def _get_toolset(
        self,
        output_toolset: AbstractToolset[AgentDepsT] | None | _utils.Unset = _utils.UNSET,
        additional_toolsets: Sequence[AbstractToolset[AgentDepsT]] | None = None,
        cap_toolsets: Sequence[AgentToolset[AgentDepsT]] | None = None,
    ) -> AbstractToolset[AgentDepsT]:
        """Get the compl
…[8628 chars omitted]…
ge_history

        # To allow this message history to be used in a future run without dangling tool calls,
        # append a new ModelRequest using the tool returns and retries
        if tool_responses:
            messages.append(_messages.ModelRequest(parts=tool_responses, run_id=ctx.state.run_id, timestamp=now_utc()))

        return End(final_result)

    __repr__ = dataclasses_no_defaults_repr

21:from pydantic_ai._tool_manager import ToolManager, ValidatedToolCall
138:    tool_manager: ToolManager[DepsT]
405:    output_tools: list[ToolDefinition] = []
408:            output_tools.append(tool_def)
429:        output_tools=output_tools,
803:                    # Don't retry if the token limit was exceeded, possibly during thinking.
853:                    # normal retry prompt below.
920:                        _messages.ModelRequest(parts=[e.tool_retry], instructions=instructions)
941:        async for event in process_tool_calls(
997:            retry=ctx.state.retries,
998:            max_retries=ctx.deps.max_result_retries,
1004:            result_data = await validator.validate(result_data, run_context)

1081:def _emit_skipped_output_tool(
1099:async def process_tool_calls(  # noqa: C901
1100:    tool_manager: ToolManager[DepsT],
1126:        # `final_result` can be passe
…[2650 chars omitted]…
nager: ToolManager[DepsT],
1478:        validated = tool_call
1481:        validated = None
1486:            if validated is not None:
1487:                tool_result = await tool_manager.execute_tool_call(validated)
1489:                raise RuntimeError('Expected validated tool call')  # pragma: no cover
1511:        return e.tool_retry, None
1617:    return g.build(validate_graph_structure=False)

async def process_tool_calls(  # noqa: C901
    tool_manager: ToolManager[DepsT],
    tool_calls: list[_messages.ToolCallPart],
    tool_call_results: dict[str, DeferredToolResult | Literal['skip']] | None,
    tool_call_metadata: dict[str, dict[str, Any]] | None,
    final_result: result.FinalResult[NodeRunEndT] | None,
    ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, NodeRunEndT]],
    output_parts: list[_messages.ModelRequestPart],
    output_final_result: deque[result.FinalResult[NodeRunEndT]] = deque(maxlen=1),
) -> AsyncIterator[_messages.HandleResponseEvent]:
    """Process function (i.e., non-result) tool calls in parallel.

    Also add stub return parts for any other tools that need it.

    Because async iterators can't have return values, we use `output_parts` and `output_final_result` as output arguments.
    """
    tool_calls_by_kind: dict[ToolKind | Lite
…[8629 chars omitted]…
idate wraps the validation
            validated_args = await cap.wrap_tool_validate(ctx, call=call, args=raw_args, handler=do_validate)

            # after_tool_validate
            validated_args = await cap.after_tool_validate(ctx, call=call, args=validated_args)
        else:
            validated_args = await do_validate(call.args if call.args is not None else {})

        return validated_args

from __future__ import annotations

import inspect
import json
from collections.abc import Iterator
from contextlib import contextmanager
from contextvars import ContextVar
from dataclasses import dataclass, field, replace
from typing import TYPE_CHECKING, Any, Generic, Literal

from opentelemetry.trace import Tracer
from pydantic import ValidationError

from typing_extensions import deprecated

from . import messages as _messages
from ._instrumentation import InstrumentationNames
from ._run_context import AgentDepsT, RunContext
from .exceptions import ModelRetry, SkipToolExecution, SkipToolValidation, ToolRetryError, UnexpectedModelBehavior
from .messages import ToolCallPart
from .tools import ToolDefinition
from .toolsets.abstract import AbstractToolset, ToolsetTool
from .usage import RunUsage

if TYPE_CHECKING:
    from .capabilities.abstract import AbstractCapability

ParallelExecu
…[8275 chars omitted]…
.isawaitable(result):
                await result

        return args_dict

    async def _run_validate_hooks(
        self,
        call: ToolCallPart,
        tool: ToolsetTool[AgentDepsT],
        ctx: RunContext[AgentDepsT],
        *,
        allow_partial: bool,
    ) -> dict[str, Any]:
        """Run validation with before/wrap/after tool_validate hooks."""
        cap = self.root_capability

async def do_validate(args: str | dict[str, Any]) -> dict[str, Any]:
            # Update call.args with the (possibly modified) args before validation
            validated = await self._validate_tool_args(call, tool, ctx, allow_partial=allow_partial, args_override=args)
            return validated

        if cap is not None:
            # before_tool_validate
            raw_args: str | dict[str, Any] = call.args if call.args is not None else {}
            raw_args = await cap.before_tool_validate(ctx, call=call, args=raw_args)

            # wrap_tool_validate wraps the validation
            validated_args = await cap.wrap_tool_validate(ctx, call=call, args=raw_args, handler=do_validate)

            # after_tool_validate
            validated_args = await cap.after_tool_validate(ctx, call=call, args=validated_args)
        else:
            validated_args = await do_v
…[8629 chars omitted]…
me': {},
                        'gen_ai.tool.call.id': {},
                    },
                }
            ),
        }

        with tracer.start_as_current_span(
            instrumentation_names.get_tool_span_name(call.tool_name),
            attributes=span_attributes,
        ) as span:
            try:
                tool_result = await self._execute_tool_call_impl(validated, usage=usage)

msg = f'Available tools: {", ".join(f"{n!r}" for n in self.tools)}'
                else:
                    msg = 'No tools available.'
                raise ModelRetry(f'Unknown tool name: {name!r}. {msg}')

            ctx = self._build_tool_context(
                call, tool, allow_partial=allow_partial, approved=approved, metadata=metadata
            )

            validated_args = await self._run_validate_hooks(call, tool, ctx, allow_partial=allow_partial)
            return ValidatedToolCall(
                call=call,

                tool=tool,
                ctx=ctx,
                args_valid=True,
                validated_args=validated_args,
                validation_error=None,
            )
        except SkipToolValidation as e:
            if tool is None:
                raise ValueError('Cannot skip validation for unknown tool')  # pragma: n
…[4414 chars omitted]…
deps.max_result_retries, error=e)
1189:                # If we already have a valid final result, don't increment retries for invalid output tools
1192:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1225:        ctx.state.increment_retries(ctx.deps.max_result_retries)
1324:                    # retries are already tracked by validate_tool_call() via failed_tools.

metadata: dict[str, dict[str, Any]] = field(default_factory=dict[str, dict[str, Any]])
    """Metadata for deferred tool calls, keyed by `tool_call_id`. Each value will be available in the tool's RunContext as `tool_call_metadata`."""


A = TypeVar('A')


class GenerateToolJsonSchema(GenerateJsonSchema):
    def _named_required_fields_schema(self, named_required_fields: Sequence[tuple[str, bool, Any]]) -> JsonSchemaValue:
        # Remove largely-useless property titles
        s = super()._named_required_fields_schema(named_required_fields)
        for p in s.get('properties', {}):

            s['properties'][p].pop('title', None)
        return s


ToolAgentDepsT = TypeVar('ToolAgentDepsT', default=object, contravariant=True)
"""Type variable for agent dependencies for a tool."""


@dataclass(init=False)
class Tool(Generic[ToolAgentDepsT]):
    """A tool function for an agent.""
…[8628 chars omitted]…
function.py:499:            tool.max_retries = self.max_retries
pydantic_ai_slim/pydantic_ai/toolsets/function.py:507:            max_retries = tool.max_retries if tool.max_retries is not None else self.max_retries
pydantic_ai_slim/pydantic_ai/toolsets/function.py:512:                max_retries=max_retries,
pydantic_ai_slim/pydantic_ai/toolsets/function.py:528:                max_retries=max_retries,

*,
        run_context: RunContext[AgentDepsT],
        validation_context: Any | None = None,
        allow_partial: bool = False,
        wrap_validation_errors: bool = True,
    ) -> OutputDataT:
        args = {self._str_argument_name: data}
        data = await execute_traced_output_function(self._function_schema, run_context, args, wrap_validation_errors)

        return await super().process(
            data,
            run_context=run_context,

            validation_context=validation_context,
            allow_partial=allow_partial,
            wrap_validation_errors=wrap_validation_errors,
        )


@dataclass(init=False)
class OutputToolset(AbstractToolset[AgentDepsT]):
    """A toolset that contains contains output tools for agent output types."""

    _tool_defs: list[ToolDefinition]
    """The tool definitions for the output tools in this toolset."""
    proc
…[5849 chars omitted]…
   """A toolset that contains contains output tools for agent output types."""
863:    """The tool definitions for the output tools in this toolset."""
865:    """The processors for the output tools in this toolset."""
866:    max_retries: int
943:        max_retries: int = 1,
948:        self.max_retries = max_retries
962:                toolset=self,
964:                max_retries=self.max_retries,

pydantic_ai_slim/pydantic_ai/output.py:76:class ToolOutput(Generic[OutputDataT]):
pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py:164:class ToolOutputAvailablePart(BaseUIPart):
pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py:178:class ToolOutputErrorPart(BaseUIPart):
pydantic_ai_slim/pydantic_ai/ui/vercel_ai/request_types.py:216:class ToolOutputDeniedPart(BaseUIPart):
pydantic_ai_slim/pydantic_ai/ui/vercel_ai/response_types.py:112:class ToolOutputAvailableChunk(BaseChunk):
pydantic_ai_slim/pydantic_ai/ui/vercel_ai/response_types.py:148:class ToolOutputErrorChunk(BaseChunk):
pydantic_ai_slim/pydantic_ai/ui/vercel_ai/response_types.py:169:class ToolOutputDeniedChunk(BaseChunk):
pydantic_ai_slim/pydantic_ai/_output.py:489:class ToolOutputSchema(OutputSchema[OutputDataT]):
from __future__ import annotations as _annotations

import inspect
import json

from abc import 
…[82 chars omitted]…
ut[OutputDataT]] = []
310:            elif isinstance(output, ToolOutput):
336:                return ToolOutputSchema(
350:            return ToolOutputSchema(
489:class ToolOutputSchema(OutputSchema[OutputDataT]):
872:        outputs: list[OutputTypeOrFunction[OutputDataT] | ToolOutput[OutputDataT]],
892:            if isinstance(output, ToolOutput):
1022:        elif isinstance(output, ToolOutput):

TextOutputFunc = TypeAliasType(
    'TextOutputFunc',
    Callable[[RunContext[Any], str], Awaitable[T_co] | T_co] | Callable[[str], Awaitable[T_co] | T_co],
    type_params=(T_co,),
)
"""Definition of a function that will be called to process the model's plain text output. The function must take a single string argument.

You should not need to import or use this type directly.

See [text output docs](../output.md#text-output) for more information.
"""

@dataclass(init=False)
class ToolOutput(Generic[OutputDataT]):
    """Marker class to use a tool for output and optionally customize the tool.

    Example:
    ```python {title="tool_output.py"}
    from pydantic import BaseModel

    from pydantic_ai import Agent, ToolOutput


    class Fruit(BaseModel):
        name: str
        color: str


    class Vehicle(BaseModel):
        name: str
        wheels: int


    agent = Agent(
…[7174 chars omitted]…
ocessor(other_outputs, name=name, description=description, strict=strict),
                toolset=toolset,
                allows_deferred_tools=allows_deferred_tools,
                allows_image=allows_image,
            )

        if allows_image:
            return ImageOutputSchema(allows_deferred_tools=allows_deferred_tools)

        raise UserError('At least one output type must be provided.')

tests/test_agent.py
tests/test_fastmcp.py
tests/test_tools.py
tests/test_toolsets.py
pydantic_ai_slim/pydantic_ai/toolsets/abstract.py
pydantic_ai_slim/pydantic_ai/toolsets/prefixed.py
pydantic_ai_slim/pydantic_ai/toolsets/__init__.py
pydantic_ai_slim/pydantic_ai/__init__.py

tests/test_agent.py:38:    PrefixedToolset,
tests/test_agent.py:6399:    toolset = CombinedToolset([server1, PrefixedToolset(server2, 'prefix')])
tests/test_agent.py:6419:    toolset = CombinedToolset([server1, PrefixedToolset(server2, 'prefix')])
tests/test_agent.py:6579:    toolset = CombinedToolset([server1, PrefixedToolset(server2, 'prefix')])
tests/test_tools.py:22:    PrefixedToolset,
tests/test_tools.py:659:            "The agent defines a tool whose name conflicts with existing tool from the agent's output tools: 'ctx_tool'. Rename the tool or wrap the toolset in a `PrefixedToolset` to avoid name conflicts."
tests/test_tools.py:671:            "PrefixedToolset(FunctionToolset 'tool') defines a tool whose name conflicts with existing tool from the agent: 'foo_tool'. Change the `prefix` attribute to avoid name conflicts."
tests/test_tools.py:682:        prefixed_toolset = Prefix
…[46 chars omitted]…
toolsets.py:18:    PrefixedToolset,
tests/test_toolsets.py:285:    prefixed_math = PrefixedToolset(math_toolset, 'math')
tests/test_toolsets.py:286:    prefixed_string = PrefixedToolset(string_toolset, 'str')

tests/test_toolsets.py:287:    prefixed_advanced = PrefixedToolset(advanced_toolset, 'adv')
tests/test_toolsets.py:497:    toolset = CombinedToolset([server1, PrefixedToolset(server2, 'prefix')])

from __future__ import annotations

from collections.abc import AsyncIterable, Awaitable, Callable, Sequence
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING, Any

from pydantic_ai import _system_prompt
from pydantic_ai._instructions import AgentInstructions, normalize_instructions
from pydantic_ai.messages import AgentStreamEvent, ModelResponse, ToolCallPart
from pydantic_ai.settings import ModelSettings, merge_model_settings
from pydantic_ai.tools import AgentBuiltinTool, AgentDepsT, RunContext, ToolDefinition
from pydantic_ai.toolsets import AbstractToolset, AgentToolset, CombinedToolset

from pydantic_ai.toolsets._dynamic import DynamicToolset

from .abstract import AbstractCapability, ModelRequestContext

if TYPE_CHECKING:
    from pydantic_ai import _agent_graph
    from pydantic_ai.result import FinalResult
    from pydantic_ai.run import AgentRunResul
…[8629 chars omitted]…
.get_serialization_name()) is not None
}
"""Registry of all capability types that have a serialization name, mapping name to class."""

__all__ = [
    'AbstractCapability',
    'ModelRequestContext',
    'CAPABILITY_TYPES',
    'BuiltinTool',
    'Instructions',
    'HistoryProcessor',
    'ModelSettings',
    'PrepareTools',
    'Thinking',
    'Toolset',
    'WebSearch',
    'CombinedCapability',
]

184:    _output_toolset: OutputToolset[AgentDepsT] | None = dataclasses.field(repr=False)
188:    _max_result_retries: int = dataclasses.field(repr=False)
417:        self._max_result_retries = output_retries if output_retries is not None else retries
430:        self._output_toolset = self._output_schema.toolset
431:        if self._output_toolset:
432:            self._output_toolset.max_retries = self._max_result_retries
860:    def iter(
880:    def iter(
900:    async def iter(  # noqa: C901
1009:        output_schema = self._prepare_output_schema(output_type)
1018:        output_toolset = self._output_toolset
1020:            output_toolset = output_schema.toolset

1021:            if output_toolset:
1022:                output_toolset.max_retries = self._max_result_retries
1023:                output_toolset.output_validators = output_validators
1106:        toolset = self._get_
…[628 chars omitted]…
mbinedToolset([output_toolset, toolset])
2164:    def _prepare_output_schema(self, output_type: None) -> _output.OutputSchema[OutputDataT]: ...
2167:    def _prepare_output_schema(
2171:    def _prepare_output_schema(self, output_type: OutputSpec[Any] | None) -> _output.OutputSchema[Any]:
2191:                    toolset = self._get_toolset()
2221:        self._get_toolset().apply(_set_sampling_model)

'HistoryProcessor',
)


T = TypeVar('T')
S = TypeVar('S')
NoneType = type(None)
EndStrategy = Literal['early', 'exhaustive']
DepsT = TypeVar('DepsT')
OutputT = TypeVar('OutputT')

@dataclasses.dataclass(kw_only=True)
class GraphAgentState:
    """State kept across the execution of the agent graph."""

    message_history: list[_messages.ModelMessage] = dataclasses.field(default_factory=list[_messages.ModelMessage])
    usage: _usage.RunUsage = dataclasses.field(default_factory=_usage.RunUsage)
    retries: int = 0
    run_step: int = 0
    run_id: str = dataclasses.field(default_factory=lambda: str(uuid.uuid4()))
    metadata: dict[str, Any] | None = None
    last_max_tokens: int | None = None
    """Last-resolved `max_tokens` from model settings, used only in error messages."""

    def increment_retries(
        self,
        max_result_retries: int,
        error: BaseExceptio
…[5015 chars omitted]…
                      await tool_manager.execute_tool_call(validated)
1349:    tool_manager: ToolManager[DepsT],
1388:    parallel_execution_mode = tool_manager.get_parallel_execution_mode(tool_calls)
1393:                    tool_manager,
1405:                    tool_manager,
1473:    tool_manager: ToolManager[DepsT],
1487:                tool_result = await tool_manager.execute_tool_call(validated)

self, ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, NodeRunEndT]]
    ) -> End[result.FinalResult[NodeRunEndT]]:
        return End(self.final_result)


def build_run_context(ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, Any]]) -> RunContext[DepsT]:
    """Build a `RunContext` object from the current agent graph run context."""
    run_context = RunContext[DepsT](
        deps=ctx.deps.user_deps,
        model=ctx.deps.model,
        usage=ctx.state.usage,
        prompt=ctx.deps.prompt,

        messages=ctx.state.message_history,
        validation_context=None,
        tracer=ctx.deps.tracer,
        trace_include_content=ctx.deps.instrumentation_settings is not None
        and ctx.deps.instrumentation_settings.include_content,
        instrumentation_version=ctx.deps.instrumentation_settings.version
        if ctx.deps.instrumentation_settings
 
…[2982 chars omitted]…
tep(run_context)

        output_parts: list[_messages.ModelRequestPart] = []
        output_final_result: deque[result.FinalResult[NodeRunEndT]] = deque(maxlen=1)

        async for event in process_tool_calls(
            tool_manager=ctx.deps.tool_manager,
            tool_calls=tool_calls,
            tool_call_results=self.tool_call_results,
            tool_call_metadata=self.tool_call_metadata,

grep: tests/test_output.py: No such file or directory
tests/test_agent.py:168:def test_result_pydantic_model_retry():
tests/test_agent.py:271:            ('request', ['retry-prompt']),
tests/test_agent.py:277:    user_retry = result.all_messages()[2]
tests/test_agent.py:278:    assert isinstance(user_retry, ModelRequest)
tests/test_agent.py:279:    retry_prompt = user_retry.parts[0]
tests/test_agent.py:280:    assert isinstance(retry_prompt, RetryPromptPart)
tests/test_agent.py:281:    assert retry_prompt.model_response() == snapshot("""\
tests/test_agent.py:370:def test_output_validator_retries():
tests/test_agent.py:371:    """Test that ctx.retry and ctx.max_retries are correctly tracked in RunContext for output validators."""
tests/test_agent.py:382:    agent = Agent(FunctionModel(return_model), output_type=Foo, output_retries=target_retries)
tests/test_agent.py:386:        retries

…[3899 chars omitted]…
ync def test_retry_message_no_tools():
tests/test_agent.py:6742:    """Test that thinking-only retry message does not suggest 'call a tool' when no function tools are registered."""
tests/test_agent.py:6799:async def test_thinking_only_response_retry_with_tool_output():
tests/test_agent.py:6800:    """Test that thinking-only responses retry with tool-output guidance when text output is not allowed."""

timestamp=IsNow(tz=timezone.utc),
                    )
                ],
                timestamp=IsNow(tz=timezone.utc),
                run_id=IsStr(),
            ),
        ]
    )


def test_output_validator_retries():
    """Test that ctx.retry and ctx.max_retries are correctly tracked in RunContext for output validators."""

    retries_log: list[int] = []
    max_retries_log: list[int] = []
    target_retries = 3

    def return_model(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
        assert info.output_tools is not None
        # Always return the same value, let the validator control retries
        args_json = '{"a": 1, "b": "foo"}'
        return ModelResponse(parts=[ToolCallPart(info.output_tools[0].name, args_json)])

    agent = Agent(FunctionModel(return_model), output_type=Foo, output_retries=target_retries)

    @agent.
…[8353 chars omitted]…
= 'valid'

    # NOTE: When changing tests in this class:
    # 1. Follow the existing order
    # 2. Update tests in `tests/test_streaming.py::TestMultipleToolCallsStreaming` as well


async def test_model_settings_override() -> None:
    def return_settings(_: list[ModelMessage], info: AgentInfo) -> ModelResponse:
        return ModelResponse(parts=[TextPart(to_json(info.model_settings).decode())])

tests/models/test_model_test.py:291:    with pytest.raises(UnexpectedModelBehavior, match=re.escape('Exceeded maximum retries (2) for output validation')):
tests/models/test_model_test.py:334:    with pytest.raises(UnexpectedModelBehavior, match=r'Exceeded maximum retries \(2\) for output validation'):
tests/models/test_openai_responses.py:7062:            UnexpectedModelBehavior, match=re.escape('Exceeded maximum retries (1) for output validation')
tests/models/test_outlines.py:455:    with pytest.raises(UnexpectedModelBehavior, match='Exceeded maximum retries .* for output validation'):
tests/test_agent.py:2839:        with pytest.raises(UnexpectedModelBehavior, match=r'Exceeded maximum retries \(1\) for output validation'):
tests/test_agent.py:3013:        with pytest.raises(UnexpectedModelBehavior, match=r'Exceeded maximum retries \(1\) for output validation'):
tests/test_agent.py
…[668 chars omitted]…
 for output validation'):
tests/test_ui.py:534:            "<error type='UnexpectedModelBehavior'>Exceeded maximum retries (1) for output validation</error>",
tests/test_vercel_ai.py:2262:            {'type': 'error', 'errorText': 'Exceeded maximum retries (1) for output validation'},
90:        self.retries += 1

91:        if self.retries > max_result_retries:
997:            retry=ctx.state.retries,

parts=[],
                usage=RequestUsage(input_tokens=51),
                model_name='function:llm:',
                timestamp=IsDatetime(),
                run_id=IsStr(),
            ),
            ModelRequest(
                parts=[],
                timestamp=IsNow(tz=timezone.utc),
                run_id=IsStr(),
            ),
            ModelResponse(

                parts=[TextPart(content='ok here is text')],
                usage=RequestUsage(input_tokens=51, output_tokens=4),
                model_name='function:llm:',
                timestamp=IsDatetime(),
                run_id=IsStr(),
            ),
        ]
    )


def test_empty_response_without_recovery():
    def llm(messages: list[ModelMessage], _info: AgentInfo) -> ModelResponse:
        return ModelResponse(parts=[])

    agent = Agent(FunctionModel(llm), output_type=tuple[str, int])

 
…[4069 chars omitted]…
, the other is used."""

        def return_model(_: list[ModelMessage], info: AgentInfo) -> ModelResponse:
            assert info.output_tools is not None
            return ModelResponse(
                parts=[
                    ToolCallPart('final_result', {'bad_value': 'first'}, tool_call_id='first'),
                    ToolCallPart('final_result', {'value': 'second'}, tool_call_id='second'),

# In both `early` and `exhaustive` modes, use the first output tool's result as the final result
            if not final_result:
                final_result = result.FinalResult(result_data, call.tool_name, call.tool_call_id)

    # Then, we handle function tool calls
    calls_to_run: list[_messages.ToolCallPart] = []
    if final_result and ctx.deps.end_strategy == 'early':
        for call in tool_calls_by_kind['function']:
            output_parts.append(
                _messages.ToolReturnPart(
                    tool_name=call.tool_name,
                    content='Tool not executed - a final result was already processed.',

                    tool_call_id=call.tool_call_id,
                )
            )
    else:
        calls_to_run.extend(tool_calls_by_kind['function'])

    # Then, we handle unknown tool calls
    if tool_calls_by_kind['unknown']:
        
…[6022 chars omitted]…
is agent.'
            )
        deferred_tool_requests = _output.DeferredToolRequests(
            calls=deferred_calls['external'],
            approvals=deferred_calls['unapproved'],
            metadata=deferred_metadata,
        )

        final_result = result.FinalResult(cast(NodeRunEndT, deferred_tool_requests), None, None)

    if final_result:
        output_final_result.append(final_result)

pydantic_ai_slim/pydantic_ai/_output.py:859:class OutputToolset(AbstractToolset[AgentDepsT]):
pydantic_ai_slim/pydantic_ai/_output.py:948:        self.max_retries = max_retries
pydantic_ai_slim/pydantic_ai/_output.py:964:                max_retries=self.max_retries,
pydantic_ai_slim/pydantic_ai/agent/__init__.py:432:            self._output_toolset.max_retries = self._max_result_retries
pydantic_ai_slim/pydantic_ai/agent/__init__.py:1022:                output_toolset.max_retries = self._max_result_retries

@property
    def event_stream_handler(self) -> EventStreamHandler[AgentDepsT] | None:
        """Optional handler for events from the model's streaming response and the agent's execution of tools."""
        return self._event_stream_handler

    def __repr__(self) -> str:
        return f'{type(self).__name__}(model={self.model!r}, name={self.name!r}, end_strategy={self.end_strategy!r}, model_settings={self.model_settings!r}, output_type={self.output_type!r}, instrument={self.instrument!r})'

    @overload
    def iter(
        self,
        user_prompt: str | Sequence[_messages.UserContent] | None = None,

        *,
        output_type: None = None,
        message_history: Sequence[_messages.ModelMessage] | None = None,
        deferred_tool_results: DeferredToolResults | None = None,
        model: models.Model | models.KnownModelName | str | None = None,
        instruction
…[6578 chars omitted]…
nfer the agent name from the call frame if it's not set.
            toolsets: Optional additional toolsets for this run.
            builtin_tools: Optional additional builtin tools for this run.

        Returns:
            The result of the run.
        """
        if infer_name and self.name is None:
            self._infer_name(inspect.currentframe())

        model_used = self._get_model(model)

48:from ..capabilities import AbstractCapability, CombinedCapability
49:from ..capabilities.builtin_tool import BuiltinTool as BuiltinToolCap
50:from ..capabilities.history_processor import HistoryProcessor as HistoryProcessorCap
229:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
261:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
291:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
369:            capabilities: Optional list of [capabilities](https://ai.pydantic.dev/capabilities/) to configure the agent with.
370:                Custom capabilities can be created by subclassing
371:                [`AbstractCapability`][pydantic_ai.capabilities.AbstractCapability].
384:        capabilities = list(capabilities or [])
386:            capabilities.append(HistoryProcessorCap(history_processor))
388:

…[8628 chars omitted]…
                description=description,
                retries=retries,
                prepare=prepare,
                args_validator=args_validator,
                docstring_format=docstring_format,
                require_parameter_descriptions=require_parameter_descriptions,
                schema_generator=schema_generator,
                strict=strict,
                sequential=sequential,

def override(  # noqa: C901
        self,
        *,
        name: str | _utils.Unset = _utils.UNSET,
        deps: AgentDepsT | _utils.Unset = _utils.UNSET,
        model: models.Model | models.KnownModelName | str | _utils.Unset = _utils.UNSET,
        toolsets: Sequence[AbstractToolset[AgentDepsT]] | _utils.Unset = _utils.UNSET,
        tools: Sequence[Tool[AgentDepsT] | ToolFuncEither[AgentDepsT, ...]] | _utils.Unset = _utils.UNSET,
        instructions: AgentInstructions[AgentDepsT] | _utils.Unset = _utils.UNSET,
        metadata: AgentMetadata[AgentDepsT] | _utils.Unset = _utils.UNSET,
        model_settings: AgentModelSettings[AgentDepsT] | _utils.Unset = _utils.UNSET,
    ) -> Iterator[None]:

        """Context manager to temporarily override agent name, dependencies, model, toolsets, tools, or instructions.

        This is particularly useful when testing.
        You ca
…[3599 chars omitted]…
entDepsT]], Awaitable[str | None]], /
    ) -> Callable[[RunContext[AgentDepsT]], Awaitable[str | None]]: ...

    @overload
    def instructions(self, func: Callable[[], str | None], /) -> Callable[[], str | None]: ...

    @overload
    def instructions(self, func: Callable[[], Awaitable[str | None]], /) -> Callable[[], Awaitable[str | None]]: ...

    @overload
    def instructions(
        self, /

48:from ..capabilities import AbstractCapability, CombinedCapability
49:from ..capabilities.builtin_tool import BuiltinTool as BuiltinToolCap
50:from ..capabilities.history_processor import HistoryProcessor as HistoryProcessorCap
229:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
261:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
291:        capabilities: Sequence[AbstractCapability[AgentDepsT]] | None = None,
369:            capabilities: Optional list of [capabilities](https://ai.pydantic.dev/capabilities/) to configure the agent with.
370:                Custom capabilities can be created by subclassing
371:                [`AbstractCapability`][pydantic_ai.capabilities.AbstractCapability].
384:        capabilities = list(capabilities or [])
386:            capabilities.append(HistoryProcessorCap(history_processor))
388:

…[738 chars omitted]…
s.append(capability)
663:        if capabilities:
664:            all_capabilities.extend(capabilities)
694:            capabilities=all_capabilities,
728:        capabilities: Sequence[AbstractCapability[Any]] | None = None,
763:        capabilities: Sequence[AbstractCapability[Any]] | None = None,
1092:            # Capability settings (e.g. from Thinking, ModelSettings capabilities), cached at init

5:from .abstract import AbstractToolset, ToolsetTool
16:AgentToolset = Union[AbstractToolset[AgentDepsT], ToolsetFunc[AgentDepsT]]  # noqa: UP007 — Union needed at runtime (no future annotations)
20:    'AbstractToolset',
21:    'AgentToolset',
from typing import Union

from .._run_context import AgentDepsT
from ._dynamic import ToolsetFunc
from .abstract import AbstractToolset, ToolsetTool
from .approval_required import ApprovalRequiredToolset
from .combined import CombinedToolset
from .external import DeferredToolset, ExternalToolset  # pyright: ignore[reportDeprecated]

from .filtered import FilteredToolset
from .function import FunctionToolset
from .prefixed import PrefixedToolset
from .prepared import PreparedToolset
from .renamed import RenamedToolset
from .wrapper import WrapperToolset

AgentToolset = Union[AbstractToolset[AgentDepsT], ToolsetFunc[AgentDepsT]]  # noqa: UP007 — U
…[55 chars omitted]…
lset or a factory function that creates a toolset from a run context."""

__all__ = (
    'AbstractToolset',
    'AgentToolset',
    'ToolsetFunc',
    'ToolsetTool',
    'CombinedToolset',
    'ExternalToolset',
    'DeferredToolset',
    'FilteredToolset',
    'FunctionToolset',
    'PrefixedToolset',
    'RenamedToolset',
    'PreparedToolset',
    'WrapperToolset',
    'ApprovalRequiredToolset',
)

class OutputToolset(AbstractToolset[AgentDepsT]):
    """A toolset that contains contains output tools for agent output types."""

    _tool_defs: list[ToolDefinition]
    """The tool definitions for the output tools in this toolset."""
    processors: dict[str, ObjectOutputProcessor[Any]]
    """The processors for the output tools in this toolset."""
    max_retries: int
    output_validators: list[OutputValidator[AgentDepsT, Any]]

    @classmethod
    def build(

        cls,
        outputs: list[OutputTypeOrFunction[OutputDataT] | ToolOutput[OutputDataT]],
        name: str | None = None,
        description: str | None = None,
        strict: bool | None = None,
    ) -> Self | None:
        if len(outputs) == 0:
            return None

        processors: dict[str, ObjectOutputProcessor[Any]] = {}
        tool_defs: list[ToolDefinition] = []

        default_name = name or DEFA
…[2807 chars omitted]…
, ctx: RunContext[AgentDepsT]) -> dict[str, ToolsetTool[AgentDepsT]]:
        return {
            tool_def.name: ToolsetTool(
                toolset=self,
                tool_def=tool_def,
                max_retries=self.max_retries,
                args_validator=self.processors[tool_def.name].validator,
            )
            for tool_def in self._tool_defs
        }

    async def call_tool(

call, 'Output tool not used - a final result was already processed.', output_parts, args_valid=None
            ):
                yield event
        # Early strategy is chosen and final result is not yet set
        # Or exhaustive strategy is chosen
        else:
            # Validate and execute the output tool call — unlike deferred tools,
            # output tools track retries and can be skipped if a final_result exists.
            try:
                validated = await tool_manager.validate_tool_call(call)
            except exceptions.UnexpectedModelBehavior as e:
                # If we already have a valid final result, don't fail the entire run

                # This allows exhaustive strategy to complete successfully when at least one output tool is valid
                if final_result:
                    # If output tool fails when we already have a 
…[2171 chars omitted]…
tries(ctx.deps.max_result_retries, error=e)
                yield _messages.FunctionToolCallEvent(call, args_valid=True)
                output_parts.append(e.tool_retry)
                yield _messages.FunctionToolResultEvent(e.tool_retry)
                continue

            part = _messages.ToolReturnPart(
                tool_name=call.tool_name,
                content='Final result processed.',

ok
.........                                                                [100%]
9 passed, 207 deselected in 2.87s

tools ['b_a_tool']
tools ['b_a_tool']
done
sunny [(0, 5), (1, 5), (2, 5), (3, 5), (4, 5), (5, 5)]

result a
{'a': [(0, 3), (1, 3), (2, 3), (3, 3)]}
tools ['b_a_tool'] outputs []
tools ['b_a_tool'] outputs []
done

........................................................................ [ 33%]
.................s...................................................... [ 66%]
................sss...s................................................. [100%]
211 passed, 5 skipped in 4.07s

..............................x..xx..................................... [ 33%]
...........ss........................................................... [ 66%]
.......................................................................  [100%]
210 passed, 2 skipped, 3 xfailed in 6.40s

test_capabilities.py
........................................................................ [ 53%]
...............................................................          [100%]
135 passed in 3.21s

85:    def increment_retries(
844:                        ctx.state.increment_retries(ctx.deps.max_result_retries)
916:                    ctx.state.increment_retries(ctx.deps.max_result_retries, error=e)
1241:        ctx.state.increment_retries(ctx.deps.max_result_retries)

grep: pydantic_ai_slim/pydantic_ai/capabilities/__pycache__/abstract.cpython-312.pyc: binary file matches
grep: pydantic_ai_slim/pydantic_ai/capabilities/__pycache__/combined.cpython-312.pyc: binary file matches
docs/durable_execution/temporal.md:71:At the time of wrapping, the agent's [model](../models/overview.md) and [toolsets](../toolsets.md) (including function tools registered on the agent and MCP servers) are frozen, activities are dynamically created for each, and the original model and toolsets are wrapped to call on the worker to execute the corresponding activities instead of directly performing the actions inside the workflow. The original agent can still be used as normal outside the Temporal workflow, but any changes to its model or toolsets after wrapping will not be reflected in the durable agent.
docs/durable_execution/temporal.md:160:When `TemporalAgent` dynamically 
…[2845 chars omitted]…
t.py:96:    def get_wrapper_toolset(self, toolset: AbstractToolset[AgentDepsT]) -> AbstractToolset[AgentDepsT] | None:
pydantic_ai_slim/pydantic_ai/capabilities/combined.py:82:    def get_wrapper_toolset(self, toolset: AbstractToolset[AgentDepsT]) -> AbstractToolset[AgentDepsT] | None:
pydantic_ai_slim/pydantic_ai/capabilities/combined.py:85:            result = capability.get_wrapper_toolset(toolset)

test_agent_output_schemas.py
test_native_output_schema.py
ERROR: file or directory not found: tests/test_output.py


no tests ran in 2.07s

],
    )
    async def test_nested_output_type_description(output_type: type, expected_schema: dict[str, object]):
        agent: Agent[None, str] = Agent('test', output_type=output_type)
>       assert agent.output_json_schema() == expected_schema
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^

tests/test_agent_output_schemas.py:704: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = {'$defs': {'DCNested': {'properties': {'category': {'default': 'all', 'title': 'Category', 'type': 'string'}}, 'title'...s': {'filters': {'$ref': '#/$defs/DCNested'}}, 'required': ['filters'], 'title': 'DCWithNestedField', 'type': 'object'}
other = {'$defs': {'DCNested': {'description': 'Nested filter criteria.', 'properties': {'category': {'default': 'all', 'title...'Output with nested dataclass.', 'properties': {'filters': {'$ref': '#/$defs/D

…[1135 chars omitted]…
ts/_inline_snapshot.py:52: UserWarning
=========================== short test summary info ============================
FAILED tests/test_agent_output_schemas.py::test_deferred_output_json_schema
FAILED tests/test_agent_output_schemas.py::test_output_type_description[dataclass]
FAILED tests/test_agent_output_schemas.py::test_nested_output_type_description[dataclass_nested]
3 failed, 17 passed in 2.87s

pydantic_ai_slim/pydantic_ai/_agent_graph.py       | 30 +++++++++++++++++-----
 pydantic_ai_slim/pydantic_ai/_output.py            | 16 ++++++++++--
 pydantic_ai_slim/pydantic_ai/agent/__init__.py     |  8 ++++++
 .../pydantic_ai/capabilities/abstract.py           | 12 ++++++++-
 .../pydantic_ai/capabilities/combined.py           |  9 +++++++
 5 files changed, 65 insertions(+), 10 deletions(-)
diff --git a/pydantic_ai_slim/pydantic_ai/_agent_graph.py b/pydantic_ai_slim/pydantic_ai/_agent_graph.py
index ae9801c..c39dc4d 100644
--- a/pydantic_ai_slim/pydantic_ai/_agent_graph.py
+++ b/pydantic_ai_slim/pydantic_ai/_agent_graph.py
@@ -1096,6 +1096,25 @@ def _emit_skipped_output_tool(
     yield _messages.FunctionToolResultEvent(part)

 
 
+def _output_tool_max_retries(
+    tool_manager: ToolManager[DepsT],
+    call: _messages.ToolCallPart,
+    ctx: GraphRunContext[GraphAgentState, Graph
…[8629 chars omitted]…
lity in self.capabilities:
+            result = capability.get_wrapper_toolset(toolset)
+            if result is not None:
+                toolset = result
+                wrapped = result
+        return wrapped
+
     def get_builtin_tools(self) -> Sequence[AgentBuiltinTool[AgentDepsT]]:
         builtin_tools: list[AgentBuiltinTool[AgentDepsT]] = []
         for capability in self.capabilities:

Traceback (most recent call last):
  File "/tmp/repro_stream.py", line 27, in <module>
    import asyncio; asyncio.run(main())
                    ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
    return future.result()

           ^^^^^^^^^^^^^^^
  File "/tmp/repro_stream.py", line 25, in main
    async with agent.run_stream('hi') as r:
               ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/contextlib.py", line 210, in __aenter__
    return await anext(self.gen)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydanti
…[1549 chars omitted]…
sr/local/lib/python3.12/contextlib.py", line 210, in __aenter__
    return await anext(self.gen)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/models/function.py", line 178, in request_stream
    assert self.stream_function is not None, (
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AssertionError: FunctionModel must receive a `stream_function` to support streamed requests

Traceback (most recent call last):
  File "/tmp/repro_stream.py", line 34, in <module>
    import asyncio; asyncio.run(main())
                    ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/runners.py", line 195, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/asyncio/base_events.py", line 691, in run_until_complete
    return future.result()

           ^^^^^^^^^^^^^^^
  File "/tmp/repro_stream.py", line 32, in main
    async with agent.run_stream('hi') as r:
               ^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/contextlib.py", line 210, in __aenter__
    return await anext(self.gen)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydanti
…[1664 chars omitted]…
^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/models/function.py", line 184, in request_stream
    first = await response_stream.peek()
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/_utils.py", line 353, in peek
    self._source_iter = aiter(self._source)
                        ^^^^^^^^^^^^^^^^^^^
TypeError: 'generator' object is not an async iterable

stream tools ['b_a_tool']
{"x": 1}

UnexpectedModelBehavior Exceeded maximum retries (1) for output validation
[(0, 1), (1, 1)]

yield stream

        # Run the stream to completion if it was not finished:
        async for _event in stream:
            pass

    async def _run_stream(  # noqa: C901
        self, ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, NodeRunEndT]]
    ) -> AsyncIterator[_messages.HandleResponseEvent]:
        if self._events_iterator is None:
            # Ensure that the stream is only run once

            output_schema = ctx.deps.output_schema

            async def _run_stream() -> AsyncIterator[_messages.HandleResponseEvent]:  # noqa: C901
                is_empty = not self.model_response.parts
                is_thinking_only = not is_empty and all(
                    isinstance(p, _messages.ThinkingPart) for p in self.model_response.parts
                )

                if is_empty or is_thinking_only:
                    # No actionable output was retu
…[7492 chars omitted]…
estNode[DepsT, NodeRunEndT](
                        _messages.ModelRequest(parts=[e.tool_retry], instructions=instructions)
                    )

            self._events_iterator = _run_stream()

        async for event in self._events_iterator:
            yield event

    async def _handle_tool_calls(
        self,
        ctx: GraphRunContext[GraphAgentState, GraphAgentDeps[DepsT, NodeRunEndT]],

grep: tests/__pycache__/test_agent.cpython-312-pytest-9.2.0.pyc: binary file matches
tests/test_agent.py:30:    IncompleteToolCall,
tests/test_agent.py:3188:        IncompleteToolCall,
tests/test_agent.py:3194:        IncompleteToolCall,
tests/test_exceptions.py:18:    IncompleteToolCall,
tests/test_exceptions.py:41:        lambda: IncompleteToolCall('test'),
tests/test_exceptions.py:54:        'IncompleteToolCall',
tests/test_exceptions.py:97:        (lambda: IncompleteToolCall('incomplete'), {'message': 'incomplete', 'body': None}),
tests/test_exceptions.py:115:        'IncompleteToolCall',

UnexpectedModelBehavior Exceeded maximum retries (3) for output validation

M pydantic_ai_slim/pydantic_ai/_agent_graph.py
 M pydantic_ai_slim/pydantic_ai/_output.py
 M pydantic_ai_slim/pydantic_ai/agent/__init__.py
 M pydantic_ai_slim/pydantic_ai/capabilities/abstract.py
 M pydantic_ai_slim/pydantic_ai/capabilities/combined.py

........................................................................ [ 12%]
........................................................................ [ 25%]
........................................................................ [ 38%]
........s............................................................... [ 50%]
.......sss...s.......................................................... [ 63%]
.....................x..xx.............................................. [ 76%]
..ss.................................................................... [ 89%]
..............................................................           [100%]
556 passed, 7 skipped, 3 xfailed in 9.77s

grep: pydantic_ai_slim/pydantic_ai/agent/__pycache__/__init__.cpython-312.pyc: binary file matches
grep: pydantic_ai_slim/pydantic_ai/__pycache__/_output.cpython-312.pyc: binary file matches
pydantic_ai_slim/pydantic_ai/agent/__init__.py:44:from .._output import OutputToolset
pydantic_ai_slim/pydantic_ai/agent/__init__.py:184:    _output_toolset: OutputToolset[AgentDepsT] | None = dataclasses.field(repr=False)
pydantic_ai_slim/pydantic_ai/_output.py:215:    toolset: OutputToolset[Any] | None = None
pydantic_ai_slim/pydantic_ai/_output.py:321:        toolset = OutputToolset.build(tool_outputs + other_outputs, name=name, description=description, strict=strict)
pydantic_ai_slim/pydantic_ai/_output.py:388:        toolset: OutputToolset[Any] | None,
pydantic_ai_slim/pydantic_ai/_output.py:493:        toolset: OutputToolset[Any] | None,
pydantic_ai_slim/pydantic_ai/_output.py:859:class OutputToolset(AbstractToolset[AgentDepsT]):

103 293                         f'Model token limit ({self.last_max_tokens or "provider default"}) exceeded while generating a tool call, resulting in incomplete arguments. Increase the `max_tokens` model setting, or simplify the prompt to result in a shorter response that will fit within the limit.'
248 122                         # If there's no new prompt or instructions, skip ModelRequestNode and go directly to CallToolsNode
346 132         """Reevaluate any `SystemPromptPart` with dynamic_ref in the provided messages by running the associated runner function."""
689 139         # `ctx.state.message_history` is the same list used by `capture_run_messages`, so we should replace its contents, not the reference
697 150         # but don't store it in the message history on state. This is just for the benefit of model classes that want clear user/assistant boundaries.
806 272         
…[471 chars omitted]…
ready included in the run because results were provided)
1350 193                 'A deferred tool call was present, but `DeferredToolRequests` is not among output types. To resolve this, add `DeferredToolRequests` to the list of output types for this agent.'
1446 135                         if event := await handle_call_or_result(coro_or_task=task, index=index):  # pyright: ignore[reportArgumentType]

tools ['a_tool'] outputs ['final_result']
tools ['a_tool'] outputs ['final_result']
Traceback (most recent call last):
  File "/repo/pydantic_ai_slim/pydantic_ai/_tool_manager.py", line 339, in validate_tool_call
    validated_args = await self._run_validate_hooks(call, tool, ctx, allow_partial=allow_partial)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/_tool_manager.py", line 252, in _run_validate_hooks
    validated_args = await cap.wrap_tool_validate(ctx, call=call, args=raw_args, handler=do_validate)
                     ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/repo/pydantic_ai_slim/pydantic_ai/capabilities/combined.py", line 234, in wrap_tool_validate
    return await chain(args)
           ^^^^^^^^^^^^^^^^^

  File "/repo/pydantic_ai_sli
…[4718 chars omitted]…
agent_graph.py", line 1177, in process_tool_calls
    _raise_output_retry_exceeded(_output_tool_max_retries(tool_manager, call, ctx), e)
  File "/repo/pydantic_ai_slim/pydantic_ai/_agent_graph.py", line 1115, in _raise_output_retry_exceeded
    raise exceptions.UnexpectedModelBehavior(message) from error
pydantic_ai.exceptions.UnexpectedModelBehavior: Exceeded maximum retries (1) for output validation

from __future__ import annotations

import asyncio
from collections.abc import AsyncIterable, AsyncIterator
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import pytest

from pydantic_ai._run_context import RunContext
from pydantic_ai.agent import Agent

from pydantic_ai.capabilities import (
    CAPABILITY_TYPES,
    BuiltinTool,
    Instructions,
    ModelSettings,
    Thinking,
    Toolset,
    WebSearch,
)
from pydantic_ai.capabilities.abstract import AbstractCapability, ModelRequestContext
from pydantic_ai.capabilities.combined import CombinedCapability
from pydantic_ai.exceptions import SkipModelRequest, SkipToolExecution, SkipToolValidation
from pydantic_ai.messages import (
    AgentStreamEvent,
    ModelMessage,
    ModelRequest,
    ModelResponse,
    TextPart,
    ToolCallPart,
    ToolReturnPart,
)
from pydantic_ai.models.function 
…[4619 chars omitted]…
s ShortCircuitRunCap(AbstractCapability[Any]):
1267:        class ErrorRecoveryCap(AbstractCapability[Any]):
1295:        class ErrorRecoveryCap(AbstractCapability[Any]):
1343:        class ModifyResponseCap(AbstractCapability[Any]):
1359:        class WrapModifyCap(AbstractCapability[Any]):
1372:        class SkipCap(AbstractCapability[Any]):
1402:        class ModifyArgsCap(AbstractCapability[Any]):

['ax_tool']
['ax_tool']
done

x=1 [(0, 5), (1, 5), (2, 5), (3, 5), (4, 5), (5, 5)]

sunny [(0, 4), (1, 4), (2, 4), (3, 4), (4, 4)]

diff --git a/pydantic_ai_slim/pydantic_ai/_output.py b/pydantic_ai_slim/pydantic_ai/_output.py
index 87f93c4..09636a6 100644
--- a/pydantic_ai_slim/pydantic_ai/_output.py
+++ b/pydantic_ai_slim/pydantic_ai/_output.py
@@ -864,6 +864,8 @@ class OutputToolset(AbstractToolset[AgentDepsT]):
     processors: dict[str, ObjectOutputProcessor[Any]]
     """The processors for the output tools in this toolset."""
     max_retries: int
+    tool_max_retries: dict[str, int | None]
+    """Explicit per-output-tool retry limits, keyed by tool name."""
     output_validators: list[OutputValidator[AgentDepsT, Any]]
 

     @classmethod
@@ -879,6 +881,7 @@ class OutputToolset(AbstractToolset[AgentDepsT]):
 
         processors: dict[str, ObjectOutputProcessor[Any]] = {}
         tool_defs: list[ToolDefinition] = []
+        tool_max_retries: dict[str, int | None] = {}
 
         default_name = name or D
…[1893 chars omitted]…
            max_retries=self.max_retries,
+                max_retries=(
+                    self.tool_max_retries[tool_def.name]
+                    if self.tool_max_retries.get(tool_def.name) is not None
+                    else self.max_retries
+                ),
                 args_validator=self.processors[tool_def.name].validator,
             )
             for tool_def in self._tool_defs

pydantic_ai_slim/pydantic_ai/agent/__init__.py:432:            self._output_toolset.max_retries = self._max_result_retries
pydantic_ai_slim/pydantic_ai/agent/__init__.py:1022:                output_toolset.max_retries = self._max_result_retries
pydantic_ai_slim/pydantic_ai/durable_exec/prefect/_cache_policies.py:40:                'max_retries': value.max_retries,
pydantic_ai_slim/pydantic_ai/durable_exec/temporal/_run_context.py:53:            'max_retries': ctx.max_retries,
pydantic_ai_slim/pydantic_ai/_output.py:901:                max_retries = output.max_retries

UnexpectedModelBehavior Exceeded maximum retries (0) for output validation
[(0, 0)]

.......sss...s.......................................................... [ 63%]
.....................x..xx.............................................. [ 76%]
..ss.................................................................... [ 89%]
..............................................................           [100%]
556 passed, 7 skipped, 3 xfailed in 9.36s

submitted.