Pipeline GitLab con SDK Headless

Descripción

Esta cápsula combina lo aprendido en las dos anteriores: el SDK headless (cápsula 02) y el modelo de pipelines de GitLab (cápsula 03). El resultado es un pipeline real de GitLab CI/CD que ejecuta Claude Code en cada Merge Request, con secrets manejados, artifacts entre stages, y output publicado al MR.

Vas a construir el pipeline paso a paso: setup del Docker executor, configuración de variables CI/CD, script SDK que genera el review, y publicación de comentarios al MR via GitLab API. Al terminar, vas a tener un pipeline operativo equivalente al de GitHub Actions del Módulo 1, pero corriendo en GitLab.

El Setup Mínimo

Estructura del repo:

mi-proyecto/
├── .gitlab-ci.yml
├── scripts/
│   ├── extract_diff.py
│   ├── code_review.py
│   └── publish_to_mr.py
├── CLAUDE.md
├── requirements.txt
└── ... (código del proyecto)

requirements.txt

anthropic>=0.39.0,<1.0.0
python-gitlab>=4.0.0

Configurar Variables CI/CD

En GitLab: Settings → CI/CD → Variables, agregar:

Nota sobre GITLAB_API_TOKEN: GitLab provee $CI_JOB_TOKEN automáticamente, pero tiene scope limitado y no permite postear comments en algunas configuraciones. Por eso configuras un PAT (Personal Access Token) explícito con scope api.

El Pipeline Completo

# .gitlab-ci.yml

stages:
  - prepare
  - analyze
  - publish

variables:
  PYTHON_VERSION: "3.11"
  CLAUDE_MODEL: "claude-haiku-4-5"
  CLAUDE_MAX_TOKENS: "4000"

default:
  image: python:3.11-slim
  before_script:
    - apt-get update && apt-get install -y --no-install-recommends git
    - pip install --cache-dir=.pip-cache --no-warn-script-location -r requirements.txt
  cache:
    key: pip-cache-${CI_COMMIT_REF_SLUG}
    paths:
      - .pip-cache/

# ============================================
# STAGE 1: Extraer el diff del MR
# ============================================
extract-diff:
  stage: prepare
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
      changes:
        - "src/**/*"
        - "lib/**/*"
        - "tests/**/*"
        - "package.json"
        - "requirements.txt"
  script:
    - python scripts/extract_diff.py
  artifacts:
    paths:
      - filtered_diff.txt
      - pr_files.json
    expire_in: 1 day

# ============================================
# STAGE 2: Análisis con Claude Code SDK
# ============================================
claude-review:
  stage: analyze
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
  needs:
    - extract-diff
  script:
    - python scripts/code_review.py
  artifacts:
    paths:
      - review_result.json
    expire_in: 7 days

# ============================================
# STAGE 3: Publicar al MR
# ============================================
publish-review:
  stage: publish
  rules:
    - if: $CI_PIPELINE_SOURCE == "merge_request_event"
  needs:
    - claude-review
  script:
    - python scripts/publish_to_mr.py

Características:

• 3 stages secuenciales
• Cada job depende explícitamente del anterior con needs:
• Artifacts pasan archivos entre stages
• Cache de pip por rama
• Solo corre en MRs con cambios en código (paths filter)

Script 1: Extraer el Diff

"""scripts/extract_diff.py — extrae diff del MR."""
import json
import os
import subprocess
import sys
from pathlib import Path
import gitlab


def get_pr_files_via_gitlab_api():
    """Obtener lista de archivos del MR via GitLab API."""
    gl = gitlab.Gitlab(
        os.environ["CI_SERVER_URL"],
        private_token=os.environ["GITLAB_API_TOKEN"],
    )
    project = gl.projects.get(os.environ["CI_PROJECT_ID"])
    mr = project.mergerequests.get(int(os.environ["CI_MERGE_REQUEST_IID"]))
    
    changes = mr.changes()
    return changes.get("changes", [])


def filter_relevant(files):
    """Filtrar archivos relevantes para review."""
    RELEVANT_EXTENSIONS = {".py", ".ts", ".tsx", ".js", ".jsx", ".go", ".rb"}
    
    def is_relevant(f):
        path = f.get("new_path", f.get("old_path", ""))
        if not any(path.endswith(ext) for ext in RELEVANT_EXTENSIONS):
            return False
        if "test" in path or "fixture" in path:
            return False
        if f.get("deleted_file"):
            return False
        return True
    
    return [f for f in files if is_relevant(f)]


