Server Tools¶

def main() -> None:
    """Start the FastMCP server with stdio transport.

    This mirrors ``mcp_zen_of_languages.__main__.main`` so packaging can expose
    a dedicated ``mcp-zen-of-languages-server`` console script without adding a
    separate runtime path.
    """
    mcp.run()

@mcp.resource(
    "zen://config",
    name="zen_config_resource",
    title="Zen config resource",
    description="Read-only resource exposing current configuration and active overrides.",
    icons=RESOURCE_ICONS,
)
def config_resource() -> "ConfigStatus":
    """Return current configuration status as a read-only MCP resource."""
    return _build_config_status()

@mcp.resource(
    "zen://rules/{language}",
    name="zen_rules_resource",
    title="Zen rules resource",
    description="Read-only resource exposing canonical zen principles for a language.",
    icons=RESOURCE_ICONS,
)
def rules_resource(language: str) -> LanguageZenPrinciples:
    """Return canonical zen principles for the requested language key."""
    zen = get_language_zen(_canonical_language(language))
    if zen is None:
        msg = f"Unsupported language '{language}'."
        raise ValueError(msg)
    return zen

@mcp.resource(
    "zen://languages",
    name="zen_languages_resource",
    title="Zen languages resource",
    description="Read-only resource listing language principle and detector coverage counts.",
    icons=RESOURCE_ICONS,
)
def languages_resource() -> LanguagesResource:
    """Return supported languages with principle and detector counts."""
    from mcp_zen_of_languages.analyzers.registry import REGISTRY

    entries: list[LanguageCoverage] = []
    for language in get_all_languages():
        detectors = [
            meta.detector_id
            for meta in REGISTRY.items()
            if meta.language in [language, "any"]
        ]
        zen = get_language_zen(language)
        entries.append(
            LanguageCoverage(
                language=language,
                principles=len(zen.principles) if zen else 0,
                detectors=len(detectors),
            ),
        )
    return LanguagesResource(languages=entries)

@mcp.prompt(
    name="zen_remediation_prompt",
    title="Zen remediation prompt",
    description="Generate a remediation prompt scaffold for violations in a language.",
    icons=PROMPT_RESOURCE_ICONS,
    tags={"prompts", "remediation"},
)
def remediation_prompt(language: str, violations: str) -> str:
    """Build a typed remediation prompt template for MCP clients."""
    return (
        "Context: Remediate zen violations in a codebase.\n"
        f"Language: {language}\n"
        "Goal: Produce precise, testable fixes aligned with zen principles.\n"
        f"Violations:\n{violations}\n"
        "Requirements:\n"
        "1. Prioritize highest severity findings first.\n"
        "2. Provide before/after guidance for each fix.\n"
        "3. Include verification steps for each remediation."
    )

@mcp.tool(
    name="detect_languages",
    title="Detect languages",
    description="Return supported language list for analysis.",
    tags={"languages", "metadata"},
    annotations=READONLY_ANNOTATIONS,
    output_schema=_output_schema(LanguagesResult),
)
async def detect_languages(repo_path: str) -> LanguagesResult:
    """Return the language identifiers listed in the active ``zen-config.yaml``.

    Unlike heuristic language-detection libraries, this tool does **not**
    scan file extensions or parse shebangs.  It simply reflects the
    ``languages`` key from the configuration that ``CONFIG`` loaded at
    server startup, giving callers a predictable, deterministic list they
    can iterate over when orchestrating multi-language analysis runs.

    Args:
        repo_path (str): Workspace root passed by the MCP client — reserved
            for future per-repo config resolution but currently unused.

    Returns:
        LanguagesResult: LanguagesResult wrapping the list of language strings declared in
        ``zen-config.yaml`` (e.g. ``["python", "typescript", "go"]``).

    Example:
        ```python
        result = await detect_languages("/home/dev/myproject")
        for lang in result.languages:
            await analyze_zen_violations(code, lang)
        ```

    See Also:
        [`get_supported_languages`][mcp_zen_of_languages.server.get_supported_languages]:
            Lists languages that have registered detectors, rather than
            configured languages.

    """
    from mcp_zen_of_languages.models import LanguagesResult

    _ = repo_path
    return LanguagesResult(languages=CONFIG.languages)

