Spawn & Announce

Non-blocking multi-agent orchestration with parallel execution

Spawn & Announce lets a parent agent delegate work to sub-agents without blocking — fire tasks in parallel and collect results via callbacks or the event bus.

from praisonaiagents import Agent, AgentTeam, Task

team = AgentTeam(agents=[], name="research_team")

researcher = Agent(name="Researcher", instructions="Research the topic briefly.")
task = Task(description="What is quantum computing?")

team.spawn_sub_agent(researcher, task)
# Parent continues immediately — no blocking

The user delegates parallel work; the parent spawns sub-agents and keeps running without blocking.

Quick Start

1. Simple Spawn with Callback

   Fire-and-forget pattern with immediate feedback.

   from praisonaiagents import Agent, AgentTeam, Task
   
   team = AgentTeam(agents=[], name="my_team")
   
   researcher = Agent(name="Researcher", instructions="Research the topic briefly.")
   task = Task(description="What is quantum computing?")
   
   def on_done(event):
       print(f"Sub-agent {event.agent_id} done: {event.result}")
   
   team.spawn_sub_agent(researcher, task, completion_callback=on_done)
   # Parent continues immediately — no blocking
   
2. Parallel Fan-out Pattern

   Spawn multiple sub-agents, do other work, then collect results.

   from praisonaiagents import Agent, AgentTeam, Task
   
   team = AgentTeam(agents=[], name="research_team")
   
   agents = [
       Agent(name="Researcher", instructions="Research briefly."),
       Agent(name="Writer", instructions="Write briefly."),
       Agent(name="Analyst", instructions="Analyze briefly."),
   ]
   tasks = [
       Task(description="Research AI trends"),
       Task(description="Write a tagline about AI"),
       Task(description="List one risk of AI"),
   ]
   
   for agent, task in zip(agents, tasks):
       team.spawn_sub_agent(agent, task)
   
   # Do other work here while sub-agents run in parallel...
   
   completions = team.wait_for_completions(timeout=60.0)
   for c in completions:
       print(c.agent_id, "->", c.result)
   

---

How It Works

The spawn-announce pattern uses an event-driven architecture to coordinate between parent and sub-agents.

Component	Role
Parent Agent	Spawns sub-agents, continues work non-blocking
Sub-Agent	Executes task, announces completion when done
AgentTeam	Orchestrates spawning and event coordination
EventBus	Delivers completion events to callbacks

---

Choosing Your Pattern

Different use cases require different coordination patterns.

---

Common Patterns

Fire-and-Forget with Callback

Best for immediate processing of sub-agent results.

from praisonaiagents import Agent, AgentTeam, Task

team = AgentTeam(agents=[], name="callback_team")

def process_result(event):
    if event.success:
        print(f"✅ {event.agent_id}: {event.result}")
    else:
        print(f"❌ {event.agent_id}: {event.error}")

team.spawn_sub_agent(
    Agent(name="Worker", instructions="Answer in one line."),
    Task(description="Say hello."),
    completion_callback=process_result
)

Parallel Fan-out with Join

Scale work across multiple sub-agents, then combine results.

from praisonaiagents import Agent, AgentTeam, Task

team = AgentTeam(agents=[], name="parallel_team")
topics = ["AI", "blockchain", "quantum computing"]

# Spawn all sub-agents
for topic in topics:
    agent = Agent(name=f"{topic}_expert", instructions=f"Expert in {topic}.")
    task = Task(description=f"Explain {topic} in one sentence.")
    team.spawn_sub_agent(agent, task)

# Collect all results
results = team.wait_for_completions(timeout=120.0)
summary = "\n".join([f"• {r.result}" for r in results if r.success])
print("Research Summary:\n" + summary)

Event-Driven Coordination

React to sub-agent events without callbacks.

from praisonaiagents import Agent, AgentTeam, Task
from praisonaiagents.bus import EventBus
from praisonaiagents.bus.event import EventType

bus = EventBus()

def on_subagent_event(event):
    print(f"[{event.type}] {event.data['agent_id']} - {event.data.get('result', 'spawned')}")

bus.subscribe(on_subagent_event, [
    EventType.SUBAGENT_SPAWNED.value,
    EventType.SUBAGENT_COMPLETED.value,
    EventType.SUBAGENT_ERROR.value,
])

team = AgentTeam(agents=[], name="event_team")
team._event_bus = bus  # share bus

team.spawn_sub_agent(
    Agent(name="Worker", instructions="Answer briefly."),
    Task(description="What is 2+2?"),
)

---

Configuration

spawn_sub_agent Parameters

