tracer-cloud--opensre/integrations/github/helpers.py

"""Helper functions for GitHub tools."""

from __future__ import annotations

from typing import Any

from integrations.github.mcp import (
    DEFAULT_GITHUB_MCP_MODE,
    GitHubMCPConfig,
    build_github_mcp_config,
    github_mcp_config_from_env,
)


def github_source_available(sources: dict[str, dict]) -> bool:
    """Check if source is available."""
    return bool(sources.get("github", {}).get("connection_verified"))


def github_creds(gh: dict) -> dict[str, Any]:
    """Map classified GitHub integration fields to tool credential kwargs."""
    creds: dict[str, Any] = {}
    url = gh.get("github_url") or gh.get("url")
    if url:
        creds["github_url"] = url
    mode = gh.get("github_mode") or gh.get("mode")
    if mode:
        creds["github_mode"] = mode
    token = gh.get("github_token") or gh.get("auth_token")
    if token:
        creds["github_token"] = token
    command = gh.get("github_command") or gh.get("command")
    if command:
        creds["github_command"] = command
    args = gh.get("github_args")
    if args is None:
        args = gh.get("args")
    if args:
        creds["github_args"] = list(args)
    return creds


def _has_explicit_github_mcp_overrides(
    github_url: str | None,
    github_mode: str | None,
    github_token: str | None,
    github_command: str | None,
    github_args: list[str] | None,
) -> bool:
    if github_url or github_token or github_command or github_args:
        return True
    return bool(github_mode and github_mode != DEFAULT_GITHUB_MCP_MODE)


def resolve_github_mcp_config(
    github_url: str | None,
    github_mode: str | None,
    github_token: str | None,
    github_command: str | None = None,
    github_args: list[str] | None = None,
) -> GitHubMCPConfig | None:
    """Resolve GitHub MCP config."""
    env_config = github_mcp_config_from_env()
    if not _has_explicit_github_mcp_overrides(
        github_url, github_mode, github_token, github_command, github_args
    ):
        return env_config
    return build_github_mcp_config(
        {
            "url": github_url or (env_config.url if env_config else ""),
            "mode": github_mode or (env_config.mode if env_config else DEFAULT_GITHUB_MCP_MODE),
            "auth_token": github_token or (env_config.auth_token if env_config else ""),
            "command": github_command or (env_config.command if env_config else ""),
            "args": github_args or (list(env_config.args) if env_config else []),
            "headers": env_config.headers if env_config else {},
            "toolsets": env_config.toolsets if env_config else (),
        }
    )


def normalize_github_tool_result(result: dict[str, Any]) -> dict[str, Any]:
    """Normalize GitHub tool result."""
    if result.get("is_error"):
        return {
            "source": "github",
            "available": False,
            "error": result.get("text") or "GitHub MCP tool call failed.",
            "tool": result.get("tool"),
            "arguments": result.get("arguments", {}),
        }
    return {
        "source": "github",
        "available": True,
        "tool": result.get("tool"),
        "arguments": result.get("arguments", {}),
        "text": result.get("text", ""),
        "structured_content": result.get("structured_content"),
        "content": result.get("content", []),
    }