@mcp.tool(
    name="analyze_zen_violations",
    version=ANALYZE_ZEN_VIOLATIONS_VERSION,
    title="Analyze zen violations",
    description="Analyze a code snippet against zen rules and return analysis results.",
    icons=ANALYSIS_TOOL_ICONS,
    tags={"analysis", "zen", "snippet"},
    annotations=READONLY_ANNOTATIONS,
    output_schema=_output_schema(AnalysisResult),
)
async def analyze_zen_violations(  # noqa: PLR0913
    code: str,
    language: str,
    severity_threshold: int | None = None,
    perspective: PerspectiveMode = PerspectiveMode.ALL,
    project_as: str | None = None,
    *,
    enable_external_tools: bool = False,
    allow_temporary_runners: bool = False,
) -> AnalysisResult:
    """Run v1.0 snippet analysis.

    Args:
        code (str): Source code to analyse.
        language (str): Programming language identifier.
        severity_threshold (int | None, optional): Minimum severity to include. Default to None.
        perspective (PerspectiveMode, optional): Requested analysis perspective. Default to ``PerspectiveMode.ALL``.
        project_as (str | None, optional): Projection-family target when ``perspective`` is ``projection``.
        enable_external_tools (bool, optional): Opt-in execution of external linters. Default to False.
        allow_temporary_runners (bool, optional): Allow temporary tool runners (e.g. npx/uvx). Default to False.
    """
    return _analyze_snippet_internal(
        code=code,
        language=language,
        tool_version=ANALYZE_ZEN_VIOLATIONS_VERSION,
        severity_threshold=severity_threshold,
        perspective=perspective,
        project_as=project_as,
        enable_external_tools=enable_external_tools,
        allow_temporary_runners=allow_temporary_runners,
        reject_empty_code=False,
    )

@mcp.tool(
    name="analyze_zen_violations",
    version=ANALYZE_ZEN_VIOLATIONS_V2_VERSION,
    title="Analyze zen violations (v2)",
    description=(
        "Analyze a code snippet against zen rules with stricter request-quality "
        "guardrails and richer telemetry metadata."
    ),
    icons=ANALYSIS_TOOL_ICONS,
    tags={"analysis", "zen", "snippet", "v2"},
    annotations=READONLY_ANNOTATIONS,
    output_schema=_output_schema(AnalysisResult),
)
async def analyze_zen_violations_v2(  # noqa: PLR0913
    code: str,
    language: str,
    severity_threshold: int | None = None,
    perspective: PerspectiveMode = PerspectiveMode.ALL,
    project_as: str | None = None,
    *,
    enable_external_tools: bool = False,
    allow_temporary_runners: bool = False,
) -> AnalysisResult:
    """Run v2.0 snippet analysis with non-empty code validation.

    Args:
        code (str): Source code to analyse.
        language (str): Programming language identifier.
        severity_threshold (int | None, optional): Severity threshold. Default to None.
        perspective (PerspectiveMode, optional): Requested analysis perspective. Default to ``PerspectiveMode.ALL``.
        project_as (str | None, optional): Projection-family target when ``perspective`` is ``projection``.
        enable_external_tools (bool, optional): Enable external tools. Default to False.
        allow_temporary_runners (bool, optional): Allow temporary runners. Default to False.
    """
    return _analyze_snippet_internal(
        code=code,
        language=language,
        tool_version=ANALYZE_ZEN_VIOLATIONS_V2_VERSION,
        severity_threshold=severity_threshold,
        perspective=perspective,
        project_as=project_as,
        enable_external_tools=enable_external_tools,
        allow_temporary_runners=allow_temporary_runners,
        reject_empty_code=True,
    )

