runner

strands_teams.runner ¶

Generic run loop for a strands_teams coordinator (Wave C).

:func:run_teams orchestrates a use-case run: it validates and builds the coordinator via :func:strands_teams.coordinator.create_coordinator, wires optional hook-driven display, invokes the coordinator with the per-run invocation_state (AC-8-1, AC-8-4), persists the structured output, and cleans up. Adapted from the pre-split code_context_agent.agent.runner skeleton (limits, Live wiring, structured-output persistence, _cleanup, StreamResult) but domain-agnostic — all domain vocabulary comes from the UseCase.

The hooks (:mod:strands_teams.hooks) and display (:mod:strands_teams.display) modules are owned by concurrent Wave C tasks; this module imports them lazily and tolerates their absence so it can be developed and tested independently. The expected names are create_all_hooks(use_case, *, full_mode, state, quiet) and the display types AgentDisplayState / RichEventConsumer / QuietConsumer.

StreamResult ¶

Bases: BaseModel

Result of a coordinator run.

run_teams `async` ¶

run_teams(
    use_case,
    *,
    repo_path,
    focus=None,
    extra_context=None,
    bundles_only=False,
    full_mode=False,
    quiet=False,
)

Run a use-case coordinator end to end.

Parameters:

Name	Type	Description	Default
`use_case`	`UseCase`	The domain use case to run.	required
`repo_path`	`str \| Path`	The target root the run operates on (passed to `pre_run`).	required
`focus`	`str \| None`	Optional focus area.	`None`
`extra_context`	`str \| None`	Optional free-form context for the kickoff prompt.	`None`
`bundles_only`	`bool`	Skip dispatch and regenerate artifacts from existing findings.	`False`
`full_mode`	`bool`	Use full-mode reasoning effort and timeouts.	`False`
`quiet`	`bool`	Suppress the Rich display.	`False`

Returns:

Type	Description
`dict[str, Any]`	Dict with run status, duration, output paths, and the structured output.

Source code in packages/strands-teams/src/strands_teams/runner.py

async def run_teams(
    use_case: UseCase,
    *,
    repo_path: str | Path,
    focus: str | None = None,
    extra_context: str | None = None,
    bundles_only: bool = False,
    full_mode: bool = False,
    quiet: bool = False,
) -> dict[str, Any]:
    """Run a use-case coordinator end to end.

    Args:
        use_case: The domain use case to run.
        repo_path: The target root the run operates on (passed to ``pre_run``).
        focus: Optional focus area.
        extra_context: Optional free-form context for the kickoff prompt.
        bundles_only: Skip dispatch and regenerate artifacts from existing findings.
        full_mode: Use full-mode reasoning effort and timeouts.
        quiet: Suppress the Rich display.

    Returns:
        Dict with run status, duration, output paths, and the structured output.
    """
    from .coordinator import create_coordinator

    repo = Path(repo_path).resolve()
    output_root = use_case.layout.output_root
    output_root.mkdir(parents=True, exist_ok=True)

    # Deterministic pre-run (e.g. indexing) unless we are only regenerating artifacts.
    if not bundles_only:
        use_case.pre_run(repo)

    state = _make_display_state(use_case, quiet=quiet, full_mode=full_mode)
    hooks = _make_hooks(use_case, full_mode=full_mode, state=state, quiet=quiet)
    coordinator, invocation_state = create_coordinator(use_case, full_mode=full_mode, focus=focus, hooks=hooks)

    prompt = use_case.kickoff_prompt(bundles_only=bundles_only, focus=focus, extra_context=extra_context)

    live = _start_live_display(use_case, state, full_mode=full_mode)
    try:
        stream_result = await _run_coordinator(coordinator, prompt, invocation_state)
    finally:
        _cleanup(live)

    _persist_structured_output(use_case, stream_result.structured_output)

    summary_path = use_case.layout.executive_summary_path()
    return {
        "status": stream_result.status,
        "error": stream_result.error_message,
        "duration_seconds": stream_result.duration_seconds,
        "repo_path": str(repo),
        "output_dir": str(output_root),
        "executive_summary_path": str(summary_path) if summary_path.exists() else None,
        "structured_output": stream_result.structured_output,
    }

run_teams_sync ¶

run_teams_sync(
    use_case,
    *,
    repo_path,
    focus=None,
    extra_context=None,
    bundles_only=False,
    full_mode=False,
    quiet=False,
)

Synchronous wrapper around :func:run_teams.