Dynamic Agent Creation: Spawning Specialist Agents On-Demand Based on Task Requirements

Why Create Agents Dynamically?

Static multi-agent architectures define all agents at design time. This works when you know exactly which specialists you need. But many real-world problems require agents whose capabilities cannot be predicted in advance.

A customer asks about a product you launched last week — you need an agent with that product's documentation loaded. A code review involves Rust and Terraform — you need agents specialized in both, not the Python expert that sits idle. A financial analysis request arrives for a market sector you rarely handle — you need to spin up an agent with the right data access.

Dynamic agent creation solves this by treating agents as runtime resources that are instantiated from templates, configured for the specific task, and cleaned up when finished.

The Agent Factory Pattern

from dataclasses import dataclass, field
from typing import Any, Callable
from datetime import datetime
import uuid

@dataclass
class AgentSpec:
    """Template for creating agents."""
    role: str
    system_prompt: str
    model: str
    tools: list[str]
    max_tokens: int = 4096
    temperature: float = 0.7

@dataclass
class AgentInstance:
    instance_id: str
    spec: AgentSpec
    created_at: str
    state: dict[str, Any] = field(default_factory=dict)
    is_active: bool = True

class AgentFactory:
    def __init__(self):
        self._templates: dict[str, AgentSpec] = {}
        self._active_instances: dict[str, AgentInstance] = {}

    def register_template(self, name: str, spec: AgentSpec):
        self._templates[name] = spec

    def create(
        self, template_name: str, overrides: dict | None = None
    ) -> AgentInstance:
        if template_name not in self._templates:
            raise KeyError(f"No template: {template_name}")

        spec = self._templates[template_name]
        if overrides:
            spec_dict = {
                "role": spec.role,
                "system_prompt": spec.system_prompt,
                "model": spec.model,
                "tools": list(spec.tools),
                "max_tokens": spec.max_tokens,
                "temperature": spec.temperature,
            }
            spec_dict.update(overrides)
            spec = AgentSpec(**spec_dict)

        instance = AgentInstance(
            instance_id=str(uuid.uuid4()),
            spec=spec,
            created_at=datetime.now().isoformat(),
        )
        self._active_instances[instance.instance_id] = instance
        return instance

    def destroy(self, instance_id: str):
        instance = self._active_instances.pop(instance_id, None)
        if instance:
            instance.is_active = False
            self._cleanup(instance)

    def _cleanup(self, instance: AgentInstance):
        instance.state.clear()

    @property
    def active_count(self) -> int:
        return len(self._active_instances)

Agent Pools for High-Throughput Workloads

Creating and destroying agents for every request is wasteful if the same types of agents are needed repeatedly. An agent pool pre-creates instances and leases them out, similar to database connection pooling.

flowchart LR
    CALLER(["Client"])
    subgraph TEL["Telephony"]
        SIP["Twilio SIP and PSTN"]
    end
    subgraph BRAIN["Salon AI Agent"]
        STT["Streaming STT<br/>Deepgram or Whisper"]
        NLU{"Intent and<br/>Entity Extraction"}
        TOOLS["Tool Calls"]
        TTS["Streaming TTS<br/>ElevenLabs or Rime"]
    end
    subgraph DATA["Live Data Plane"]
        CRM[("CRM and Notes")]
        CAL[("Calendar and<br/>Schedule")]
        KB[("Knowledge Base<br/>and Policies")]
    end
    subgraph OUT["Outcomes"]
        O1(["Appointment booked"])
        O2(["Reschedule completed"])
        O3(["Stylist handoff"])
    end
    CALLER --> SIP --> STT --> NLU
    NLU -->|Lookup| TOOLS
    TOOLS <--> CRM
    TOOLS <--> CAL
    TOOLS <--> KB
    NLU --> TTS --> SIP --> CALLER
    NLU -->|Resolved| O1
    NLU -->|Schedule| O2
    NLU -->|Escalate| O3
    style CALLER fill:#f1f5f9,stroke:#64748b,color:#0f172a
    style NLU fill:#4f46e5,stroke:#4338ca,color:#fff
    style O1 fill:#059669,stroke:#047857,color:#fff
    style O2 fill:#0ea5e9,stroke:#0369a1,color:#fff
    style O3 fill:#f59e0b,stroke:#d97706,color:#1f2937

import asyncio
from collections import defaultdict