@mcp.tool(
    name="generate_prompts",
    version=GENERATE_PROMPTS_VERSION,
    title="Generate remediation prompts",
    description="Generate remediation prompts from zen analysis results.",
    icons=PROMPT_TOOL_ICONS,
    tags={"prompts", "remediation"},
    annotations=READONLY_ANNOTATIONS,
    output_schema=_output_schema(PromptBundle),
)
async def generate_prompts_tool(  # noqa: PLR0913
    code: str,
    language: str,
    perspective: PerspectiveMode = PerspectiveMode.ALL,
    project_as: str | None = None,
    *,
    enable_external_tools: bool = False,
    allow_temporary_runners: bool = False,
) -> PromptBundle:
    """Generate remediation prompts for v1.0 prompt generation.

    Args:
        code (str): Source code to analyse.
        language (str): Programming language identifier.
        perspective (PerspectiveMode, optional): Requested analysis perspective. Default to ``PerspectiveMode.ALL``.
        project_as (str | None, optional): Projection-family target when ``perspective`` is ``projection``.
        enable_external_tools (bool, optional): Enable external tools. Default to False.
        allow_temporary_runners (bool, optional): Allow temporary runners. Default to False.
    """
    return _generate_prompts_internal(
        code=code,
        language=language,
        tool_version=GENERATE_PROMPTS_VERSION,
        perspective=perspective,
        project_as=project_as,
        enable_external_tools=enable_external_tools,
        allow_temporary_runners=allow_temporary_runners,
    )

@mcp.tool(
    name="generate_prompts",
    version=GENERATE_PROMPTS_V2_VERSION,
    title="Generate remediation prompts (v2)",
    description=(
        "Generate remediation prompts with MCP-first guidance metadata and v2 "
        "versioned prompt semantics."
    ),
    icons=PROMPT_TOOL_ICONS,
    tags={"prompts", "remediation", "v2"},
    annotations=READONLY_ANNOTATIONS,
    output_schema=_output_schema(PromptBundle),
)
async def generate_prompts_tool_v2(  # noqa: PLR0913
    code: str,
    language: str,
    perspective: PerspectiveMode = PerspectiveMode.ALL,
    project_as: str | None = None,
    *,
    enable_external_tools: bool = False,
    allow_temporary_runners: bool = False,
) -> PromptBundle:
    """Generate remediation prompts for v2.0 prompt generation.

    Args:
        code (str): Source code to analyse.
        language (str): Programming language identifier.
        perspective (PerspectiveMode, optional): Requested analysis perspective. Default to ``PerspectiveMode.ALL``.
        project_as (str | None, optional): Projection-family target when ``perspective`` is ``projection``.
        enable_external_tools (bool, optional): Enable external tools. Default to False.
        allow_temporary_runners (bool, optional): Allow temporary runners. Default to False.
    """
    return _generate_prompts_internal(
        code=code,
        language=language,
        tool_version=GENERATE_PROMPTS_V2_VERSION,
        perspective=perspective,
        project_as=project_as,
        enable_external_tools=enable_external_tools,
        allow_temporary_runners=allow_temporary_runners,
    )

@mcp.tool(
    name="analyze_repository",
    title="Analyze repository",
    description="Analyze a repository path and return per-file analysis results.",
    tags={"analysis", "repository"},
    annotations=READONLY_ANNOTATIONS,
    task=BACKGROUND_TASK,
)
async def analyze_repository(  # noqa: PLR0913
    repo_path: str,
    languages: list[str] | None = None,
    max_files: int = 100,
    ctx: fastmcp.Context | None = None,
    *,
    enable_external_tools: bool = False,
    allow_temporary_runners: bool = False,
) -> list[RepositoryAnalysis]:
    """Analyse every eligible file in a repository and return per-file results.

    This is the public MCP tool that wraps
    ``_analyze_repository_internal``.  It exists as a thin async façade so
    that the internal helper can also be called from non-tool code paths
    (such as ``generate_agent_tasks_tool``) without duplicating parameter
    validation or the ``@mcp.tool`` decorator.

    Args:
        repo_path (str): Absolute path to the repository root.  The MCP
            client typically resolves this from the active workspace.
        languages (list[str] | None, optional): Restrict analysis to specific
            language identifiers.  Defaults to ``["python"]`` internally.
        max_files (int, optional): Per-language cap on the number of files to
            analyse, protecting against excessive runtime on monorepos. Default to 100.
        ctx (fastmcp.Context | None, optional): Optional FastMCP context for progress
            and log updates during repository analysis. Default to None.
        enable_external_tools (bool, optional): Opt-in execution of allow-listed
            external tools while analyzing files. Default to False.
        allow_temporary_runners (bool, optional): Permit temporary-runner fallback
            strategies for external tools. Default to False.

    Returns:
        list[RepositoryAnalysis]: List of ``RepositoryAnalysis`` entries, each pairing a file path
        and language with its ``AnalysisResult``.

    Example:
        ```python
        entries = await analyze_repository(
            "/home/dev/myproject", languages=["python", "go"], max_files=50
        )
        for entry in entries:
            print(entry.path, entry.result.overall_score)
        ```

    See Also:
        [`generate_agent_tasks_tool`][mcp_zen_of_languages.server.generate_agent_tasks_tool]:
            Builds actionable remediation tasks from repository analysis.

    """
    return await _analyze_repository_internal(
        repo_path,
        languages,
        max_files,
        ctx,
        enable_external_tools=enable_external_tools,
        allow_temporary_runners=allow_temporary_runners,
    )

