A crewai custom manager agent hierarchical process replaces the default manager with a purpose-built agent that controls task delegation, sequencing, and termination logic. Instead of relying on CrewAI’s built-in manager — which uses a fixed prompt and simple heuristics — you supply your own agent with custom instructions, tools, and decision-making behavior. This gives you deterministic control over how work flows through a multi-agent system.
How the hierarchical process works
In CrewAI’s hierarchical process, a single manager agent sits above a pool of worker agents. The manager receives the crew’s high-level goal, breaks it into subtasks, assigns each subtask to a worker, collects results, and decides whether to continue, reassign, or terminate. The default manager uses a generic system prompt that asks it to “delegate tasks to the most appropriate agent” and “ensure all tasks are completed.”
The flow looks like this:
User Goal → Manager Agent → Task Decomposition → Worker Assignment → Result Aggregation → Next DecisionThe manager operates in a loop. Each iteration, it sees the conversation history, the list of available workers (their roles, goals, and tools), and any prior task outputs. It then emits a delegation action: either assign a new task to a specific worker, request clarification, or signal completion.
The default manager is a black box. You cannot inspect its reasoning, adjust its delegation criteria, or inject domain-specific logic like “always route billing questions to the finance agent” or “escalate after three failed attempts.” That’s where a custom manager agent becomes necessary.
Why customize the manager
Three scenarios justify a crewai custom manager agent hierarchical setup:
1. Domain-specific routing rules
Your workers have overlapping capabilities. The default manager picks based on role descriptions alone. A custom manager can implement explicit routing: “If the task mentions PCI compliance, assign to security-auditor; if it mentions SQL optimization, assign to db-engineer.”
2. Complex termination conditions
The default manager stops when it “feels” the goal is met. A custom manager can enforce hard criteria: “All acceptance criteria checked,” “No critical findings remain,” “Budget threshold not exceeded.”
3. Observability and audit trails
You need to log every delegation decision with rationale for compliance or debugging. A custom manager can emit structured events at each step.
Building a custom manager agent
A custom manager is just a regular CrewAI Agent with allow_delegation=True and a carefully crafted system prompt. You pass it to the Crew constructor via the manager_agent parameter.
from crewai import Agent, Crew, Process, Task
from crewai.tools import BaseTool
from typing import List, Dict, Any
import json
class DelegationLogger(BaseTool):
name: str = "log_delegation"
def _run(self, worker: str, task: str, rationale: str) -> str:
entry = {
"timestamp": datetime.utcnow().isoformat(),
"worker": worker,
"task": task,
"rationale": rationale
}
# In production: write to your observability backend
print(f"DELEGATION: {json.dumps(entry)}")
return "Logged"
manager = Agent(
role="Project Manager",
goal="Decompose the user request into discrete tasks, assign each to the correct specialist, "
"track progress, and terminate only when all acceptance criteria are satisfied.",
backstory=(
"You are a technical program manager with 15 years of experience shipping "
"distributed systems. You know every team's strengths and you never delegate "
"vaguely. You require concrete deliverables and explicit acceptance criteria."
),
allow_delegation=True,
tools=[DelegationLogger()],
verbose=True,
# Critical: the system prompt must instruct the manager HOW to delegate
system_template="""
You are the manager of a specialist team. Your job is to complete the user's request
by delegating tasks to the right agents.
AVAILABLE AGENTS:
{{agents}}
CURRENT CONVERSATION:
{{conversation}}
RULES:
1. ALWAYS use the log_delegation tool before assigning a task. Include your rationale.
2. NEVER assign a task to more than one agent simultaneously.
3. If a task fails, you may retry ONCE with a different agent. Log the reason.
4. TERMINATE only when every acceptance criterion in the original request is met.
5. Output your decision as a single JSON object with keys: action, agent, task, rationale.
ACTIONS: "delegate" | "retry" | "complete" | "clarify"
"""
)The system_template uses CrewAI’s templating syntax. {{agents}} injects the worker roster; {{conversation}} injects history. The manager must output structured JSON so the framework can parse its decisions. CrewAI’s hierarchical process expects the manager to return a specific format — consult the version you’re using, but typically it looks for action, agent, task, and rationale fields.
Workers and tasks
Define workers with narrow, non-overlapping responsibilities. Overlap creates ambiguity the manager must resolve.
security_auditor = Agent(
role="Security Auditor",
goal="Identify security vulnerabilities in code and architecture",
backstory="You specialize in OWASP Top 10, threat modeling, and secure code review.",
tools=[StaticAnalysisTool(), DependencyScanner()],
allow_delegation=False,
verbose=True
)
db_engineer = Agent(
role="Database Engineer",
goal="Optimize queries, schema, and indexes for PostgreSQL workloads",
backstory="You have deep expertise in query planning, partitioning, and connection pooling.",
tools=[QueryAnalyzer(), IndexAdvisor()],
allow_delegation=False,
verbose=True
)
devops_engineer = Agent(
role="DevOps Engineer",
goal="Automate deployments, observe production systems, and manage infrastructure",
backstory="You build CI/CD pipelines, configure observability stacks, and handle incidents.",
tools=[TerraformPlanner(), LogQueryTool()],
allow_delegation=False,
verbose=True
)Tasks for a hierarchical crew are high-level. The manager decomposes them.
tasks = [
Task(
description=(
"Perform a comprehensive security and performance review of the payment service. "
"Deliverables: (1) vulnerability report with CVSS scores, (2) query optimization "
"recommendations with estimated latency improvement, (3) deployment risk assessment."
),
expected_output="JSON report with three sections: vulnerabilities, optimizations, deployment_risks",
agent=None # None = manager decides
)
]Assembling the crew
crew = Crew(
agents=[security_auditor, db_engineer, devops_engineer],
tasks=tasks,
process=Process.hierarchical,
manager_agent=manager,
verbose=True,
memory=True, # enables cross-task context sharing
max_iter=15, # safety cap on manager loop iterations
output_log_file="crew_execution.log"
)
result = crew.kickoff(inputs={"service": "payment-service", "environment": "staging"})
print(result)Key parameters:
manager_agent: your custom manager instancemax_iter: hard limit on delegation cycles — prevents infinite loopsmemory=True: lets workers see each other’s outputs via the shared conversation historyoutput_log_file: captures the full trace for debugging
Concrete example: incident response crew
Here’s a production-style pattern. An incident comes in; the manager triages, delegates investigation, coordinates mitigation, and verifies resolution.
from crewai import Agent, Crew, Process, Task
from datetime import datetime
from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional, List
import json
class IncidentSeverity(str, Enum):
SEV1 = "sev1" # customer-facing outage
SEV2 = "sev2" # degraded performance
SEV3 = "sev3" # minor issue
class IncidentContext(BaseModel):
incident_id: str
title: str
severity: IncidentSeverity
affected_services: List[str]
started_at: datetime
symptoms: List[str]
runbook_urls: List[str] = Field(default_factory=list)
class DelegationDecision(BaseModel):
action: str # delegate, escalate, mitigate, verify, close
assignee: str
task: str
rationale: str
expected_duration_minutes: int
class IncidentManager(Agent):
"""Custom manager with structured output and state machine logic."""
def __init__(self, **kwargs):
super().__init__(
role="Incident Commander",
goal="Resolve the incident with minimum customer impact. Follow the incident command system.",
backstory=(
"You are a certified Incident Commander. You run the incident bridge, "
"delegate investigation and mitigation tasks, track status on a public timeline, "
"and ensure post-incident review is scheduled. You never assume — you verify."
),
allow_delegation=True,
verbose=True,
**kwargs
)
def get_system_prompt(self) -> str:
return f"""
You are the Incident Commander for incident {{incident_id}}.
INCIDENT CONTEXT:
{json.dumps(self.incident_context.model_dump(), default=str, indent=2)}
AVAILABLE RESPONDERS:
{{agents}}
CURRENT TIMELINE:
{{conversation}}
YOUR AUTHORITY:
- You command all responders. They report to you.
- You communicate status to stakeholders every 15 minutes (SEV1) or 30 minutes (SEV2/3).
- You can escalate to on-call directors via the escalation tool.
DECISION FRAMEWORK (output JSON matching DelegationDecision):
1. TRIAGE: If root cause unknown → delegate investigation to relevant specialist.
2. MITIGATE: If root cause known and mitigation exists → delegate mitigation.
3. VERIFY: If mitigation deployed → delegate verification to independent responder.
4. ESCALATE: If no progress in 30 min (SEV1) or 60 min (SEV2) → escalate.
5. CLOSE: If verified resolved for 15 min → close incident, schedule postmortem.
RULES:
- One active delegation at a time. Wait for result before next decision.
- Log every decision with the log_delegation tool.
- Never delegate the same task to two responders.
- If a responder fails, reassign with context — don't retry blindly.
"""
# Responders with explicit capabilities
investigator = Agent(
role="Investigator",
goal="Determine root cause using logs, metrics, traces, and recent changes",
backstory="You are a systems engineer who reads flame graphs for fun. You know every service's failure modes.",
tools=[LogQueryTool(), MetricsQueryTool(), TraceAnalyzer(), RecentChangesTool()],
allow_delegation=False
)
mitigator = Agent(
role="Mitigator",
goal="Execute safe, reversible mitigations: rollback, feature flag, traffic shift, capacity add",
backstory="You have run the rollback playbook 200 times. You verify each step before proceeding.",
tools=[RollbackTool(), FeatureFlagTool(), TrafficShiftTool(), CapacityTool()],
allow_delegation=False
)
verifier = Agent(
role="Verifier",
goal="Independently confirm mitigation restored service health",
backstory="You trust but verify. You check customer-facing endpoints, error rates, and latency percentiles.",
tools=[HealthCheckTool(), SyntheticMonitorTool(), ErrorBudgetTool()],
allow_delegation=False
)
communicator = Agent(
role="Communicator",
goal="Maintain the public incident timeline and stakeholder updates",
backstory="You write clear, blameless updates every 15 minutes. Engineers love your summaries.",
tools=[StatusPageTool(), SlackAnnouncementTool(), TimelineTool()],
allow_delegation=False
)
# Instantiate manager with incident context
incident = IncidentContext(
incident_id="INC-2024-03-15-0042",
title="Payment service 5xx spike after deploy v2.3.1",
severity=IncidentSeverity.SEV1,
affected_services=["payment-api", "billing-worker"],
started_at=datetime.utcnow(),
symptoms=["5xx rate > 5%", "p99 latency > 10s", "checkout failures"],
runbook_urls=["https://runbooks.company.com/payment-5xx"]
)
manager = IncidentManager()
manager.incident_context = incident
crew = Crew(
agents=[investigator, mitigator, verifier, communicator],
tasks=[
Task(
description=(
f"Resolve incident {incident.incident_id}: {incident.title}. "
"Follow the incident command process. Deliver: root cause, mitigation applied, "
"verification evidence, and postmortem scheduled."
),
expected_output="Incident resolution report with timeline, root cause, and action items",
agent=None
)
],
process=Process.hierarchical,
manager_agent=manager,
verbose=True,
max_iter=20,
memory=True
)
result = crew.kickoff()This pattern encodes the incident command system directly into the manager’s prompt. The structured DelegationDecision output (enforced by the prompt) makes the manager’s reasoning auditable. Each responder has disjoint tools — no ambiguity about who does what.
Common misconceptions
“The manager is just another worker”
False. The manager never executes the actual work. It only decides who does what and when to stop. Giving the manager tools that perform domain work (e.g., a QueryDatabase tool) defeats the architecture. The manager’s tools should be meta-tools: logging, escalation, status reporting, clarification requests.
“Hierarchical is always better than sequential”
Sequential crews execute a fixed task list in order. Hierarchical crews dynamically decide the task graph. Use sequential when the workflow is deterministic (e.g., “extract → transform → load”). Use hierarchical when the path depends on intermediate discoveries (e.g., “investigate → maybe mitigate → verify → maybe escalate”). Hierarchical adds latency — each delegation round trips through the LLM. Don’t pay that cost if you don’t need it.
“Custom manager prompts are set-and-forget”
Manager prompts are the most sensitive prompt in your system. A single ambiguous instruction (“delegate appropriately”) causes cascading misrouting. Version your manager prompts. Test them with adversarial inputs: ambiguous goals, missing context, worker failures. Treat them like infrastructure code — review, CI, rollback capability.
“Memory solves context sharing automatically”
memory=True shares the conversation transcript. It does not create a structured shared state. If worker A produces a root_cause object, worker B sees it as text in the history. For reliable handoffs, have the manager explicitly pass structured context in the task description: “Investigate using this root cause hypothesis: {{root_cause}}.” Or use a shared artifact store (Redis, S3) that workers read/write via tools.
“Max iterations prevents all infinite loops”
max_iter caps the manager’s decision cycles. It does not prevent a worker from looping internally, or a tool from hanging. Set timeouts on tools. Use CrewAI’s task_callback to monitor progress and inject a “stuck” signal the manager can act on.
When to reach for this pattern
You need a crewai custom manager agent hierarchical setup when:
- Routing logic exceeds what role descriptions can express
- Termination criteria are multi-condition and explicit
- You need an audit trail of every delegation decision
- The workflow graph is discovered at runtime, not known at design time
Otherwise, a sequential crew with a well-ordered task list is simpler, faster, and easier to debug. Start there. Graduate to hierarchical when the complexity justifies it.