def main() -> int:
    try:
        files = get_pr_files_via_gitlab_api()
        relevant = filter_relevant(files)
        
        if not relevant:
            print("No hay archivos relevantes para revisar.")
            Path("filtered_diff.txt").write_text("")
            Path("pr_files.json").write_text("[]")
            return 0
        
        # Construir diff combinado
        diff_parts = []
        for f in relevant:
            path = f.get("new_path", "")
            patch = f.get("diff", "")
            if patch:
                diff_parts.append(f"--- {path} ---\n{patch}\n")
        
        combined_diff = "\n".join(diff_parts)
        Path("filtered_diff.txt").write_text(combined_diff)
        Path("pr_files.json").write_text(json.dumps(relevant, indent=2))
        
        print(f"Total archivos: {len(files)}")
        print(f"Relevantes: {len(relevant)}")
        print(f"Diff size: {len(combined_diff.splitlines())} líneas")
        return 0
    
    except Exception as e:
        print(f"ERROR: {e}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())

Diferencias con la versión de GitHub:

• Usa python-gitlab en lugar de requests directo
• Variables: CI_SERVER_URL, CI_PROJECT_ID, CI_MERGE_REQUEST_IID (no GITHUB_*)
• El método mr.changes() retorna estructura de GitLab (new_path, diff, deleted_file)

Script 2: Code Review con SDK

"""scripts/code_review.py — review con Claude Code SDK."""
import json
import os
import re
import sys
from pathlib import Path
from anthropic import Anthropic, APIError


def load_conventions() -> str:
    """Cargar CLAUDE.md si existe."""
    p = Path("CLAUDE.md")
    return p.read_text() if p.exists() else "Sin convenciones documentadas."


def build_prompt(diff: str, conventions: str) -> str:
    """Construir el prompt completo del review."""
    return f"""Eres un code reviewer profesional para este proyecto.

CONVENCIONES DEL PROYECTO:
{conventions}

INSTRUCCIONES:
- Aplica las convenciones del proyecto, no convenciones genéricas
- Solo comenta issues accionables (no "se ve bien")
- Severidad "critical" solo para bugs reales o vulnerabilidades

OUTPUT: devuelve SOLO un JSON con esta estructura:
{{
  "summary": {{
    "overview": "1-2 párrafos describiendo qué cambia y la calidad general",
    "highlights": ["punto 1", "punto 2", "..."],
    "verdict": "ready_to_merge|needs_minor_changes|needs_major_changes"
  }},
  "comments": [
    {{
      "path": "ruta/al/archivo",
      "line": 42,
      "severity": "critical|warning|suggestion",
      "body": "Texto del comentario en markdown"
    }}
  ]
}}

DIFF A REVISAR:

{diff}

"""


def parse_review_json(text: str) -> dict:
    """Parsear JSON manejando code fences."""
    text = text.strip()
    text = re.sub(r"^```(?:json)?\n?", "", text)
    text = re.sub(r"\n?```$", "", text)
    return json.loads(text)


def main() -> int:
    diff_path = Path("filtered_diff.txt")
    if not diff_path.exists() or not diff_path.read_text().strip():
        print("Sin diff que analizar. Skipping.")
        Path("review_result.json").write_text(
            json.dumps({"summary": None, "comments": [], "skipped": True})
        )
        return 0
    
    try:
        diff = diff_path.read_text()
        conventions = load_conventions()
        prompt = build_prompt(diff, conventions)
        
        client = Anthropic()
        response = client.messages.create(
            model=os.environ.get("CLAUDE_MODEL", "claude-haiku-4-5"),
            max_tokens=int(os.environ.get("CLAUDE_MAX_TOKENS", "4000")),
            messages=[{"role": "user", "content": prompt}],
        )
        
        text = response.content[0].text
        parsed = parse_review_json(text)
        
        result = {
            "summary": parsed["summary"],
            "comments": parsed["comments"],
            "tokens_used": {
                "input": response.usage.input_tokens,
                "output": response.usage.output_tokens,
            },
            "skipped": False,
        }
        
        Path("review_result.json").write_text(json.dumps(result, indent=2))
        print(f"Review generado: {len(result['comments'])} comments")
        print(f"Tokens: {result['tokens_used']['input']} in / {result['tokens_used']['output']} out")
        return 0
    
    except APIError as e:
        print(f"ERROR de Anthropic API: {e}", file=sys.stderr)
        return 1
    except json.JSONDecodeError as e:
        print(f"ERROR parseando JSON: {e}\nTexto:\n{text}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())

Notas:

• review_result.json queda como artifact para el siguiente stage
• Se maneja el caso de diff vacío (skip silencioso pero ordenado)
• Errores de API y de JSON parsing tienen mensajes específicos para debug

Script 3: Publicar al MR

"""scripts/publish_to_mr.py — publicar review como comments en GitLab MR."""
import json
import os
import sys
from pathlib import Path
import gitlab


MARKER = "<!-- claude-code-bot -->"


def build_summary_markdown(review: dict) -> str:
    """Construir el markdown del summary."""
    summary = review["summary"]
    comments = review["comments"]
    
    by_severity = {"critical": 0, "warning": 0, "suggestion": 0}
    for c in comments:
        by_severity[c.get("severity", "suggestion")] += 1
    
    verdict_emoji = {
        "ready_to_merge": "✅",
        "needs_minor_changes": "⚠️",
        "needs_major_changes": "🚨",
    }.get(summary.get("verdict", ""), "💬")
    
    body = f"""{MARKER}

# 🤖 Code Review (Claude Code)

## Resumen

{summary.get("overview", "")}

### Puntos clave
"""
    for h in summary.get("highlights", []):
        body += f"- {h}\n"
    
    body += f"""
### Veredicto

{verdict_emoji} **{summary.get("verdict", "").replace("_", " ").title()}**

### Severidad
- 🚨 Critical: {by_severity['critical']}
- ⚠️ Warning: {by_severity['warning']}
- 💡 Suggestion: {by_severity['suggestion']}

Detalles inline en los archivos afectados.
"""
    return body


def find_existing_bot_comment(notes):
    """Buscar comment existente del bot."""
    for note in notes:
        if MARKER in note.body:
            return note
    return None


def publish_summary(mr, body: str):
    """Crear o actualizar el summary general."""
    notes = mr.notes.list(get_all=True)
    existing = find_existing_bot_comment(notes)
    
    if existing:
        existing.body = body
        existing.save()
        print(f"Summary actualizado en note {existing.id}")
    else:
        new_note = mr.notes.create({"body": body})
        print(f"Summary creado: note {new_note.id}")


def publish_inline_comments(mr, comments: list):
    """Publicar inline comments en líneas específicas."""
    # GitLab usa "discussions" con position para inline comments
    for c in comments:
        try:
            mr.discussions.create({
                "body": format_severity(c["severity"]) + c["body"],
                "position": {
                    "base_sha": mr.diff_refs["base_sha"],
                    "start_sha": mr.diff_refs["start_sha"],
                    "head_sha": mr.diff_refs["head_sha"],
                    "position_type": "text",
                    "new_path": c["path"],
                    "new_line": c["line"],
                },
            })
            print(f"  ✓ Inline comment: {c['path']}:{c['line']}")
        except Exception as e:
            print(f"  ✗ Skip inline ({c['path']}:{c['line']}): {e}")


def format_severity(severity: str) -> str:
    return {
        "critical": "🚨 **CRITICAL** — ",
        "warning": "⚠️ **WARNING** — ",
        "suggestion": "💡 **Suggestion** — ",
    }.get(severity, "")


def main() -> int:
    review_file = Path("review_result.json")
    if not review_file.exists():
        print("ERROR: review_result.json no encontrado.", file=sys.stderr)
        return 1
    
    review = json.loads(review_file.read_text())
    
    if review.get("skipped"):
        print("Review fue skipped (sin diff). No publico.")
        return 0
    
    try:
        gl = gitlab.Gitlab(
            os.environ["CI_SERVER_URL"],
            private_token=os.environ["GITLAB_API_TOKEN"],
        )
        project = gl.projects.get(os.environ["CI_PROJECT_ID"])
        mr = project.mergerequests.get(int(os.environ["CI_MERGE_REQUEST_IID"]))
        
        # Summary general
        summary_md = build_summary_markdown(review)
        publish_summary(mr, summary_md)
        
        # Inline comments
        if review.get("comments"):
            print(f"Publicando {len(review['comments'])} inline comments...")
            publish_inline_comments(mr, review["comments"])
        
        return 0
    
    except Exception as e:
        print(f"ERROR: {e}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    sys.exit(main())

Notas:

• python-gitlab simplifica las llamadas API (en vez de requests crudo)
• Inline comments en GitLab se llaman "discussions" con position
• Necesita diff_refs del MR para anclar inline comments al diff correcto
• Maneja errores de inline comments individualmente (no falla todo el job)

Ejecutar el Pipeline

1. Push de .gitlab-ci.yml + scripts/ a una rama
2. Configurar variables ANTHROPIC_API_KEY y GITLAB_API_TOKEN en Settings
3. Crear un MR contra main
4. Ir a CI/CD → Pipelines → ver el run

Resultado esperado:

• Stage prepare → extrae diff y archivos en artifacts (~10-30s)
• Stage analyze → corre Claude Code, genera review_result.json (~10-60s)
• Stage publish → postea summary + inline comments al MR (~5-10s)

En el MR vas a ver:

• Comment general del bot al final
• Inline comments en archivos específicos

Trampas Comunes en Pipelines GitLab

Error 1: Usar $CI_JOB_TOKEN para API calls

Síntoma: API responde 401 al intentar postear comments.

Por qué pasa: CI_JOB_TOKEN tiene scope limitado en GitLab self-hosted o configuraciones específicas. No siempre permite escribir.

Cómo corregir: Crear un Personal Access Token (PAT) o Project Access Token con scope api y configurarlo como GITLAB_API_TOKEN.

Error 2: Olvidar apt-get install git en Docker image

Síntoma: git diff falla con "command not found".

Por qué pasa: Imágenes minimal como python:3.11-slim no incluyen git por defecto.

Cómo corregir: Instalar git en before_script:

before_script:
  - apt-get update && apt-get install -y --no-install-recommends git

Error 3: needs: sin estar en el mismo pipeline

Síntoma: El job que necesita artifacts falla porque "depends on extract-diff which is not in this pipeline".

Por qué pasa: Los rules: excluyeron extract-diff pero claude-review igual depende de él.

Cómo corregir: Mismas rules en jobs que se necesitan, o usar needs:[]:optional.

Error 4: Inline comments con líneas fuera del diff

Síntoma: El job falla con "the position is invalid" al postear inline comments.

Por qué pasa: El modelo sugirió comentar una línea que no está en el diff del MR.

Cómo corregir: Validar línea contra el patch antes de publicar (similar al patrón de la cápsula 03 del módulo 2). Skipear inline comments inválidos en lugar de fallar todo el job.

Error 5: Cache de pip por job sin compartir

Síntoma: Cada job instala dependencias desde cero, pipeline tarda mucho.

Por qué pasa: Sin cache configurado o con key: distinto en cada job.

Cómo corregir: Cache compartido con key: pip-cache-${CI_COMMIT_REF_SLUG} y paths: [.pip-cache/] en el default:.

Diagnóstico

Ejercicios

Ejercicio 1: Pipeline mínimo funcional (Medio)

Configura los 3 archivos (.gitlab-ci.yml + 2 scripts) en un repo de prueba con un MR. Verifica que:

1. El pipeline corre en el MR
2. Los 3 stages se ejecutan en orden
3. Aparece un comment del bot en el MR

Ejercicio 2: Agregar paths filter (Fácil)

Modifica las rules para que el pipeline solo corra cuando el MR cambia archivos *.py o *.ts. Verifica con un MR que solo cambia docs (no debe disparar) y otro que cambia código (sí debe disparar).

Ejercicio 3: Inline comments con validación (Difícil)

Implementa la validación de línea contra el diff antes de publicar inline comments. Si un comment apunta a una línea no presente en el diff, skipearlo y loguear (en lugar de fallar el job).

def is_line_in_diff(comment, files_in_diff):
    """Verificar si la línea está en el diff."""
    for f in files_in_diff:
        if f.get("new_path") == comment["path"]:
            patch = f.get("diff", "")
            # Parsear hunks: @@ -A,B +C,D @@
            for hunk in re.finditer(r"@@\s+-\d+,\d+\s+\+(\d+),(\d+)\s+@@", patch):
                start = int(hunk.group(1))
                count = int(hunk.group(2))
                if start <= comment["line"] < start + count:
                    return True
    return False

# Filtrar antes de publicar
valid_comments = [c for c in review["comments"] if is_line_in_diff(c, files)]

Resumen

• El pipeline GitLab tiene 3 stages que separan responsabilidades: prepare, analyze, publish
• Variables CI/CD con flags Protected + Masked manejan secrets
• needs: asegura dependencias entre jobs y pasaje de artifacts
• python-gitlab simplifica llamadas a la API (vs requests crudo)
• Inline comments en GitLab son "discussions" con position
• Cache de pip compartido entre jobs acelera el pipeline
• Validar líneas en el diff antes de inline comments evita errores

Próxima cápsula: 05 — Proyecto: Portar el bot del Módulo 2 a GitLab. Tomas el bot funcional que armaste para GitHub Actions y lo portas a GitLab demostrando portabilidad cross-platform — la lección central del módulo.

Recursos Adicionales

1. python-gitlab — Cliente Python oficial para GitLab API
2. GitLab Merge Requests API — Endpoint completo
3. GitLab Notes/Discussions API — Para comments y inline comments
4. GitLab CI/CD Predefined Variables — Lista completa
5. Project Access Tokens — Más seguros que PATs personales
6. GitLab Docker Images — Imágenes oficiales