@mcp.tool(
    name="analyze_batch",
    title="Analyze repository (batch / LLM-safe)",
    description=(
        "Analyse a repository path and return token-budgeted, paginated violations "
        "designed for LLM context windows. Highest-severity violations appear first. "
        "Pass the returned cursor to resume from the next page."
    ),
    tags={"analysis", "batch", "pagination"},
    annotations=READONLY_ANNOTATIONS,
    output_schema=_output_schema(BatchPage),
    task=BACKGROUND_TASK,
)
async def analyze_batch(  # noqa: PLR0913
    path: str,
    language: str,
    cursor: str | None = None,
    max_tokens: int = 8000,
    max_files: int = 100,
    *,
    enable_external_tools: bool = False,
    allow_temporary_runners: bool = False,
) -> BatchPage:
    """Analyse a repository and return one token-budgeted page of violations.

    This tool is explicitly designed for LLM agent workflows where the
    full violation list would exceed the model's context window.  It
    analyses the repository, sorts violations globally by severity
    (highest first), and returns only as many as fit within *max_tokens*.
    The opaque *cursor* field in the response encodes the resume position;
    pass it back unchanged on the next call to advance to the next page.

    Design principles:

    * **Stateless** — the cursor encodes the exact position in the sorted
      violation list; no server-side session state is required.
    * **Token-budget aware** — the response is trimmed so that the
      serialised payload stays within *max_tokens*.
    * **Priority ordering** — highest-severity violations are surfaced
      first across all pages.

    Args:
        path (str): Absolute or relative path to the repository root.
        language (str): Language identifier to restrict analysis (e.g. ``"python"``).
        cursor (str | None, optional): Opaque continuation token from a previous
            call.  Omit or pass ``None`` to start from the first page. Default to None.
        max_tokens (int, optional): Approximate token budget for the ``violations``
            payload.  Violations are added until the budget would be exceeded;
            the envelope overhead is excluded from this count. Default to 8000.
        max_files (int, optional): Cap on the number of files to analyse.
            Default to 100.
        enable_external_tools (bool, optional): Opt-in execution of external linters.
            Default to False.
        allow_temporary_runners (bool, optional): Permit temporary-runner strategies.
            Default to False.

    Returns:
        BatchPage: A page carrying the token-budgeted violations, a continuation
        cursor (``None`` when all violations have been returned), and file
        count metadata.

    Example:
        ```python
        # First page
        page = await analyze_batch("/repo", "python", max_tokens=4000)
        while page.has_more:
            page = await analyze_batch("/repo", "python", cursor=page.cursor)
        ```

    See Also:
        [`analyze_batch_summary`][mcp_zen_of_languages.server.analyze_batch_summary]:
            Returns a compact health-score overview that always fits in one
            context window.
        [`analyze_repository`][mcp_zen_of_languages.server.analyze_repository]:
            Full (unpaginated) repository analysis for non-LLM consumers.
    """
    canonical_language = _canonical_language(language)
    results = await _analyze_repository_internal(
        path,
        [canonical_language],
        max_files,
        None,
        enable_external_tools=enable_external_tools,
        allow_temporary_runners=allow_temporary_runners,
    )
    files_total = len(results)

    all_violations = _build_batch_violations_list(results)

    start_index, _ = _decode_cursor(cursor) if cursor else (0, 0)
    start_index = max(0, min(start_index, len(all_violations)))

    page_violations: list[BatchViolation] = []
    used_tokens = 0
    token_budget = max(1, max_tokens - _BATCH_ENVELOPE_TOKEN_OVERHEAD)

    next_index = start_index
    for bv in all_violations[start_index:]:
        serialised = json.dumps(bv.model_dump())
        cost = _estimate_tokens(serialised)
        if used_tokens + cost > token_budget:
            if not page_violations:
                # Single violation is too large for the budget; skip it so the
                # cursor advances and we do not get stuck on the same item.
                next_index += 1
            break
        page_violations.append(bv)
        used_tokens += cost
        next_index += 1

    has_more = next_index < len(all_violations)
    next_cursor = _encode_cursor(next_index) if has_more else None

    # files_in_page: distinct files represented in this page's violations
    files_in_page = len({bv.file for bv in page_violations})
    # Use a fixed logical page size (independent of token budget) so that the
    # page number is stable across calls with different max_tokens values.
    _logical_page_size = 50
    page_num = start_index // _logical_page_size + 1 if all_violations else 1

    return BatchPage(
        cursor=next_cursor,
        page=page_num,
        has_more=has_more,
        violations=page_violations,
        files_in_page=files_in_page,
        files_total=files_total,
    )