class AgentPool:
    def __init__(
        self, factory: AgentFactory, max_per_type: int = 5
    ):
        self.factory = factory
        self.max_per_type = max_per_type
        self._available: dict[str, list[AgentInstance]] = defaultdict(list)
        self._leased: dict[str, AgentInstance] = {}
        self._lock = asyncio.Lock()

    async def acquire(self, template_name: str) -> AgentInstance:
        async with self._lock:
            pool = self._available[template_name]

            if pool:
                instance = pool.pop()
                instance.state.clear()  # reset state for reuse
            else:
                instance = self.factory.create(template_name)

            self._leased[instance.instance_id] = instance
            return instance

    async def release(self, instance_id: str):
        async with self._lock:
            instance = self._leased.pop(instance_id, None)
            if not instance:
                return

            template = instance.spec.role
            pool = self._available[template]

            if len(pool) < self.max_per_type:
                instance.state.clear()
                pool.append(instance)
            else:
                self.factory.destroy(instance.instance_id)

    async def drain(self):
        """Gracefully shut down all pooled agents."""
        async with self._lock:
            for pool in self._available.values():
                for instance in pool:
                    self.factory.destroy(instance.instance_id)
            self._available.clear()

Lifecycle Management with Context Managers

To prevent resource leaks, wrap agent usage in context managers that guarantee cleanup.

from contextlib import asynccontextmanager

class AgentOrchestrator:
    def __init__(self, pool: AgentPool):
        self.pool = pool

    @asynccontextmanager
    async def specialist(self, template_name: str, **overrides):
        instance = await self.pool.acquire(template_name)
        try:
            yield instance
        finally:
            await self.pool.release(instance.instance_id)

    async def handle_task(self, task: dict) -> dict:
        required_roles = self._analyze_requirements(task)
        results = {}

        for role in required_roles:
            async with self.specialist(role) as agent:
                result = await self._execute_agent(agent, task)
                results[role] = result

        return self._merge_results(results)

    def _analyze_requirements(self, task: dict) -> list[str]:
        """Determine which specialist templates are needed."""
        requirements = []
        content = task.get("content", "").lower()

        if "code" in content or "implement" in content:
            requirements.append("code_specialist")
        if "analyze" in content or "data" in content:
            requirements.append("data_analyst")
        if "review" in content or "security" in content:
            requirements.append("security_reviewer")

        return requirements or ["generalist"]

    async def _execute_agent(self, agent, task):
        pass

    def _merge_results(self, results):
        return results

Dynamic Tool Assignment

Beyond just selecting templates, you can dynamically compose an agent's tool set based on what the task requires.

class DynamicToolAssigner:
    def __init__(self):
        self._tool_registry: dict[str, Callable] = {}
        self._tool_metadata: dict[str, dict] = {}

    def register_tool(
        self, name: str, fn: Callable, metadata: dict
    ):
        self._tool_registry[name] = fn
        self._tool_metadata[name] = metadata

    def select_tools(
        self, task_description: str, max_tools: int = 6
    ) -> list[str]:
        scored = []
        task_lower = task_description.lower()

        for name, meta in self._tool_metadata.items():
            keywords = meta.get("keywords", [])
            relevance = sum(
                1 for kw in keywords if kw in task_lower
            )
            if relevance > 0:
                scored.append((name, relevance))

        scored.sort(key=lambda x: x[1], reverse=True)
        return [name for name, _ in scored[:max_tools]]

This prevents the tool-overload problem where agents degrade when given too many tools. Each dynamically created agent gets only the tools relevant to its task, keeping the tool set small and focused.

FAQ

How do I prevent runaway agent creation from exhausting resources?

Set hard limits at multiple levels: maximum active instances per template type, maximum total active instances across all types, and a global timeout after which any agent is forcefully destroyed. The agent pool pattern with max_per_type handles the first two. For the timeout, add a background reaper task that checks created_at timestamps and destroys any instance older than your threshold.

Should I create a new agent for every user message, or reuse agents across a conversation?

Reuse agents within a single conversation session. Create a fresh agent when a new conversation starts or when the topic shifts to a completely different domain. The agent pool pattern supports this — acquire an agent at conversation start, use it across multiple turns, and release it when the conversation ends.

How do I handle dynamic agent failures mid-task?

Wrap each agent execution in a try/except that catches failures, logs the error with the agent's instance ID and configuration, releases the failed agent back to the pool (or destroys it if the failure corrupted its state), and retries with a fresh instance. Limit retries to 2-3 attempts before escalating to a human or returning an error to the caller.

---

#DynamicAgents #FactoryPattern #AgentLifecycle #ResourceManagement #Python #AgenticAI #LearnAI #AIEngineering