ado_api.py

import requests
from base64 import b64encode
from config import ADO_ORGANIZATION, ADO_PROJECT, ADO_PIPELINE_ID, ADO_PAT


def call_ado_api(min_time: str = None, max_time: str = None) -> str:
    """
    Queries Azure DevOps for the latest successful build in the configured pipeline.

    Args:
        min_time: ISO8601 start of time window (optional)
        max_time: ISO8601 end of time window (optional)

    Returns:
        Build number string (e.g. '2026.3.0.71') or None if not found.
    """
    url = (
        f"https://dev.azure.com/{ADO_ORGANIZATION}/{ADO_PROJECT}"
        f"/_apis/build/builds"
        f"?definitions={ADO_PIPELINE_ID}"
        f"&statusFilter=completed"
        f"&resultFilter=succeeded"
        f"&$top=1"
        f"&api-version=7.0"
    )

    if min_time:
        url += f"&minTime={min_time}"
    if max_time:
        url += f"&maxTime={max_time}"

    token = b64encode(f":{ADO_PAT}".encode()).decode()
    headers = {"Authorization": f"Basic {token}"}

    try:
        response = requests.get(url, headers=headers, timeout=15)
        response.raise_for_status()
        data = response.json()
        if data.get("value"):
            return data["value"][0]["buildNumber"]
        return None
    except requests.RequestException as e:
        print(f"[ado_api] ERROR querying ADO: {e}")
        return None


def get_build_history(top: int = 10) -> list:
    """
    Returns the last N successful builds from Azure DevOps.
    Used by the deployment history tool in mcp_server.py.
    """
    url = (
        f"https://dev.azure.com/{ADO_ORGANIZATION}/{ADO_PROJECT}"
        f"/_apis/build/builds"
        f"?definitions={ADO_PIPELINE_ID}"
        f"&statusFilter=completed"
        f"&resultFilter=succeeded"
        f"&$top={top}"
        f"&api-version=7.0"
    )

    token = b64encode(f":{ADO_PAT}".encode()).decode()
    headers = {"Authorization": f"Basic {token}"}

    try:
        response = requests.get(url, headers=headers, timeout=15)
        response.raise_for_status()
        data = response.json()
        builds = []
        for b in data.get("value", []):
            builds.append({
                "buildNumber": b.get("buildNumber"),
                "finishTime": b.get("finishTime"),
                "requestedBy": b.get("requestedBy", {}).get("displayName", "unknown")
            })
        return builds
    except requests.RequestException as e:
        print(f"[ado_api] ERROR fetching build history: {e}")
        return []