@mcp.tool(
    name="analyze_batch_summary",
    title="Analyze repository — batch summary",
    description=(
        "Return a compact project health score and top-5 hotspot files from a "
        "repository scan. Always fits within a single LLM context window. "
        "Use this before analyze_batch to decide whether full pagination is needed."
    ),
    tags={"analysis", "batch", "summary"},
    annotations=READONLY_ANNOTATIONS,
    output_schema=_output_schema(BatchSummary),
    task=BACKGROUND_TASK,
)
async def analyze_batch_summary(
    path: str,
    language: str,
    max_files: int = 100,
    *,
    enable_external_tools: bool = False,
    allow_temporary_runners: bool = False,
) -> BatchSummary:
    """Return a compact health overview for a repository — always one page.

    Unlike ``analyze_batch``, which paginates a potentially large violation
    list, this tool summarises the entire repository in a single response.
    It is designed to fit comfortably inside any LLM context window so that
    an agent can assess project health and identify where to focus before
    deciding whether to call ``analyze_batch`` for deeper detail.

    The returned ``health_score`` is the repository's average ``overall_score``
    expressed on a 0-100 scale (higher is better).  The ``hotspots`` list
    contains the five files with the highest violation count, ordered by
    descending total violations.

    Args:
        path (str): Absolute or relative path to the repository root.
        language (str): Language identifier to restrict analysis (e.g. ``"python"``).
        max_files (int, optional): Cap on the number of files to analyse.
            Default to 100.
        enable_external_tools (bool, optional): Opt-in execution of external linters.
            Default to False.
        allow_temporary_runners (bool, optional): Permit temporary-runner strategies.
            Default to False.

    Returns:
        BatchSummary: Compact summary with ``health_score`` (0-100),
        up to five ``hotspots``, ``total_violations``, and ``total_files``.

    Example:
        ```python
        summary = await analyze_batch_summary("/repo", "python")
        print(summary.health_score, summary.hotspots)
        ```

    See Also:
        [`analyze_batch`][mcp_zen_of_languages.server.analyze_batch]:
            Full paginated violation detail for LLM agents.
        [`analyze_repository`][mcp_zen_of_languages.server.analyze_repository]:
            Complete unpaginated results for non-LLM consumers.
    """
    canonical_language = _canonical_language(language)
    results = await _analyze_repository_internal(
        path,
        [canonical_language],
        max_files,
        None,
        enable_external_tools=enable_external_tools,
        allow_temporary_runners=allow_temporary_runners,
    )

    total_files = len(results)
    total_violations = sum(len(entry.result.violations) for entry in results)

    health_score = _compute_health_score(results)

    # Top-5 hotspots by violation count, tie-broken by top severity
    sorted_results = sorted(
        results,
        key=lambda e: (
            len(e.result.violations),
            max((v.severity for v in e.result.violations), default=0),
        ),
        reverse=True,
    )
    hotspots = [
        BatchHotspot(
            path=entry.path,
            language=entry.language,
            violations=len(entry.result.violations),
            top_severity=max((v.severity for v in entry.result.violations), default=0),
        )
        for entry in sorted_results[:5]
    ]

    return BatchSummary(
        health_score=health_score,
        hotspots=hotspots,
        total_violations=total_violations,
        total_files=total_files,
    )

