> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# 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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
subgraph "Spawn & Announce Pattern"
A[📋 Parent Agent] --> B[🚀 Spawn Sub-Agents]
B --> C[⚡ Continue Work]
B --> D[🔄 Sub-Agent Execution]
D --> E[📢 Announce Completion]
E --> F[✅ Callback/Event]
F --> C
end
classDef input fill:#8B0000,stroke:#7C90A0,color:#fff
classDef process fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef output fill:#10B981,stroke:#7C90A0,color:#fff
class A input
class B,C,D,E process
class F output
```
## Quick Start
Fire-and-forget pattern with immediate feedback.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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
```
Spawn multiple sub-agents, do other work, then collect results.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant Parent as Parent Agent
participant Team as AgentTeam
participant Sub as Sub-Agent
participant Bus as EventBus
Parent->>Team: spawn_sub_agent(agent, task)
Team->>Sub: start(task)
Team-->>Parent: SpawnedSubAgent (immediate)
Parent->>Parent: Continue other work
Sub->>Sub: Execute task
Sub->>Team: announce_completion(result)
Team->>Bus: SUBAGENT_COMPLETED event
Bus-->>Parent: callback(event)
```
| 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.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph TB
Start[Need sub-agent execution?] --> Check{What coordination?}
Check -->|"Immediate feedback"| Callback[spawn_sub_agent + callback]
Check -->|"Batch processing"| Wait[spawn_sub_agent + wait_for_completions]
Check -->|"Event-driven"| Bus[spawn_sub_agent + EventBus subscription]
Check -->|"Async code"| Async[aspawn_sub_agent + await_for_completions]
classDef decision fill:#F59E0B,stroke:#7C90A0,color:#fff
classDef option fill:#10B981,stroke:#7C90A0,color:#fff
class Start,Check decision
class Callback,Wait,Bus,Async option
```
***
## Common Patterns
### Fire-and-Forget with Callback
Best for immediate processing of sub-agent results.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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())
```
***
## Best Practices
Prevent indefinite waiting by setting reasonable timeout values.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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()
```
Inside `async def` functions, use `aspawn_sub_agent` and `await_for_completions`.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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)
```
Completion callbacks should be lightweight to avoid blocking the event loop.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 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)
```
Attach metadata to track spawned sub-agents back to your domain context.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
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)
```
***
## Related
Core multi-agent coordination
Event-driven architecture