Agent Tool-Call Budgets: Bound Delegation and Latency

[!NOTE] TL;DR Agent tool-call budgets need a shared usage counter and one deadline across delegated runs. A tool timeout alone does not stop repeated model requests; a request cap alone does not bound elapsed time. The BND reference implementation below combines both, while keeping failed calls and unknown costs visible.

An unbounded delegate

My first agent budget ignored the delegated agent. I had capped the retrieval tool at 1 second, yet the model could keep asking; my kitchen timer had better governance. Agent tool-call budgets begin where that convenient mistake ends.

Here I present BND, a runnable reference design rather than a claim about a customer deployment. It has a coordinator, a researcher and a deliberately local evidence fixture. I chose 6 seconds for the whole request and 4 model requests as design inputs, not measured production targets. Those numbers make the failure modes legible before you substitute an HTTP or MCP tool.

One operational question

My main question was how to keep a delegated agent inside one wall-clock deadline without mistaking model usage for all downstream work. That distinction matters when you own the user-facing API. A request can exceed its time allowance while consuming surprisingly few model calls, or consume its call allowance while every tool is fast.

The objectives are simple: bound elapsed time across the hand-off; count model requests across agents; distinguish successful tool executions from attempts. I won't call any of those a cost guarantee. Model billing, downstream service charges and cancelled work have different accounting surfaces.

The false comfort of local limits

One spontaneously imagines that tool_calls_limit=2 means no more than 2 attempts at a tool. The Pydantic AI usage reference says otherwise: RunUsage.tool_calls counts successful executions, while the default model request_limit is 50. Its request limit is checked before a model request; token limits are generally checked after the response. A token ceiling can therefore reject an already billed response.

The advanced tools documentation also excludes structured-output tools from that successful-call count. Tool failures need their own monitoring. Although successful-call limits prevent one kind of loop, they cannot tell you how often a flaky endpoint was tried. And a tool_timeout can return a retry prompt to the model, increasing work instead of ending the run.

Delegation adds another accounting error. If the coordinator calls a researcher without passing its RunUsage, you lose the combined count. The Pydantic AI multi-agent guide passes usage=ctx.usage for precisely this reason. I pass the live ctx.usage_limits too, so the child checks the same ceiling rather than starting with an unrelated configuration. Do not treat this shared in-process object as a distributed quota for separate workers.

BND architecture

flowchart LR
    A[API request] --> B[Single monotonic deadline]
    B --> C[Coordinator agent]
    C --> D[Delegate tool]
    D --> E[Researcher agent]
    E --> F[Lookup tool with remaining time]
    F --> E
    E --> D
    D --> C
    C --> G[Typed answer or deadline failure]
    C -. shared RunUsage and UsageLimits .-> E

The diagram contains one clock, not a new clock for each agent. Picture a wall clock above a workbench where the request moves between stations. Each station reads the time remaining on that same clock; handing over the work never resets it.

I would place this code behind the API boundary in an AgentCore Runtime container when the deployment calls for it. An AWS session lifetime is not an API deadline: AgentCore lifecycle documentation gives default idle and instance lifetimes of 900 and 28800 seconds for the documented runtime configuration. Neither number promises a user an answer in 6 seconds. If you deploy the example elsewhere, the same application deadline still belongs at the request boundary.

The BND implementation

The example targets Python 3.11 or newer with pydantic-ai, pydantic and logfire installed, plus an OPENAI_API_KEY for the model in the code. I use the openai:gpt-5.2 identifier shown in Pydantic AI's examples; substitute a supported model if your account uses another provider. Save this as bnd.py and run python bnd.py. The fixture quotes one documented rule, not a live retrieval result.

import asyncio
from time import monotonic
from typing import Self
import logfire
from pydantic import BaseModel, ConfigDict, Field, computed_field, model_validator
from pydantic_ai import Agent, RunContext, RunUsage, UsageLimits
from pydantic_ai.models.instrumented import InstrumentationSettings
class BndConfig(BaseModel):
    model_config = ConfigDict(strict=True)
    wall_seconds: float = Field(6.0, gt=0.0, description="Deadline for the whole run in seconds.")
    tool_seconds: float = Field(1.0, gt=0.0, description="Maximum seconds for a lookup attempt.")
    max_requests: int = Field(4, ge=1, description="Shared model request ceiling.")
    max_successful_tools: int = Field(2, ge=1, description="Shared successful tool ceiling.")
    @model_validator(mode="after")
    def check_clock(self) -> Self:
        if self.tool_seconds >= self.wall_seconds:
            raise ValueError("The tool window must be shorter than the run deadline")
        return self
    @computed_field
    @property
    def wall_milliseconds(self) -> int:
        return int(self.wall_seconds * 1000)
class Deps(BaseModel):
    model_config = ConfigDict(strict=True)
    config: BndConfig = Field(..., description="Immutable-by-convention run policy.")
    deadline: float = Field(..., gt=0.0, description="Monotonic absolute deadline.")
class Evidence(BaseModel):
    source_url: str = Field(..., description="URL from the evidence fixture.")
    passage: str = Field(..., description="Short sourced passage.")
