Dynamic Agent Orchestration
The coordinator pattern assumes the orchestrator knows its workers. At design time, you wire up a research agent, a coding agent, and a writing agent. The coordinator picks which ones to call, but the roster is fixed. Adding a new capability — say, a diagram-generation agent — means changing the coordinator's code, updating its system prompt, and redeploying.
Dynamic orchestration obtains its worker roster at runtime. The orchestrator discovers available agents from a registry, reads their capability descriptions, matches those capabilities to the task at hand, and builds an execution plan from the current pool. Adding a new capability means registering a new agent — the orchestrator picks it up automatically the next time it plans a task.
This is the difference between a team where the manager has a fixed org chart and a team where the manager has access to a company directory and can assemble a project team on the fly based on who has the right skills. The second approach is more flexible, but it requires the directory to be well-organized, the skill descriptions to be accurate, and the manager to be good at matching skills to tasks.
The Break Down #
Static coordinator setups work when the agent roster is small and stable. Three workers, five workers — you can manage the system prompt, test the routing, and keep everything in your head. But several forces push toward dynamic orchestration:
Agent proliferation. As organizations build more agents, the number of available capabilities grows. A customer service system might start with three agents (FAQ, billing, technical support) and grow to fifteen (returns, shipping, loyalty, escalation, fraud, product recommendations, account changes, and so on). Maintaining a single coordinator with fifteen hardcoded workers means a system prompt that is thousands of tokens long and a routing decision that gets less reliable with every new worker.
Independent deployment. In larger teams, different groups build and deploy agents independently. The payments team ships a refund agent. The logistics team ships a tracking agent. The marketing team ships a promotions agent. If the coordinator must be updated every time a downstream team deploys a new agent, it becomes a bottleneck — the one component every team depends on, often with diffuse ownership.
Multi-tenant and marketplace scenarios. When agents come from different providers — a tool marketplace, a partner ecosystem, an agent-to-agent protocol — the available roster emerges at runtime. The orchestrator discovers and evaluates that dynamic set as part of planning.
Static orchestration: Dynamic orchestration:
Coordinator Coordinator
(knows workers at build time) (discovers workers at runtime)
│ │
┌────┼────┐ ┌────┴─────┐
│ │ │ │ Registry │
▼ ▼ ▼ └────┬─────┘
W1 W2 W3 │
┌────┼────┬────┐
Fixed roster. ▼ ▼ ▼ ▼
Change = redeploy. W1 W2 W3 W4...
Roster changes without
touching the coordinator.
The Agent Registry #
The registry is the central data structure that makes dynamic orchestration possible. It is a catalog of available agents, each described by metadata that tells the orchestrator what the agent can do, how to invoke it, and what constraints apply.
from dataclasses import dataclass, field
@dataclass
class AgentSkill:
"""One capability an agent offers."""
skill_id: str
name: str
description: str
input_modes: list[str] = field(default_factory=lambda: ["text"])
output_modes: list[str] = field(default_factory=lambda: ["text"])
examples: list[str] = field(default_factory=list)
tags: list[str] = field(default_factory=list)
@dataclass
class AgentCard:
"""
A self-description published by an agent,
inspired by the A2A protocol's Agent Card concept.
"""
agent_id: str
name: str
description: str
skills: list[AgentSkill]
endpoint: str
version: str = "1.0"
max_concurrent: int = 5
avg_latency_ms: float = 2000.0
cost_per_call: float = 0.0
tags: list[str] = field(default_factory=list)
requires_auth: bool = False
class AgentRegistry:
"""
A catalog of available agents that the orchestrator
queries at runtime to discover capabilities.
"""
def __init__(self):
self._agents: dict[str, AgentCard] = {}
def register(self, card: AgentCard) -> None:
self._agents[card.agent_id] = card
def deregister(self, agent_id: str) -> None:
self._agents.pop(agent_id, None)
def list_all(self) -> list[AgentCard]:
return list(self._agents.values())
def search_by_tags(self, tags: list[str]) -> list[AgentCard]:
return [
card for card in self._agents.values()
if any(tag in card.tags for tag in tags)
]
def search_by_skill(self, skill_query: str) -> list[AgentCard]:
"""Simple keyword match on skill descriptions."""
query_lower = skill_query.lower()
results = []
for card in self._agents.values():
for skill in card.skills:
if (query_lower in skill.description.lower()
or query_lower in skill.name.lower()):
results.append(card)
break
return results
def get_catalog_summary(self) -> str:
"""
Generate a text summary of all available agents
for injection into the orchestrator's context.
"""
lines = []
for card in self._agents.values():
skills_text = ", ".join(s.name for s in card.skills)
lines.append(
f"- {card.name} (id: {card.agent_id}): {card.description}. "
f"Skills: {skills_text}. "
f"Avg latency: {card.avg_latency_ms:.0f}ms. "
f"Cost: ${card.cost_per_call:.4f}/call."
)
return "\n".join(lines)
The AgentCard is the key abstraction. It contains everything the orchestrator needs to decide whether an agent is useful for a given task: a human-readable description, a list of skills with examples, performance characteristics (latency, cost), and an endpoint for invocation. This is the same concept behind the A2A protocol's Agent Card — a machine-readable self-description that enables discovery.
Registration Patterns #
Agents can join the registry in several ways:
Self-registration. When an agent starts up, it publishes its card to the registry. When it shuts down, it deregisters. This is the microservices pattern — each service registers with a service discovery mechanism on startup.
External registration. An operator or deployment pipeline registers agent cards. This approach gives the registry an operator-approved roster.
Discovery-based registration. The registry periodically scans known endpoints (well-known URIs, network ranges) and fetches agent cards from any agent that responds. This is how A2A's well-known URI pattern works — agents publish their card at /.well-known/agent-card.json and registries can crawl for them.
Capability Matching #
Given a user task and a catalog of available agents, the orchestrator needs to decide which agents to use. This is a matching problem: map task requirements to agent capabilities.
LLM-Based Matching #
The most flexible approach is to give the orchestrator model the full catalog and let it reason about which agents fit the task.
async def select_agents(
task: str,
registry: AgentRegistry,
model: "ModelClient",
max_agents: int = 5,
) -> list[AgentCard]:
"""
Use the model to select relevant agents from the registry
based on the task description.
"""
import json
if max_agents < 1:
raise ValueError("max_agents must be at least 1")
catalog = registry.get_catalog_summary()
response = await model.chat(
system="""You are an orchestration planner. Given a task and a catalog
of available agents, select the agents needed to complete the task.
Return a JSON array of agent IDs. Select only agents whose skills
are directly relevant. Prefer fewer agents over more.
Available agents:
""" + catalog,
messages=[{"role": "user", "content": f"Task: {task}"}],
)
selected_ids = json.loads(response.text)
if not isinstance(selected_ids, list):
raise ValueError("Model response must be a JSON array of agent IDs")
cards_by_id = {card.agent_id: card for card in registry.list_all()}
selected = []
seen = set()
for agent_id in selected_ids:
if agent_id in cards_by_id and agent_id not in seen:
selected.append(cards_by_id[agent_id])
seen.add(agent_id)
if len(selected) == max_agents:
break
return selected
The model reads the catalog, understands the task, and picks agents. This works when the catalog fits in the context window and the model can reason about the match. For catalogs under a hundred agents, this is the simplest and most effective approach.
Embedding-Based Matching #
For large registries — hundreds or thousands of agents — injecting the full catalog into the context window is impractical. Instead, embed agent descriptions into a vector store and retrieve the most relevant ones.
class EmbeddingMatcher:
"""
Match tasks to agents using semantic similarity
between the task description and agent skill descriptions.
"""
def __init__(self, registry: AgentRegistry, embedder: "EmbeddingModel"):
self.registry = registry
self.embedder = embedder
self._index: list[tuple[str, str, list[float]]] = []
def build_index(self) -> None:
self._index = []
for card in self.registry.list_all():
for skill in card.skills:
text = f"{card.name}: {skill.name} — {skill.description}"
embedding = self.embedder.embed(text)
self._index.append((card.agent_id, skill.skill_id, embedding))
def match(self, task: str, top_k: int = 5) -> list[AgentCard]:
if top_k < 1:
raise ValueError("top_k must be at least 1")
task_embedding = self.embedder.embed(task)
scored = []
for agent_id, _skill_id, emb in self._index:
score = cosine_similarity(task_embedding, emb)
scored.append((score, agent_id))
scored.sort(reverse=True)
seen = set()
selected = []
cards_by_id = {card.agent_id: card for card in self.registry.list_all()}
for score, agent_id in scored:
if agent_id not in seen and agent_id in cards_by_id and score > 0.5:
seen.add(agent_id)
selected.append(cards_by_id[agent_id])
if len(selected) >= top_k:
break
return selected
The hybrid approach works best: use embedding-based matching to narrow the catalog from hundreds to a dozen, then inject those dozen into the model's context for final selection and plan generation. This extends the same RAG pattern from document retrieval to agent discovery.
Dynamic Plan Generation #
Once the orchestrator knows which agents are available, it builds an execution plan — the sequence and structure of agent invocations needed to complete the task. Static workflows hardcode this topology; dynamic orchestration derives it from the model's reasoning about the task and available capabilities.
import json
from dataclasses import dataclass, field
@dataclass
class PlanStep:
step_id: str
agent_id: str
task_description: str
depends_on: list[str] = field(default_factory=list)
timeout_seconds: float = 60.0
@dataclass
class ExecutionPlan:
steps: list[PlanStep]
parallel_groups: list[list[str]] = field(default_factory=list)
async def generate_plan(
task: str,
available_agents: list[AgentCard],
model: "ModelClient",
) -> ExecutionPlan:
"""
Generate a dynamic execution plan using selected agents.
"""
agent_descriptions = "\n".join(
f"- {card.agent_id}: {card.description} "
f"(skills: {', '.join(s.name for s in card.skills)})"
for card in available_agents
)
response = await model.chat(
system=f"""You are an execution planner. Given a task and a set of
available agents, produce an execution plan as JSON.
Each step has:
- "step_id": unique identifier
- "agent_id": which agent to invoke
- "task_description": what to ask that agent
- "depends_on": list of step_ids that must complete first
Steps with no dependencies can run in parallel.
Use only the agents listed below. Minimize the number of steps.
Available agents:
{agent_descriptions}""",
messages=[{"role": "user", "content": f"Task: {task}"}],
)
plan_data = json.loads(response.text)
if not isinstance(plan_data, dict) or not isinstance(plan_data.get("steps"), list):
raise ValueError("Model response must contain a JSON 'steps' array")
steps = [PlanStep(**step) for step in plan_data["steps"]]
allowed_agent_ids = {card.agent_id for card in available_agents}
unknown_agents = {step.agent_id for step in steps} - allowed_agent_ids
if unknown_agents:
raise ValueError(f"Plan references unavailable agents: {sorted(unknown_agents)}")
# Compute parallel groups from dependency graph
parallel_groups = compute_parallel_groups(steps)
return ExecutionPlan(steps=steps, parallel_groups=parallel_groups)
def compute_parallel_groups(steps: list[PlanStep]) -> list[list[str]]:
"""
Group steps into waves that can execute in parallel.
Each wave contains steps whose dependencies are all
satisfied by previous waves.
"""
step_ids = [step.step_id for step in steps]
if len(step_ids) != len(set(step_ids)):
raise ValueError("step_id values must be unique")
known_ids = set(step_ids)
unknown_dependencies = {
dependency
for step in steps
for dependency in step.depends_on
if dependency not in known_ids
}
if unknown_dependencies:
raise ValueError(
f"Plan contains unknown dependencies: {sorted(unknown_dependencies)}"
)
completed: set[str] = set()
remaining = {s.step_id: s for s in steps}
groups = []
while remaining:
ready = [
sid for sid, step in remaining.items()
if all(dep in completed for dep in step.depends_on)
]
if not ready:
raise ValueError(
f"Plan contains a dependency cycle: {sorted(remaining)}"
)
groups.append(ready)
for sid in ready:
completed.add(sid)
del remaining[sid]
return groups
The plan is a dependency graph. Independent steps form the first wave and execute in parallel. Steps that depend on first-wave results form the second wave, and so on. This is the same topological execution model used in workflow orchestration, with the graph generated at runtime.
Plan Execution #
import asyncio
async def execute_plan(
plan: ExecutionPlan,
agent_invoker: "AgentInvoker",
) -> dict[str, str]:
"""
Execute a dynamic plan, running parallel groups concurrently
and passing results to dependent steps.
"""
results: dict[str, str] = {}
for group in plan.parallel_groups:
group_steps = [
s for s in plan.steps if s.step_id in group
]
# Build context for each step from its dependencies
tasks = []
for step in group_steps:
dep_context = "\n".join(
f"Result from {dep}: {results[dep]}"
for dep in step.depends_on
if dep in results
)
full_task = step.task_description
if dep_context:
full_task = f"{step.task_description}\n\nContext:\n{dep_context}"
tasks.append(
invoke_with_timeout(
agent_invoker, step.agent_id, full_task, step.timeout_seconds
)
)
group_results = await asyncio.gather(*tasks, return_exceptions=True)
for step, result in zip(group_steps, group_results):
if isinstance(result, Exception):
results[step.step_id] = f"Error: {result}"
else:
results[step.step_id] = result
return results
async def invoke_with_timeout(
invoker: "AgentInvoker",
agent_id: str,
task: str,
timeout: float,
) -> str:
try:
return await asyncio.wait_for(
invoker.invoke(agent_id, task),
timeout=timeout,
)
except asyncio.TimeoutError:
return f"Error: agent {agent_id} timed out after {timeout}s"
Example: "Research competitor pricing and write a summary report"
Registry contains:
- web_search: searches the web
- data_analyst: analyzes structured data
- writer: produces polished prose
- chart_gen: creates charts from data
Generated plan:
Wave 1 (parallel):
┌──────────────────────┐ ┌───────────────────────┐
│ step_1: web_search │ │ step_2: web_search │
│ "Find pricing for │ │ "Find pricing for │
│ competitor A" │ │ competitor B" │
└──────────┬───────────┘ └──────────┬────────────┘
│ │
└────────────┬────────────┘
│
Wave 2: ▼
┌──────────────────────────────────────┐
│ step_3: data_analyst │
│ "Compare pricing across competitors" │
│ depends_on: [step_1, step_2] │
└──────────────────┬───────────────────┘
│
Wave 3: ▼
┌──────────────────────────────────────┐
│ step_4: writer │
│ "Write a summary report" │
│ depends_on: [step_3] │
└──────────────────────────────────────┘
The registry supplied web_search, data_analyst, and writer; the orchestrator selected them based on the task and wired them together in a plan. A visualization request would similarly bring chart_gen into the plan, while the current task selects only the three relevant agents.
Handling Failures in Dynamic Plans #
Static orchestrators can have handcrafted error handling for each worker. Dynamic orchestrators select workers at runtime, so their error handling must be generic and adaptive.
async def execute_with_fallback(
plan: ExecutionPlan,
registry: AgentRegistry,
invoker: "AgentInvoker",
model: "ModelClient",
) -> dict[str, str]:
"""
Execute a plan with automatic replanning on failure.
"""
results: dict[str, str] = {}
for group in plan.parallel_groups:
group_steps = [s for s in plan.steps if s.step_id in group]
tasks = []
for step in group_steps:
dep_context = build_dep_context(step, results)
tasks.append(
invoke_with_timeout(invoker, step.agent_id,
step.task_description + dep_context,
step.timeout_seconds)
)
group_results = await asyncio.gather(*tasks, return_exceptions=True)
failures = []
for step, result in zip(group_steps, group_results):
if isinstance(result, Exception) or (
isinstance(result, str) and result.startswith("Error:")
):
failures.append(step)
results[step.step_id] = str(result)
else:
results[step.step_id] = result
# Replan failed steps
if failures:
for failed_step in failures:
replacement = await find_replacement(
failed_step, registry, model, results
)
if replacement:
retry_task = (
failed_step.task_description
+ build_dep_context(failed_step, results)
)
retry_result = await invoke_with_timeout(
invoker,
replacement.agent_id,
retry_task,
failed_step.timeout_seconds,
)
results[failed_step.step_id] = retry_result
return results
def build_dep_context(step: PlanStep, results: dict[str, str]) -> str:
dependency_results = [
f"Result from {dependency}: {results[dependency]}"
for dependency in step.depends_on
if dependency in results
]
if not dependency_results:
return ""
return "\n\nContext:\n" + "\n".join(dependency_results)
async def find_replacement(
failed_step: PlanStep,
registry: AgentRegistry,
model: "ModelClient",
current_results: dict[str, str],
) -> AgentCard | None:
"""
Find an alternative agent that can handle the failed step.
"""
candidates = [
card for card in registry.list_all()
if card.agent_id != failed_step.agent_id
]
if not candidates:
return None
# Let the model pick the best replacement
candidate_text = "\n".join(
f"- {c.agent_id}: {c.description}" for c in candidates
)
response = await model.chat(
system=f"""An agent failed to complete a task. Select the best
replacement from the candidates below, or respond "none" if no
candidate is suitable.
Failed task: {failed_step.task_description}
Error: {current_results.get(failed_step.step_id, 'unknown')}
Candidates:
{candidate_text}""",
messages=[{"role": "user", "content": "Which agent should retry this?"}],
)
selected_id = response.text.strip().strip("`\"'")
return next((c for c in candidates if c.agent_id == selected_id), None)
The pattern: when an agent fails, query the registry for alternative agents with similar capabilities, let the model pick a replacement, and retry. Dynamic discovery gives the orchestrator access to suitable replacements beyond its original roster.
The Orchestrator's System Prompt #
The orchestrator itself is an agent, and its system prompt must be written to handle the dynamic nature of the agent pool. A static coordinator's prompt lists specific workers; a dynamic orchestrator's prompt describes how to work with any set of agents.
DYNAMIC_ORCHESTRATOR_PROMPT = """You are a task orchestrator. You receive a
user task and a catalog of available agents. Your job is to:
1. Analyze the task and identify what capabilities are needed.
2. Select agents from the catalog whose skills match those needs.
3. Create a step-by-step execution plan that assigns work to agents.
4. Specify dependencies between steps so independent work runs in parallel.
Rules:
- Use only agents from the catalog. Do not invent agents.
- Each step should have a clear, focused task description.
- Minimize the number of steps and agents used.
- If no available agent can handle a required capability, note this
as a gap and produce the best plan possible with what is available.
- Prefer agents with lower latency and cost when multiple agents
have equivalent capabilities.
The catalog will be provided with each request. Do not assume agents
from previous requests are still available."""
The last line is important. In a dynamic system, agents come and go. The orchestrator must treat each task as a fresh discovery — the registry might have changed since the last request.
Registry Health and Agent Quality #
Reliable quality signals are essential to a useful registry and sound agent selection.
@dataclass
class AgentHealthMetrics:
agent_id: str
success_rate: float # 0.0 to 1.0
avg_latency_ms: float
p99_latency_ms: float
error_count_last_hour: int
last_successful_call: str | None # ISO timestamp
is_healthy: bool = True
class HealthAwareRegistry(AgentRegistry):
"""Registry that tracks agent health and filters unhealthy agents."""
def __init__(self):
super().__init__()
self._health: dict[str, AgentHealthMetrics] = {}
def update_health(self, metrics: AgentHealthMetrics) -> None:
self._health[metrics.agent_id] = metrics
def list_healthy(self, min_success_rate: float = 0.8) -> list[AgentCard]:
return [
card for card in self._agents.values()
if self._is_healthy(card.agent_id, min_success_rate)
]
def _is_healthy(self, agent_id: str, min_success_rate: float) -> bool:
metrics = self._health.get(agent_id)
if metrics is None:
return True # No data yet — assume healthy
return (
metrics.is_healthy
and metrics.success_rate >= min_success_rate
)
def get_catalog_summary(self) -> str:
lines = []
for card in self.list_healthy():
metrics = self._health.get(card.agent_id)
health_note = ""
if metrics:
health_note = (
f" Success rate: {metrics.success_rate:.0%}."
f" Avg latency: {metrics.avg_latency_ms:.0f}ms."
)
skills_text = ", ".join(s.name for s in card.skills)
lines.append(
f"- {card.name} (id: {card.agent_id}): {card.description}. "
f"Skills: {skills_text}.{health_note}"
)
return "\n".join(lines)
The registry filters unhealthy agents — those with high error rates or extreme latency — before presenting the catalog. The orchestrator receives a healthy pool with quality metrics in the catalog summary, allowing the model to prefer faster or more reliable agents when multiple options exist.
When Static Orchestration Is Better #
Dynamic orchestration adds a discovery step, a matching step, and a plan generation step before any real work begins. That is three model calls before the first agent even starts. For simple, stable tasks, this overhead is pure waste.
Use static orchestration when:
- The agent roster is small and stable (under ten agents)
- The routing logic is simple (classify and route to one of a few workers)
- The task structure is predictable (always the same sequence of steps)
- Latency is critical (every extra model call hurts)
Use dynamic orchestration when:
- Agents are deployed independently by different teams
- The agent roster changes frequently (new agents, retired agents)
- Tasks are diverse enough to require multiple topologies
- You are building a platform where third parties contribute agents
- Failure recovery benefits from finding alternative agents at runtime
The heuristic: a topology that stays stable month to month suits static orchestration. A topology that depends on what is available right now calls for dynamic discovery.
Conclusion #
Dynamic agent orchestration is the coordinator pattern with the roster decoupled from the code. The key components:
- Agent registries hold capability descriptions (Agent Cards) for available agents. Agents self-register, or operators register them, or the registry discovers them by crawling known endpoints.
- Capability matching maps task requirements to agent skills. For small catalogs, inject the full catalog into the model's context. For large catalogs, use embedding-based retrieval to narrow candidates before model-based selection.
- Dynamic plan generation produces an execution graph — steps, dependencies, parallel groups — from the model's reasoning about the task and the selected agents. The topology emerges at runtime.
- Failure recovery benefits from the registry: when an agent fails, the orchestrator queries for alternatives with similar capabilities and retries with a replacement from the wider agent pool.
- Health-aware registries filter unreliable agents before the orchestrator sees them, so routing decisions are based on what is actually working.
The trade-off is overhead versus flexibility. Every dynamic orchestration request pays the cost of discovery, matching, and plan generation before any agent starts working. That cost is justified when the agent pool is large, evolving, or contributed by independent teams — and wasteful when the same three agents handle every request.