Skip to main content

delegation.delegate

Agent delegation — cross-agent usage tracking and OpenTelemetry propagation.

Aligned with pydantic-ai's recommended pattern: result = await sub_agent.run(task, usage=ctx.usage)

Usage (tokens, costs) from delegate agents accumulates into the parent's RunUsage, giving accurate cost tracking across the entire delegation chain.

UsageLimits set at the top level are enforced across all delegated runs.

DelegationConfig Objects​

@dataclass
class DelegationConfig()

Configuration for agent delegation.

Parameters​

model : str | None Model override for the sub-agent (e.g. 'openai:gpt-4.1-mini'). If None, the sub-agent uses its default model. propagate_usage : bool Whether to propagate the parent's usage tracker to the sub-agent. Default True — this is the recommended pattern from pydantic-ai. propagate_trace : bool Whether to propagate OpenTelemetry trace context. timeout : float | None Timeout for the delegated run in seconds.

DelegationResult Objects​

@dataclass
class DelegationResult()

Result from a delegated agent run.

Parameters​

output : str The text output from the sub-agent. agent_id : str Identifier of the sub-agent that produced the result. tokens_used : int Tokens consumed by this delegation (input + output). cost_usd : float Estimated cost for this delegation. metadata : dict Additional metadata (model used, request count, etc.).

delegate_to_agent​

async def delegate_to_agent(
        sub_agent: Any,
        task: str,
        parent_usage: Any | None = None,
        config: DelegationConfig | None = None) -> DelegationResult

Delegate work to a sub-agent with usage tracking and OTEL propagation.

This implements the pydantic-ai recommended delegation pattern:

  1. Pass parent's usage tracker to accumulate tokens/cost
  2. Propagate OTEL trace context for full-stack observability
  3. Return structured result with usage metrics

Parameters​

sub_agent : Any A pydantic-ai Agent instance to delegate to. task : str The task/prompt to send to the sub-agent. parent_usage : Any | None The parent agent's RunUsage instance. Passing it ensures tokens from the sub-agent accumulate into the parent's totals. config : DelegationConfig | None Optional delegation configuration.

Returns​

DelegationResult Structured result with output and usage metrics.

Example​

@agent.tool
async def delegate_analysis(ctx: RunContext, task: str) -> str:
    result = await delegate_to_agent(
        analysis_agent, task, parent_usage=ctx.usage
    )
    return result.output