runner

code_context_agent.agent.runner ¶

Agent runner with event streaming and display.

This module provides functions to run the analysis agent and stream events to consumers for display or further processing.

AnalysisContext ¶

Bases: BaseModel

Container for analysis components and configuration.

StreamResult ¶

Bases: BaseModel

Result of streaming analysis execution.

run_analysis `async` ¶

run_analysis(
    repo_path,
    output_dir=None,
    focus=None,
    consumer=None,
    quiet=False,
    issue_context=None,
    since_context=None,
    mode="standard",
)

Run code context analysis on a repository.

This function orchestrates the analysis by: 1. Creating an agent with tools, prompt, hooks, and structured output 2. Wrapping it with ag-ui-strands for typed event streaming 3. Streaming events to the consumer for display 4. Returning analysis results

Parameters:

Name	Type	Description	Default
`repo_path`	`str \| Path`	Path to the repository to analyze.	required
`output_dir`	`str \| Path \| None`	Output directory for context files. Defaults to repo/.code-context	`None`
`focus`	`str \| None`	Optional focus area to steer analysis (e.g., "authentication", "API layer").	`None`
`consumer`	`EventConsumer \| None`	Event consumer for display. Defaults to RichEventConsumer.	`None`
`quiet`	`bool`	If True and no consumer, use QuietConsumer.	`False`
`issue_context`	`str \| None`	Optional XML-wrapped issue context for issue-focused analysis.	`None`
`since_context`	`str \| None`	Optional XML-wrapped incremental analysis context from --since.	`None`
`mode`	`str`	Analysis mode ("standard", "full", "full+focus", "focus", "incremental").	`'standard'`

Returns:

Type	Description
`dict[str, Any]`	Dict with analysis status and output paths.

Source code in src/code_context_agent/agent/runner.py

async def run_analysis(
    repo_path: str | Path,
    output_dir: str | Path | None = None,
    focus: str | None = None,
    consumer: EventConsumer | None = None,
    quiet: bool = False,
    issue_context: str | None = None,
    since_context: str | None = None,
    mode: str = "standard",
) -> dict[str, Any]:
    """Run code context analysis on a repository.

    This function orchestrates the analysis by:
    1. Creating an agent with tools, prompt, hooks, and structured output
    2. Wrapping it with ag-ui-strands for typed event streaming
    3. Streaming events to the consumer for display
    4. Returning analysis results

    Args:
        repo_path: Path to the repository to analyze.
        output_dir: Output directory for context files. Defaults to repo/.code-context
        focus: Optional focus area to steer analysis (e.g., "authentication", "API layer").
        consumer: Event consumer for display. Defaults to RichEventConsumer.
        quiet: If True and no consumer, use QuietConsumer.
        issue_context: Optional XML-wrapped issue context for issue-focused analysis.
        since_context: Optional XML-wrapped incremental analysis context from --since.
        mode: Analysis mode ("standard", "full", "full+focus", "focus", "incremental").

    Returns:
        Dict with analysis status and output paths.
    """
    # Setup phase
    context = _setup_analysis_context(repo_path, output_dir, consumer, quiet, mode=mode)

    # Build prompt
    prompt = _build_analysis_prompt(context.repo, context.output, focus, issue_context, since_context)

    # Execution phase
    try:
        stream_result = await _execute_analysis_stream(context, prompt)
    finally:
        await _cleanup_context(context)

    # Build final result
    context_path = context.output / "CONTEXT.md"

    return {
        "status": stream_result.status,
        "error": stream_result.error_message,
        "exceeded_limit": stream_result.exceeded_limit,
        "turn_count": stream_result.turn_count,
        "duration_seconds": stream_result.duration_seconds,
        "repo_path": str(context.repo),
        "output_dir": str(context.output),
        "context_path": str(context_path) if context_path.exists() else None,
    }

run_analysis_sync ¶

run_analysis_sync(repo_path, output_dir=None, quiet=False)

Synchronous wrapper for run_analysis.