Better handling when max tokens limit results in incomplete tool call arguments

[erhuve]

Initial Checks

• I confirm that I'm using the latest version of Pydantic AI
• I confirm that I searched for my issue in https://github.com/pydantic/pydantic-ai/issues before opening this issue

Description

I encounter this only with Anthropic models.

From what I gather, it's pretty easy for me to hit the output token limit for some tool calls that I make for file editing, and it seems that this results in completely dropping whichever arg was being written, and leaving an incomplete tool args (even missing the closing brace)

There's a retry prompt by default, but it's ultimately not helpful because I suppose the agent doesn't even know it's input was actually cut off.

My example code calls with and without retries to show what seems to be happening to the tool args

*Saw 1.0.16 included changes around anthropic token usage, so upgraded and still erroring this way

---With Retry---
[Request] Starting part 0: ToolCallPart(tool_name='write_long_text', tool_call_id='toolu_012yy98oVc8ZuDPeY4uT75pd')
[Error] EOF while parsing an object at line 1 column 30
Output: 30
{"title": "example with retry"

Example Code

from pydantic_ai import Agent
from pydantic_ai.messages import (
    FinalResultEvent,
    FunctionToolCallEvent,
    FunctionToolResultEvent,
    PartDeltaEvent,
    PartStartEvent,
    TextPartDelta,
    ThinkingPartDelta,
    ToolCallPartDelta,
)
import asyncio

SONNET_4 = "claude-sonnet-4-5-20250929"

def write_long_text(title: str, text: str):
    return f"Wrote {len(text)} chars for {title}"

async def main(agent):
    prompt = (
        "Call the write_long_text tool with a single argument named 'text' containing approximately "
        "300 lines of physics simulation code. Do not return any assistant text; only make the tool call."
    )

    output = []

    try:
        async with agent.iter(user_prompt=prompt) as agent_run:
            async for node in agent_run:
                if Agent.is_model_request_node(node):
                    n = node  # type: ignore[assignment]
                    async with n.stream(agent_run.ctx) as request_stream:
                        async for chunk in request_stream:
                            if isinstance(chunk, PartStartEvent):
                                print(f'[Request] Starting part {chunk.index}: {chunk.part!r}')
                            elif isinstance(chunk, PartDeltaEvent):
                                if isinstance(chunk.delta, TextPartDelta):
                                    print(f'[Request] Part {chunk.index} text delta: {chunk.delta.content_delta!r}')
                                elif isinstance(chunk.delta, ThinkingPartDelta):
                                    print(f'[Request] Part {chunk.index} thinking delta: {chunk.delta.content_delta!r}')
                                elif isinstance(chunk.delta, ToolCallPartDelta):
                                    # print(f'[Request] Part {chunk.index} args delta: {chunk.delta.args_delta}')
                                    output.append(chunk.delta.args_delta)
                            elif isinstance(chunk, FunctionToolCallEvent):
                                print(
                                    f'[Tools] The LLM calls tool={chunk.part.tool_name!r} with args={chunk.part.args} (tool_call_id={chunk.part.tool_call_id!r})'
                                )
                            elif isinstance(chunk, FunctionToolResultEvent):
                                print(f'[Tools] Tool call {chunk.tool_call_id!r} returned => {chunk.result.content}')
                            elif isinstance(chunk, FinalResultEvent):
                                finish_reason = getattr(chunk, "finish_reason", None)
                                if finish_reason is not None:
                                    print(f"[Result] Finish reason: {finish_reason}")
                                print(f'[Result] The model starting producing a final result (tool_name={chunk.tool_name})')
    except Exception as e:
        print(f"[Error] {e}")

    str_output = "".join(output)
    print(f"Output: {len(str_output)}")
    print(str_output)

if __name__ == "__main__":
    agent = Agent(
        model=SONNET_4,
        output_type=str,
        tools=[write_long_text],
        instructions=(
            "You can use tools. When the user asks to write long text, call the write_long_text tool "
            "with the full text as the 'text' argument and 'example with retry' as the title."
        ),
    )
    print("---With Retry---")
    asyncio.run(main(agent))

    agent_no_retry = Agent(
        model=SONNET_4,
        output_type=str,
        tools=[write_long_text],
        instructions=(
            "You can use tools. When the user asks to write long text, call the write_long_text tool "
            "with the full text as the 'text' argument and 'example without retry' as the title."
        ),
        retries=0
    )
    print("---Without retry---")
    asyncio.run(main(agent_no_retry))

"""
---With Retry---
[Request] Starting part 0: ToolCallPart(tool_name='write_long_text', tool_call_id='toolu_01Pxsax9GPLwJJzDu5YhESoc')
[Error] Exceeded maximum retries (1) for output validation
Output: 14343
{"title": "example with retry", "text": "import [...i skip here, just important to note that this is a complete args json]"}
---Without retry---
[Request] Starting part 0: ToolCallPart(tool_name='write_long_text', tool_call_id='toolu_015hU11V6Xd9wrs4Su3GiU1X')
[Error] Tool 'write_long_text' exceeded max retries count of 0
Output: 33
{"title": "example without retry"

---^ Observe that the offending long arg gets dropped completely

"""

Python, Pydantic AI & LLM client version

python 3.12.6
pydantic-ai 1.0.15 -> 1.0.16

[DouweM]

@erhuve Have you tried increasing the max_tokens model setting? The default for Anthropic is 4096, see #2553 for more context.

There's a retry prompt by default, but it's ultimately not helpful because I suppose the agent doesn't even know it's input was actually cut off.

The incomplete tool call is sent back to the model along with the retry message, so it would have its original attempt in the context.

---

Besides increasing the max tokens, what kind of solution do you have in mind for the case where a model hits its token limit while in the middle of outputting a tool call?

[erhuve]

@DouweM I had tried increasing it, but I just set the limit even higher and it works okay, just really unexpectedly high token usage from Anthropic!

The incomplete tool call is sent back to the model along with the retry message, so it would have its original attempt in the context.

I didn't look too deeply into it but was finding that the agent would basically attempt the same thing, so even with its original attempt I don't think it understood why it really got the error.

Which does lead into the question of what kind of solution there should be for this. My layers to this are

• Repair any broken json in args_as_dict

def args_as_dict(self) -> dict[str, Any]:
        """Return the arguments as a Python dictionary.

        This is just for convenience with models that require dicts as input.
        """
        if not self.args:
            return {}
        if isinstance(self.args, dict):
            return self.args
        args = pydantic_core.from_json(self.args) # Wrap in try except or check beforehand starts and ends with curly braces { }
        assert isinstance(args, dict), 'args should be a dict'
        return cast(dict[str, Any], args)

This would at least make a valid json but still invalid args, and no context to the agent of what it needs to do to alleviate it. So,

• Enhance retry prompt- are we able to check stop_reason for length so that we can instead return some message in the prompt "Args exceed token limit, try smaller input" ?

Happy to make the PR for this if so

[DouweM]

I didn't look too deeply into it but was finding that the agent would basically attempt the same thing, so even with its original attempt I don't think it understood why it really got the error.

@erhuve That makes sense, it wouldn't know its own token limit so it'd just try the same thing again.

• Enhance retry prompt- are we able to check stop_reason for length so that we can instead return some message in the prompt "Args exceed token limit, try smaller input" ?

I agree we should handle ModelResponse.finish_reason=='length' some special way. I'm not sure if it's reasonable to ask the model to be more concise though; maybe the tool needs longer arguments, or it was asked (by the user or the instructions) to generate a 10-page story, and an automatic message telling it to be more concise would break that.

I think the right solution would be for the developer to get a clearer error message, and then either increase the max_tokens settings, or modify the prompt/tool definitions to need fewer tokens. So maybe the solution is to have a special exception if model_response.finish_reason == 'length' and parts[-1] is an incomplete ToolCallPart?

[erhuve]

Sounds reasonable, will work on it