@mcp.tool(
    name="analyze_batch_auto",
    title="Analyze repository (auto-routing)",
    description=(
        "Smart entry point for LLM agents: automatically decides between returning "
        "all violations at once (small repos) or paginating (large repos). "
        "Pass the returned cursor back to continue pagination if has_more is true. "
        "Prefer this over manually choosing between analyze_repository and analyze_batch."
    ),
    tags={"analysis", "batch", "pagination", "auto"},
    annotations=READONLY_ANNOTATIONS,
    output_schema=_output_schema(BatchPage),
    task=BACKGROUND_TASK,
)
async def analyze_batch_auto(  # noqa: PLR0913
    path: str,
    language: str,
    cursor: str | None = None,
    max_tokens: int = 8000,
    max_files: int = 100,
    *,
    enable_external_tools: bool = False,
    allow_temporary_runners: bool = False,
) -> BatchPage:
    """Analyse a repository, routing automatically between full and paginated mode.

    This tool eliminates the need to choose between ``analyze_repository``
    (full, unbounded) and ``analyze_batch`` (always paginated).  It analyses
    the repository once, then decides:

    * If all violations fit within *max_tokens*: returns them all in a single
      page (``has_more=False``, ``cursor=None``).
    * If violations exceed *max_tokens*: returns the first token-budgeted page
      with a continuation cursor (``has_more=True``).

    Cursor continuation is handled identically to ``analyze_batch`` — pass the
    opaque cursor from the previous response unchanged.

    Args:
        path (str): Absolute or relative path to the repository root.
        language (str): Language identifier to restrict analysis (e.g. ``"python"``).
        cursor (str | None, optional): Opaque continuation token from a previous
            call.  Omit or pass ``None`` to start from the first page. Default to None.
        max_tokens (int, optional): Approximate token budget for the ``violations``
            payload.  When all violations fit, they are returned in full; otherwise
            the first page is returned. Default to 8000.
        max_files (int, optional): Cap on the number of files to analyse.
            Default to 100.
        enable_external_tools (bool, optional): Opt-in execution of external linters.
            Default to False.
        allow_temporary_runners (bool, optional): Permit temporary-runner strategies.
            Default to False.

    Returns:
        BatchPage: A page with ``has_more=False`` when all violations were returned,
        or a page with a continuation cursor when more pages remain.

    Example:
        ```python
        # Single call for small repos; LLM paginates only when needed
        page = await analyze_batch_auto("/repo", "python")
        while page.has_more:
            page = await analyze_batch_auto("/repo", "python", cursor=page.cursor)
        ```

    See Also:
        [`analyze_batch`][mcp_zen_of_languages.server.analyze_batch]:
            Always-paginated variant with explicit cursor management.
        [`analyze_batch_summary`][mcp_zen_of_languages.server.analyze_batch_summary]:
            Compact overview (health score + hotspots) to decide if pagination is needed.
    """
    canonical_language = _canonical_language(language)

    # Cursor continuation: delegate entirely to analyze_batch
    if cursor is not None:
        return await analyze_batch.fn(
            path,
            canonical_language,
            cursor=cursor,
            max_tokens=max_tokens,
            max_files=max_files,
            enable_external_tools=enable_external_tools,
            allow_temporary_runners=allow_temporary_runners,
        )

    results = await _analyze_repository_internal(
        path,
        [canonical_language],
        max_files,
        None,
        enable_external_tools=enable_external_tools,
        allow_temporary_runners=allow_temporary_runners,
    )
    files_total = len(results)
    all_violations = _build_batch_violations_list(results)

    # Estimate total token cost for all violations
    token_budget = max(1, max_tokens - _BATCH_ENVELOPE_TOKEN_OVERHEAD)
    total_estimated = sum(
        _estimate_tokens(json.dumps(bv.model_dump())) for bv in all_violations
    )

    if total_estimated <= token_budget:
        # Small enough: return everything at once
        return BatchPage(
            cursor=None,
            page=1,
            has_more=False,
            violations=all_violations,
            files_in_page=len({bv.file for bv in all_violations}),
            files_total=files_total,
        )

    # Too large: apply token-budget paging from the start (cursor=None → index 0)
    page_violations: list[BatchViolation] = []
    used_tokens = 0
    next_index = 0
    for bv in all_violations:
        serialised = json.dumps(bv.model_dump())
        cost = _estimate_tokens(serialised)
        if used_tokens + cost > token_budget:
            if not page_violations:
                next_index += 1
            break
        page_violations.append(bv)
        used_tokens += cost
        next_index += 1

    has_more = next_index < len(all_violations)
    next_cursor = _encode_cursor(next_index) if has_more else None
    files_in_page = len({bv.file for bv in page_violations})

    return BatchPage(
        cursor=next_cursor,
        page=1,
        has_more=has_more,
        violations=page_violations,
        files_in_page=files_in_page,
        files_total=files_total,
    )

