[`agentic_crew.core.manager`](#module-agentic_crew.core.manager)

agentic_crew.core.manager

Hierarchical Manager Agent for crew orchestration.

This module provides a manager agent that can orchestrate multiple crews, enabling complex workflows with smart delegation, sequential and parallel execution, and human-in-the-loop checkpoints.

Example: ```python from agentic_crew.core.manager import ManagerAgent

class GameDevManager(ManagerAgent):
    '''Manager that orchestrates game development crews.'''

    def __init__(self):
        super().__init__(
            crews={
                "design": "gameplay_design",
                "implementation": "ecs_implementation",
                "assets": "asset_pipeline",
                "qa": "qa_validation",
            }
        )

    async def execute_workflow(self, task: str):
        # Sequential execution
        design_result = await self.delegate_async("design", task)

        # Parallel execution
        impl_result, asset_result = await asyncio.gather(
            self.delegate_async("implementation", design_result),
            self.delegate_async("assets", design_result)
        )

        # Final QA
        return await self.delegate_async("qa", {
            "implementation": impl_result,
            "assets": asset_result,
        })

## Module Contents

### Classes

| [`ManagerAgent`](#agentic_crew.core.manager.ManagerAgent)   | Base class for hierarchical manager agents.   |
|-------------------------------------------------------------|-----------------------------------------------|

### Data

| [`logger`](#agentic_crew.core.manager.logger)   |    |
|-------------------------------------------------|----|

### API

### agentic_crew.core.manager.logger

‘getLogger(…)’

### *class* agentic_crew.core.manager.ManagerAgent(crews: [dict](https://docs.python.org/3/library/stdtypes.html#dict)[[str](https://docs.python.org/3/library/stdtypes.html#str), [str](https://docs.python.org/3/library/stdtypes.html#str)], package_name: [str](https://docs.python.org/3/library/stdtypes.html#str) | [None](https://docs.python.org/3/library/constants.html#None) = None, workspace_root: [pathlib.Path](https://docs.python.org/3/library/pathlib.html#pathlib.Path) | [None](https://docs.python.org/3/library/constants.html#None) = None)

Base class for hierarchical manager agents.

A manager agent orchestrates multiple specialized crews to accomplish
complex tasks that require coordination between different domains or phases.

Attributes:
crews: Dict mapping crew role names to crew names in packages.
package_name: Optional package name if all crews are in one package.
workspace_root: Optional workspace root for crew discovery.

### Initialization

Initialize the manager agent.

Args:
crews: Dict mapping role names to crew names (e.g., {“design”: “game_design”}).
package_name: Optional package name if all crews are in the same package.
workspace_root: Optional workspace root for discovering packages.

#### \_get_packages() → [dict](https://docs.python.org/3/library/stdtypes.html#dict)[[str](https://docs.python.org/3/library/stdtypes.html#str), [pathlib.Path](https://docs.python.org/3/library/pathlib.html#pathlib.Path)]

Get discovered packages, using cache if available.

#### delegate(crew_role: [str](https://docs.python.org/3/library/stdtypes.html#str), inputs: [dict](https://docs.python.org/3/library/stdtypes.html#dict)[[str](https://docs.python.org/3/library/stdtypes.html#str), [Any](https://docs.python.org/3/library/typing.html#typing.Any)] | [str](https://docs.python.org/3/library/stdtypes.html#str), framework: [str](https://docs.python.org/3/library/stdtypes.html#str) | [None](https://docs.python.org/3/library/constants.html#None) = None) → [str](https://docs.python.org/3/library/stdtypes.html#str)

Delegate a task to a specific crew synchronously.

Args:
crew_role: Role name from the crews dict (e.g., “design”).
inputs: Input dict or string to pass to the crew.
framework: Optional framework override.

Returns:
Crew output as a string.

Raises:
ValueError: If crew_role not found in crews mapping.

#### *async* delegate_async(crew_role: [str](https://docs.python.org/3/library/stdtypes.html#str), inputs: [dict](https://docs.python.org/3/library/stdtypes.html#dict)[[str](https://docs.python.org/3/library/stdtypes.html#str), [Any](https://docs.python.org/3/library/typing.html#typing.Any)] | [str](https://docs.python.org/3/library/stdtypes.html#str), framework: [str](https://docs.python.org/3/library/stdtypes.html#str) | [None](https://docs.python.org/3/library/constants.html#None) = None) → [str](https://docs.python.org/3/library/stdtypes.html#str)

Delegate a task to a specific crew asynchronously.

This runs the crew in a thread pool to avoid blocking the event loop.

Args:
crew_role: Role name from the crews dict (e.g., “design”).
inputs: Input dict or string to pass to the crew.
framework: Optional framework override.

Returns:
Crew output as a string.

#### *async* delegate_parallel(delegations: [list](https://docs.python.org/3/library/stdtypes.html#list)[[tuple](https://docs.python.org/3/library/stdtypes.html#tuple)[[str](https://docs.python.org/3/library/stdtypes.html#str), [dict](https://docs.python.org/3/library/stdtypes.html#dict)[[str](https://docs.python.org/3/library/stdtypes.html#str), [Any](https://docs.python.org/3/library/typing.html#typing.Any)] | [str](https://docs.python.org/3/library/stdtypes.html#str)]], framework: [str](https://docs.python.org/3/library/stdtypes.html#str) | [None](https://docs.python.org/3/library/constants.html#None) = None) → [list](https://docs.python.org/3/library/stdtypes.html#list)[[str](https://docs.python.org/3/library/stdtypes.html#str)]

Delegate tasks to multiple crews in parallel.

Args:
delegations: List of (crew_role, inputs) tuples.
framework: Optional framework override for all crews.

Returns:
List of crew outputs in the same order as delegations.

Example:
`python     results = await manager.delegate_parallel([         ("design", "Create game concept"),         ("assets", "Generate placeholder assets"),     ])     design_result, assets_result = results     `

#### delegate_sequential(delegations: [list](https://docs.python.org/3/library/stdtypes.html#list)[[tuple](https://docs.python.org/3/library/stdtypes.html#tuple)[[str](https://docs.python.org/3/library/stdtypes.html#str), [dict](https://docs.python.org/3/library/stdtypes.html#dict)[[str](https://docs.python.org/3/library/stdtypes.html#str), [Any](https://docs.python.org/3/library/typing.html#typing.Any)] | [str](https://docs.python.org/3/library/stdtypes.html#str)]], framework: [str](https://docs.python.org/3/library/stdtypes.html#str) | [None](https://docs.python.org/3/library/constants.html#None) = None) → [list](https://docs.python.org/3/library/stdtypes.html#list)[[str](https://docs.python.org/3/library/stdtypes.html#str)]

Delegate tasks to multiple crews sequentially.

Args:
delegations: List of (crew_role, inputs) tuples.
framework: Optional framework override for all crews.

Returns:
List of crew outputs in the same order as delegations.

Example:
`python     results = manager.delegate_sequential([         ("design", "Create game concept"),         ("implementation", "Implement the design"),         ("qa", "Test the implementation"),     ])     `

#### checkpoint(message: [str](https://docs.python.org/3/library/stdtypes.html#str), result: [Any](https://docs.python.org/3/library/typing.html#typing.Any), auto_approve: [bool](https://docs.python.org/3/library/functions.html#bool) = False) → [tuple](https://docs.python.org/3/library/stdtypes.html#tuple)[[bool](https://docs.python.org/3/library/functions.html#bool), [Any](https://docs.python.org/3/library/typing.html#typing.Any)]

Create a human-in-the-loop checkpoint.

Args:
message: Message to display to the human reviewer.
result: Current result to review.
auto_approve: If True, automatically approve without waiting.

Returns:
Tuple of (approved, result). If not approved, result may be modified.

Note:
In this base implementation, checkpoints are logged but auto-approved.
Subclasses can override to implement actual HITL workflows.

#### *abstractmethod async* execute_workflow(task: [str](https://docs.python.org/3/library/stdtypes.html#str), \*\*kwargs: [Any](https://docs.python.org/3/library/typing.html#typing.Any)) → [str](https://docs.python.org/3/library/stdtypes.html#str)

Execute the manager’s workflow.

This is the main entry point for the manager agent. Subclasses should
override this method to implement their specific orchestration logic.

Args:
task: The main task to accomplish.
\*\*kwargs: Additional keyword arguments.

Returns:
Final result as a string.

Raises:
NotImplementedError: If not overridden by subclass.