Generic coordinator-agent factory (Wave C).
:func:create_coordinator builds the domain-agnostic coordinator Agent from a :class:~strands_teams.usecase.UseCase: it validates the use case (AC-5-5), composes the system prompt from the use case's rendered template plus a framework-owned "teams protocol" partial, assembles the five generic team tools plus use_case.domain_tools() (AC-5-3), builds the tool registry that dispatch_team resolves node tool names against, and sets the coordinator's structured_output_model (AC-5-3).
Per-run shared state (layout, registry, node model config, timeouts, rubric) is returned as a :class:~strands_teams.teams.TeamInvocationState alongside the agent so the runner can pass it via coordinator.invoke_async(prompt, invocation_state=...) — no module-level globals (AC-8-4).
create_coordinator(
use_case, *, full_mode=False, focus=None, hooks=None
)
Create the coordinator Agent and its per-run invocation state.
Parameters:
| Name | Type | Description | Default |
use_case | UseCase | The domain use case supplying tools, prompt, model, layout, rubric, and structured-output model. | required |
full_mode | bool | Select full-mode reasoning effort and timeouts when True. | False |
focus | str | None | Optional focus area (the use case bakes this into the kickoff prompt built by the runner; kept for signature symmetry). | None |
hooks | list[HookProvider] | None | Optional HookProvider list added to the coordinator agent. | None |
Returns:
| Type | Description |
Agent | A (agent, invocation_state) tuple. The runner invokes the agent with |
TeamInvocationState | coordinator.invoke_async(prompt, invocation_state=invocation_state).
|
Raises:
Source code in packages/strands-teams/src/strands_teams/coordinator.py
| def create_coordinator(
use_case: UseCase,
*,
full_mode: bool = False,
focus: str | None = None, # noqa: ARG001 — reserved for use-case kickoff; prompt handles focus
hooks: list[HookProvider] | None = None,
) -> tuple[Agent, TeamInvocationState]:
"""Create the coordinator ``Agent`` and its per-run invocation state.
Args:
use_case: The domain use case supplying tools, prompt, model, layout,
rubric, and structured-output model.
full_mode: Select full-mode reasoning effort and timeouts when True.
focus: Optional focus area (the use case bakes this into the kickoff
prompt built by the runner; kept for signature symmetry).
hooks: Optional ``HookProvider`` list added to the coordinator agent.
Returns:
A ``(agent, invocation_state)`` tuple. The runner invokes the agent with
``coordinator.invoke_async(prompt, invocation_state=invocation_state)``.
Raises:
MissingUseCaseMemberError: If the use case omits a required member (AC-5-5).
"""
validate_use_case(use_case)
settings = use_case.settings
domain_tools = use_case.domain_tools()
team_tools = list(TEAM_TOOLS)
tools = [*team_tools, *domain_tools]
tool_registry = _build_tool_registry(team_tools, domain_tools)
node_model_config = _node_model_config(use_case, full_mode=full_mode)
if full_mode:
execution_timeout = float(settings.full_team_execution_timeout)
node_timeout = float(settings.full_team_node_timeout)
else:
execution_timeout = float(settings.team_execution_timeout)
node_timeout = float(settings.team_node_timeout)
invocation_state: TeamInvocationState = {
"layout": use_case.layout,
"tool_registry": tool_registry,
"node_model_config": node_model_config,
"execution_timeout": execution_timeout,
"node_timeout": node_timeout,
"use_case_name": use_case.name,
"rubric": use_case.rubric,
}
system_prompt = _compose_system_prompt(use_case)
model = BedrockModel(**_coordinator_model_config(node_model_config))
conversation_manager = SummarizingConversationManager(summary_ratio=0.3, preserve_recent_messages=10)
agent = Agent(
name="coordinator",
system_prompt=system_prompt,
model=model,
tools=tools,
structured_output_model=use_case.structured_output_model,
callback_handler=None,
conversation_manager=conversation_manager,
)
# The Agent's process_tools() expands ToolProvider instances (e.g. MCPClient
# for GitNexus/context7) into individual tool objects in its registry. Merge
# that expanded registry over the static one so dispatch_team resolves MCP
# tool names like "gitnexus_query" to real tool objects instead of dropping
# them as unknown strings.
invocation_state["tool_registry"] = {**tool_registry, **dict(agent.tool_registry.registry)}
if hooks:
for hook in hooks:
agent.hooks.add_hook(hook)
logger.info(
f"Created coordinator for use_case={use_case.name!r} with {len(tools)} tools "
f"({len(team_tools)} team + {len(domain_tools)} domain), full_mode={full_mode}",
)
return agent, invocation_state
|