class Answer(BaseModel):
    answer: str = Field(..., description="User-facing answer.")
    source_url: str = Field(..., description="Evidence URL to check.")
logfire.configure(send_to_logfire="if-token-present")
# Trace structure without exporting prompts or tool payloads.
Agent.instrument_all(InstrumentationSettings(include_content=False))
researcher: Agent[Deps, Evidence] = Agent("openai:gpt-5.2", name="bnd_researcher", deps_type=Deps, output_type=Evidence, retries=0, instructions="Call lookup once. Return only the passage and URL it supplies.")
coordinator: Agent[Deps, Answer] = Agent("openai:gpt-5.2", name="bnd_coordinator", deps_type=Deps, output_type=Answer, retries=0, instructions="Call research once. Answer from its evidence and copy its source URL.")
@researcher.tool
async def lookup(ctx: RunContext[Deps], query: str) -> Evidence:
    """Fetch the local demonstration fixture; replace this body with I/O."""
    if not query.strip():
        raise ValueError("A nonempty query is required")
    remaining: float = max(0.0, ctx.deps.deadline - monotonic())
    # Never allocate more time than the caller has left.
    async with asyncio.timeout(min(ctx.deps.config.tool_seconds, remaining)):
        await asyncio.sleep(0.05)
        return Evidence(source_url="https://pydantic.dev/docs/ai/api/pydantic-ai/usage/", passage="The request limit is checked before each model request.")
@coordinator.tool
async def research(ctx: RunContext[Deps], topic: str) -> Evidence:
    """Delegate using the parent's live counters and ceiling."""
    result = await researcher.run(topic, deps=ctx.deps, usage=ctx.usage, usage_limits=ctx.usage_limits)
    return result.output
async def main() -> None:
    config = BndConfig()
    usage = RunUsage()
    deps = Deps(config=config, deadline=monotonic() + config.wall_seconds)
    limits = UsageLimits(request_limit=config.max_requests, tool_calls_limit=config.max_successful_tools)
    # A single absolute deadline covers model turns and delegated tools.
    async with asyncio.timeout_at(deps.deadline):
        result = await coordinator.run("When does Pydantic AI check its model request limit?", deps=deps, usage=usage, usage_limits=limits)
    print(result.output.model_dump_json())
    print(f"requests={usage.requests} successful_tools={usage.tool_calls} estimated_cost={usage.cost}")
if __name__ == "__main__":
    asyncio.run(main())

I made retries=0 explicit so this small example doesn't conceal extra model turns behind validation retries. A production adapter can allow a retry only if the remaining deadline admits one. The computed 6000 milliseconds is just a unit conversion from the chosen 6-second policy, not a latency observation.

The local fixture has a 0.05-second artificial wait. It checks the plumbing; it does not simulate a provider's tail latency or prove that a downstream cancellation took effect. Replace its body with an async client whose own timeout receives remaining. For a state-changing tool, pass an idempotency key to the service before introducing automatic retries. Otherwise the timeout may cancel your coroutine after the remote side has already acted.

Reference measurements and invariants

I cannot honestly present a p95 from a fixture and a model call I haven't run on your traffic. The table separates numbers reported in a public example from the limits encoded above. This is a modest result, but a useful one.

The same documented example reports an estimated 0.00051200 USD of model cost. I would not transplant that figure to another model or count it as the price of an AgentCore invocation. The model can return usage.cost=None when pricing cannot be determined; a configured monetary ceiling cannot price an unknown response.

My named hypothesis is the borrowed clock: delegation should inherit a deadline rather than receive a fresh duration. The code implements that proposition with asyncio.timeout_at and an absolute deadline in typed dependencies. A failed lookup can still consume time and model requests, even when it never increments successful tool calls. That's why I inspect separate traces for the run, model requests and tools. Pydantic AI's Logfire guide documents those spans and the include_content=False setting that omits prompt and tool content from telemetry. A beautiful dashboard that records every prompt is not my idea of discretion.

Limits in live traffic

Although BND closes the accidental reset at an in-process hand-off, it doesn't enforce a global tenant quota. A second worker starts with another RunUsage. Neither asyncio.timeout_at nor an in-process tool_timeout can undo an HTTP write that has already committed. Even the fixture's apparent success depends on the model actually calling the tools; the output schema does not attest that its source was read.

This could easily be tested with controlled fault injection. I would delay the lookup beyond 1 second, make its replacement endpoint return an error, and compare elapsed time with counts of attempted and successful calls. Then I would add a real model and gather p50 and p95 from representative requests, including cold starts. Until those runs exist, the production latency distribution and per-invocation cost remain [to verify]. No napkin should be promoted to an SLO.

For an API rollout, I would make the trace carry a safe request identifier and record each tool attempt separately from RunUsage.tool_calls. I'd also check how provider retries and client disconnects behave at the boundary. If you want a second pair of eyes on that boundary, contact me; my agent systems work is built around measured operations rather than a tidy demo.

The surviving principle

More generally, an agent's autonomy is only as useful as the boundary that can stop it. The clock belongs to the caller, while evidence and usage must survive the hand-off. Even my kitchen timer understood that part.