Dynamic agent API

Helpers shared by the create_agent and delegate tools and by create_agent_factory_toolset: they validate a runtime agent request and build the agent instance. See Dynamic agents.

AgentFactory

subagents_pydantic_ai.AgentFactory = Callable[[SubAgentConfig], Any] module-attribute

Builds the agent instance for a dynamically created subagent.

Receives the freshly built SubAgentConfig and returns an agent instance. The return type is Any because the agent need not be a pydantic_ai.Agent — consumers such as pydantic-deep return their own agent objects.

A factory must not attach an ask_parent toolset; the subagent toolset injects one at run time for registry-backed and one-shot agents, and a second copy would collide on the tool name.

build_dynamic_agent

subagents_pydantic_ai.dynamic_agent.build_dynamic_agent(ctx, *, name, description, instructions, model, can_ask_questions=True, capabilities=None, allowed_models=None, toolsets_factory=None, capabilities_map=None, default_agent_factory=None)

Validate inputs and build a dynamic agent.

Returns:

Type	Description
str | tuple[Any, SubAgentConfig]	An error string on failure, or (agent, config) on success.

Source code in src/subagents_pydantic_ai/dynamic_agent.py

def build_dynamic_agent(
    ctx: RunContext[SubAgentDepsProtocol],
    *,
    name: str,
    description: str,
    instructions: str,
    model: str | Model,
    can_ask_questions: bool = True,
    capabilities: list[str] | None = None,
    allowed_models: list[str] | None = None,
    toolsets_factory: ToolsetFactory | None = None,
    capabilities_map: dict[str, CapabilityFactory] | None = None,
    default_agent_factory: AgentFactory | None = None,
) -> str | tuple[Any, SubAgentConfig]:
    """Validate inputs and build a dynamic agent.

    Returns:
        An error string on failure, or ``(agent, config)`` on success.
    """
    name_error = validate_agent_name(name)
    if name_error is not None:
        return name_error

    model_error = validate_model(model, allowed_models)
    if model_error is not None:
        return model_error

    caps_error = validate_capabilities(capabilities, capabilities_map)
    if caps_error is not None:
        return caps_error

    factory_caps_error = validate_capabilities_with_factory(
        capabilities,
        default_agent_factory,
    )
    if factory_caps_error is not None:
        return factory_caps_error

    config = build_subagent_config(
        name=name,
        description=description,
        instructions=instructions,
        model=model,
        can_ask_questions=can_ask_questions,
    )

    try:
        agent = build_agent_instance(
            ctx,
            config,
            model=model,
            capabilities=capabilities,
            toolsets_factory=toolsets_factory,
            capabilities_map=capabilities_map,
            default_agent_factory=default_agent_factory,
        )
    except ValueError as exc:
        return f"Error: {exc}"
    except Exception as exc:
        return f"Error creating agent: {exc}"

    return agent, config

Validation helpers

subagents_pydantic_ai.dynamic_agent

Shared helpers for building dynamic subagents at runtime.

validate_agent_name(name)

Return an error message when the agent name is invalid.

Source code in src/subagents_pydantic_ai/dynamic_agent.py

def validate_agent_name(name: str) -> str | None:
    """Return an error message when the agent name is invalid."""
    if not _AGENT_NAME_PATTERN.fullmatch(name):
        return "Error: Name must contain only letters, numbers, and hyphens"
    return None

validate_model(model, allowed_models)

Return an error message when the model is not allowed.

Source code in src/subagents_pydantic_ai/dynamic_agent.py

def validate_model(
    model: str | Model,
    allowed_models: list[str] | None,
) -> str | None:
    """Return an error message when the model is not allowed."""
    if allowed_models and model not in allowed_models:
        allowed = ", ".join(allowed_models)
        return f"Error: Model '{model}' is not allowed. Use one of: {allowed}"
    return None

validate_capabilities(capabilities, capabilities_map)

Return an error message when capabilities are unknown.

Source code in src/subagents_pydantic_ai/dynamic_agent.py

def validate_capabilities(
    capabilities: list[str] | None,
    capabilities_map: dict[str, CapabilityFactory] | None,
) -> str | None:
    """Return an error message when capabilities are unknown."""
    if capabilities and capabilities_map:
        invalid_caps = [cap for cap in capabilities if cap not in capabilities_map]
        if invalid_caps:
            available = ", ".join(capabilities_map.keys())
            invalid = ", ".join(invalid_caps)
            return f"Error: Unknown capabilities: {invalid}. Available: {available}"
    return None

validate_capabilities_with_factory(capabilities, default_agent_factory)

Return an error when capabilities cannot be injected with a custom factory.

Source code in src/subagents_pydantic_ai/dynamic_agent.py

def validate_capabilities_with_factory(
    capabilities: list[str] | None,
    default_agent_factory: AgentFactory | None,
) -> str | None:
    """Return an error when capabilities cannot be injected with a custom factory."""
    if default_agent_factory is not None and capabilities:
        return (
            "Error: capabilities are not supported when a custom "
            "default_agent_factory is configured. The factory builds the "
            "agent itself; create the agent without capabilities or "
            "configure the factory to attach the required toolsets."
        )
    return None

build_subagent_config(*, name, description, instructions, model, can_ask_questions=True)

Build a runtime subagent configuration.

Source code in src/subagents_pydantic_ai/dynamic_agent.py

def build_subagent_config(
    *,
    name: str,
    description: str,
    instructions: str,
    model: str | Model,
    can_ask_questions: bool = True,
) -> SubAgentConfig:
    """Build a runtime subagent configuration."""
    return SubAgentConfig(
        name=name,
        description=description,
        instructions=instructions,
        model=model,
        can_ask_questions=can_ask_questions,
    )

collect_agent_toolsets(ctx, capabilities, *, toolsets_factory, capabilities_map)

Collect toolsets to attach to a dynamically built agent.

Source code in src/subagents_pydantic_ai/dynamic_agent.py

def collect_agent_toolsets(
    ctx: RunContext[SubAgentDepsProtocol],
    capabilities: list[str] | None,
    *,
    toolsets_factory: ToolsetFactory | None,
    capabilities_map: dict[str, CapabilityFactory] | None,
) -> list[Any]:
    """Collect toolsets to attach to a dynamically built agent."""
    agent_toolsets: list[Any] = []
    if toolsets_factory:
        agent_toolsets.extend(toolsets_factory(ctx.deps))
    elif capabilities and capabilities_map:
        for cap_name in capabilities:
            cap_factory = capabilities_map[cap_name]
            agent_toolsets.extend(cap_factory(ctx.deps))
    return agent_toolsets

build_agent_instance(ctx, config, *, model, capabilities, toolsets_factory, capabilities_map, default_agent_factory)

Build a pydantic-ai Agent instance for a dynamic subagent.

Source code in src/subagents_pydantic_ai/dynamic_agent.py

def build_agent_instance(
    ctx: RunContext[SubAgentDepsProtocol],
    config: SubAgentConfig,
    *,
    model: str | Model,
    capabilities: list[str] | None,
    toolsets_factory: ToolsetFactory | None,
    capabilities_map: dict[str, CapabilityFactory] | None,
    default_agent_factory: AgentFactory | None,
) -> Any:
    """Build a pydantic-ai Agent instance for a dynamic subagent."""
    if default_agent_factory is not None:
        return default_agent_factory(config)

    agent_toolsets = collect_agent_toolsets(
        ctx,
        capabilities,
        toolsets_factory=toolsets_factory,
        capabilities_map=capabilities_map,
    )
    return Agent(
        model,
        system_prompt=config["instructions"],
        toolsets=agent_toolsets or None,
    )