@mcp.tool(
    name="generate_agent_tasks",
    title="Generate agent tasks",
    description=(
        "Convert zen violations into structured agent task lists for automated "
        "remediation."
    ),
    tags={"agent", "tasks", "automation"},
    annotations=READONLY_ANNOTATIONS,
    output_schema=_output_schema(AgentTaskList),
    task=BACKGROUND_TASK,
)
async def generate_agent_tasks_tool(
    repo_path: str,
    languages: list[str] | None = None,
    min_severity: int = 5,
    *,
    enable_external_tools: bool = False,
    allow_temporary_runners: bool = False,
) -> AgentTaskList:
    """Convert repository-level violations into prioritised remediation tasks.

    Agent workflows need structured, machine-readable work items — not
    prose reports.  This tool analyses the repository via
    ``_analyze_repository_internal``, extracts every violation whose
    severity meets *min_severity*, and transforms them into an
    ``AgentTaskList`` ordered by priority.  Each task carries the file
    path, rule identifier, and a concise action description an automated
    agent can execute without further context.

    Args:
        repo_path (str): Absolute path to the repository to scan.  All
            eligible source files are discovered recursively.
        languages (list[str] | None, optional): Restrict scanning to these
            languages.  Omit to analyse only Python files by default. Default to None.
        min_severity (int, optional): Severity floor (1-10 scale).  Violations
            below this threshold are excluded from the task list. Default to 5.
        enable_external_tools (bool, optional): Opt-in execution of allow-listed
            external tools while gathering repository analysis. Default to False.
        allow_temporary_runners (bool, optional): Permit temporary-runner fallback
            strategies for external tools. Default to False.

    Returns:
        AgentTaskList: AgentTaskList containing prioritised tasks ready for automated
        remediation, sorted from highest to lowest severity.

    See Also:
        [`analyze_repository`][mcp_zen_of_languages.server.analyze_repository]:
            Retrieves the raw per-file results that feed task generation.
        [`generate_prompts_tool`][mcp_zen_of_languages.server.generate_prompts_tool]:
            Provides human-readable remediation text rather than
            structured tasks.

    """
    repo_results = await _analyze_repository_internal(
        repo_path,
        languages=languages,
        enable_external_tools=enable_external_tools,
        allow_temporary_runners=allow_temporary_runners,
    )
    analysis_results = [entry.result for entry in repo_results]
    return build_agent_tasks(
        analysis_results,
        project=repo_path,
        min_severity=min_severity,
    )