Parameter	Type	Default	Description
agent	Agent	required	The agent that will execute the task.
task	Task or str	required	Task to run. If it has .description, that string is sent to agent.start().
completion_callback	Callable[[SubAgentCompletionEvent], Any]	None	Called once the sub-agent finishes.
metadata	Dict[str, Any]	None	Free-form metadata stored on the SpawnedSubAgent.

announce_completion Parameters

Parameter	Type	Default	Description
agent_id	str	required	ID returned by spawn_sub_agent.
task_id	str	required	Task ID from SpawnedSubAgent.task_id.
result	Any	required	The sub-agent’s output.
success	bool	True	Whether the task succeeded.
error	Optional[str]	None	Error message if success=False.
metadata	Dict[str, Any]	None	Free-form metadata on the completion event.

wait_for_completions Parameters

Parameter	Type	Default	Description
timeout	Optional[float]	None	Max seconds to wait. None = wait forever (not recommended).
agent_ids	Optional[List[str]]	None	Specific IDs to wait for. None = wait for all currently spawned.

---

Async Usage

For async code, use the async variants with proper event-driven waiting.

import asyncio
from praisonaiagents import Agent, AgentTeam, Task

async def main():
    team = AgentTeam(agents=[], name="async_team")
    
    agent = Agent(name="Researcher", instructions="Research briefly.")
    task = Task(description="What is RAG?")

    await team.aspawn_sub_agent(agent, task)
    completions = await team.await_for_completions(timeout=60.0)
    
    print(f"Result: {completions[0].result}")

asyncio.run(main())

aspawn_sub_agent + await_for_completions are reliable — a sub-agent running on a background thread wakes the awaiter as soon as it finishes, and mixing sync and async spawn calls on the same AgentTeam is safe because both paths share one threading.RLock.

Fixed in PraisonAI PR #4887. Prior to this release the async spawn/completion path had three defects reachable from ordinary usage: (1) await_for_completions() could hang until timeout — or forever with timeout=None — because check_completions() called asyncio.get_running_loop() from a background spawn thread, which raised RuntimeError that was silently swallowed, so the awaiter never woke even after the sub-agent finished; (2) the sync path used threading.RLock while the async path used asyncio.Lock, which excludes nothing against a real OS thread, so mixing sync and async spawn on one AgentTeam was a genuine data race; (3) fire-and-forget asyncio.create_task(...) results were not stored, so the event loop’s weak reference let a sub-agent run be garbage-collected mid-run. After this release, await_for_completions captures the running loop in its own coroutine frame and closes over it in check_completions, both paths share one threading.RLock, and background spawn tasks are held in a strong-reference set (self._bg_spawn_tasks) so they cannot be collected. No API surface changed.

await_for_completions subscribes to the event bus before running the initial state check, so a sub-agent that finishes between the check and the subscribe no longer causes a lost signal.

---

Best Practices

Always Set Timeouts

Prevent indefinite waiting by setting reasonable timeout values.

# Good: Set timeout to prevent hanging
completions = team.wait_for_completions(timeout=60.0)

# Bad: No timeout - may wait forever
completions = team.wait_for_completions()

Use Async Variants for Async Code

Inside async def functions, use aspawn_sub_agent and await_for_completions.

async def process_requests():
    # Good: Async variant
    await team.aspawn_sub_agent(agent, task)
    results = await team.await_for_completions(timeout=30.0)
    
    # Bad: Blocking sync calls in async context
    team.spawn_sub_agent(agent, task)
    results = team.wait_for_completions(timeout=30.0)

Keep Callbacks Fast

Completion callbacks should be lightweight to avoid blocking the event loop.

# Good: Fast processing
def quick_callback(event):
    print(f"Done: {event.agent_id}")
    results_queue.put(event.result)

# Bad: Heavy processing in callback
def slow_callback(event):
    heavy_analysis(event.result)  # Blocks other events
    expensive_db_save(event.result)

Use Metadata for Correlation

Attach metadata to track spawned sub-agents back to your domain context.

metadata = {"request_id": "req_123", "user_id": "user_456"}
spawned = team.spawn_sub_agent(agent, task, metadata=metadata)

# Later, correlate completion back to request
def on_completion(event):
    request_id = event.metadata.get("request_id")
    notify_user_of_completion(request_id, event.result)

Mixing sync and async spawn on the same team is safe

Since PR #4887, spawn_sub_agent (sync) and aspawn_sub_agent / aannounce_completion / await_for_completions (async) share one threading.RLock, so calling both on the same AgentTeam no longer races on shared completion state.

# Both paths guard the same state — safe to interleave
team.spawn_sub_agent(agent_a, task_a)
await team.aspawn_sub_agent(agent_b, task_b)
completions = await team.await_for_completions(timeout=60.0)

---

Related

Subagent Delegation

Core multi-agent coordination

Event Bus

Event-driven architecture