Skip to main content
Multi-Agent Hooks let you run custom code when tasks start, complete, or when you need custom completion detection in multi-agent workflows. 
from praisonaiagents import Agent, Task, PraisonAIAgents, MultiAgentHooksConfig

def on_task_start(task):
    print(f"Starting: {task.name}")

def on_task_complete(task, result):
    print(f"Done: {task.name}")

agent = Agent(name="Researcher", instructions="Research topics thoroughly.")
task = Task(description="Research quantum computing", agent=agent)

workflow = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    hooks=MultiAgentHooksConfig(
        on_task_start=on_task_start,
        on_task_complete=on_task_complete,
    ),
)
workflow.start()
The user registers hooks; the workflow fires them as each task starts and completes.

​
Quick Start

1

Task Lifecycle Logging

Log every task transition with on_task_start and on_task_complete:
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import MultiAgentHooksConfig
import time

start_times = {}

def on_start(task):
    start_times[task.name] = time.time()
    print(f"[START] {task.name}")

def on_complete(task, result):
    duration = time.time() - start_times.get(task.name, time.time())
    print(f"[DONE]  {task.name} ({duration:.1f}s)")

researcher = Agent(name="Researcher", instructions="Research topics thoroughly.")
writer = Agent(name="Writer", instructions="Write clear summaries.")

research_task = Task(description="Research quantum computing", agent=researcher)
write_task = Task(description="Write a summary", agent=writer)

team = PraisonAIAgents(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    hooks=MultiAgentHooksConfig(
        on_task_start=on_start,
        on_task_complete=on_complete,
    )
)

team.start()
2

Custom Completion Checker

Override the default completion logic with your own validation:
from praisonaiagents import Agent, Task, PraisonAIAgents
from praisonaiagents import MultiAgentHooksConfig

def quality_checker(task, result):
    """Return True if output meets quality standards."""
    if len(result) < 100:
        print(f"Output too short ({len(result)} chars), retrying...")
        return False
    if "error" in result.lower():
        print("Output contains error, retrying...")
        return False
    return True

agent = Agent(name="Analyst", instructions="Provide detailed analysis.")
task = Task(description="Analyze the market trends for 2025", agent=agent)

team = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    hooks=MultiAgentHooksConfig(
        completion_checker=quality_checker,
        on_task_complete=lambda t, r: print(f"✅ {t.name} passed quality check"),
    ),
)
workflow.start()

​
How It Works

HookWhen it firesUse case
on_task_startBefore each task beginsLogging, resource allocation
on_task_completeAfter each task succeedsNotifications, result storage
completion_checkerAfter task, before marking doneQuality gates, validation

​
Configuration Options

MultiAgentHooksConfig SDK Reference

Full parameter reference for MultiAgentHooksConfig
OptionTypeDefaultDescription
on_task_startCallable | NoneNoneCalled before a task begins. Signature: (task) -> None
on_task_completeCallable | NoneNoneCalled after a task finishes. Signature: (task, result) -> None
completion_checkerCallable | NoneNoneCustom validator. Signature: (task, result) -> bool

​
Common Patterns

​
Pattern 1 — Logging workflow progress

from praisonaiagents import Agent, Task, PraisonAIAgents, MultiAgentHooksConfig
import time

start_times = {}

def log_start(task):
    start_times[task.name] = time.time()
    print(f"[START] {task.name}")

def log_complete(task, result):
    elapsed = time.time() - start_times.get(task.name, 0)
    print(f"[DONE] {task.name} in {elapsed:.1f}s")

agent = Agent(name="Worker", instructions="Complete tasks efficiently.")
task = Task(description="Analyze quarterly sales data", agent=agent)

workflow = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    hooks=MultiAgentHooksConfig(on_task_start=log_start, on_task_complete=log_complete),
)
workflow.start()

​
Pattern 2 — Quality gate with retry logic

from praisonaiagents import Agent, Task, PraisonAIAgents, MultiAgentHooksConfig

def length_checker(task, result):
    return len(result.raw) >= 200

agent = Agent(name="Analyst", instructions="Write comprehensive analyses.")
task = Task(description="Analyze the impact of remote work on productivity", agent=agent)

response = PraisonAIAgents(
    agents=[agent],
    tasks=[task],
    hooks=MultiAgentHooksConfig(completion_checker=length_checker),
).start()
print(response)

​
Best Practices

Hook callbacks run synchronously in the orchestrator loop. Expensive operations (database writes, HTTP requests) should be done asynchronously or offloaded to a background thread to avoid slowing down task execution.
The completion_checker is ideal for enforcing output quality requirements — minimum length, required fields, absence of error text. Return False to trigger a retry and True to accept the result.
Hook callbacks receive task references. Mutating task properties during execution can cause unexpected behavior. Use hooks for observation and notification, not for changing task state.

Hooks

Single-agent lifecycle hooks

Callbacks

Agent callback system

Multi-Agent Output

Control output verbosity across agents

Multi-Agent Execution

Configure iteration and retry limits