Overview

What We'll Build

The Architecture

Prerequisites

Detailed setup instructions for Snowflake, Python, Slack, and GitHub are in the companion repo. This guide focuses on the pipeline itself.

Setup

CREATE DATABASE IF NOT EXISTS PREFECT_QUICKSTART;
CREATE SCHEMA IF NOT EXISTS PREFECT_QUICKSTART.RAW;
CREATE SCHEMA IF NOT EXISTS PREFECT_QUICKSTART.ANALYTICS;
CREATE WAREHOUSE IF NOT EXISTS PREFECT_WH WITH WAREHOUSE_SIZE='XSMALL' AUTO_SUSPEND=60 AUTO_RESUME=TRUE;

-- Verify Cortex AI works in your region
SELECT SNOWFLAKE.CORTEX.COMPLETE('mistral-large2', 'Say hello') AS test;

If the Cortex test fails, try llama3.1-70b or claude-3-5-sonnet — model availability varies by region. You can also enable cross-region inference: ALTER ACCOUNT SET CORTEX_ENABLED_CROSS_REGION = 'ANY_REGION';

Full table DDL and seed data are in the repo's setup.sql.

uv pip install prefect requests pandas "snowflake-connector-python[pandas]"
prefect cloud login

pause_flow_run requires a connection to Prefect Cloud (or a self-hosted Prefect server). Make sure prefect cloud login completes before running the pipeline.

export GITHUB_TOKEN=your_github_pat
export SNOWFLAKE_CONNECTION_NAME=your_connection
export SLACK_BOT_TOKEN=xoxb-your-bot-token
export SLACK_CHANNEL=your-channel

See the repo README for Snowflake connection configuration, Slack app setup, GitHub token scopes, and uv / pyproject.toml setup.

Fetch and Ingest

# retries with exponential backoff — handles GitHub rate limits automatically
@task(retries=3, retry_delay_seconds=exponential_backoff(backoff_factor=2), log_prints=True)
def fetch_pr_data(repo: str, pr_number: int) -> dict:
    logger = get_run_logger()
    headers = {"Authorization": f"token {GITHUB_TOKEN}"} if GITHUB_TOKEN else {}

    # Fetch PR metadata (title, author, stats)
    pr_resp = requests.get(f"https://api.github.com/repos/{repo}/pulls/{pr_number}", headers=headers)
    pr_resp.raise_for_status()
    pr = pr_resp.json()

    # Fetch the raw diff separately (same endpoint, different Accept header)
    diff_resp = requests.get(
        f"https://api.github.com/repos/{repo}/pulls/{pr_number}",
        headers={**headers, "Accept": "application/vnd.github.v3.diff"},
    )

    # Return a clean dict — truncate diff to 50k chars to stay within Cortex limits
    return {
        "pr_number": pr_number, "repo": repo,
        "title": pr["title"], "author": pr["user"]["login"],
        "diff_content": diff_resp.text[:50000],
        "files_changed": pr["changed_files"],
        "additions": pr["additions"], "deletions": pr["deletions"],
        "created_at": pr["created_at"],
    }

@task(retries=2, retry_delay_seconds=exponential_backoff(backoff_factor=2), log_prints=True)
def load_pr_to_snowflake(pr_data: dict) -> int:
    conn = get_connection()
    # Build a single-row DataFrame matching the PULL_REQUESTS table schema
    df = pd.DataFrame([{
        "PR_NUMBER": pr_data["pr_number"], "REPO": pr_data["repo"],
        "TITLE": pr_data["title"], "AUTHOR": pr_data["author"],
        "DIFF_CONTENT": pr_data["diff_content"],
        "FILES_CHANGED": pr_data["files_changed"],
        "ADDITIONS": pr_data["additions"], "DELETIONS": pr_data["deletions"],
        "CREATED_AT": pd.Timestamp(pr_data["created_at"]),
    }])

    # write_pandas bulk-loads via PUT + COPY INTO — fast even for large diffs
    write_pandas(conn=conn, df=df, table_name="PULL_REQUESTS",
                 database="PREFECT_QUICKSTART", schema="RAW")
    return 1

Cortex AI Assessment

@task(retries=2, retry_delay_seconds=exponential_backoff(backoff_factor=2), log_prints=True)
def cortex_assess_pr(pr_data: dict) -> dict:
    logger = get_run_logger()
    conn = get_connection()
    # Truncate diff to 5k chars — balances context vs. token cost
    diff_preview = pr_data["diff_content"][:5000]

    # Build a shared context string for both Cortex calls
    pr_context = (
        f"PR Title: {pr_data['title']}\n"
        f"Files changed: {pr_data['files_changed']}, "
        f"+{pr_data['additions']}/-{pr_data['deletions']}\n"
        f"Diff:\n{diff_preview}"
    )

    summarize_prompt = (
        f"Summarize this PR in 2-3 sentences for a technical reviewer.\n\n{pr_context}"
    )

    cur = conn.cursor()
    # CLASSIFY_TEXT: purpose-built for classification — returns one of the given categories
    cur.execute(
        "SELECT SNOWFLAKE.CORTEX.CLASSIFY_TEXT(%s, ['low', 'medium', 'high'])",
        (pr_context,),
    )
    risk_level = cur.fetchone()[0].strip().lower()

    # COMPLETE: free-form generation for the summary
    cur.execute("SELECT SNOWFLAKE.CORTEX.COMPLETE(%s, %s)", ("mistral-large2", summarize_prompt))
    summary = cur.fetchone()[0].strip()
    cur.close()

    # Defensive fallback in case CLASSIFY_TEXT returns something unexpected
    if risk_level not in ("low", "medium", "high"):
        risk_level = "medium"

    return {"risk_level": risk_level, "summary": summary}

Why two different Cortex functions? Use the right tool for the job. CLASSIFY_TEXT guarantees a clean category from your list — no regex parsing, no prompt engineering for output format. COMPLETE is for open-ended generation where you need free-form text.

Decision and Human Gate

@task(retries=2, retry_delay_seconds=exponential_backoff(backoff_factor=2), log_prints=True)
def get_historical_context(pr_data: dict) -> dict:
    conn = get_connection()
    cur = conn.cursor()
    # Bind parameter for AUTHOR — safe against SQL injection
    cur.execute(
        "SELECT COUNT(*) AS total_prs, "
        "SUM(CASE WHEN HAD_INCIDENT THEN 1 ELSE 0 END) AS incidents "
        "FROM PREFECT_QUICKSTART.ANALYTICS.PR_HISTORY "
        "WHERE AUTHOR = %s",
        (pr_data["author"],),
    )
    result = cur.fetchone()
    cur.close()
    return {"total_prs": result[0] or 0, "incident_count": result[1] or 0}


@task(log_prints=True)
def make_decision(pr_data: dict, assessment: dict, history: dict) -> dict:
    risk = assessment["risk_level"]

    # Elevate risk if Cortex said "medium" but the author has caused incidents before
    if risk == "high" or (risk == "medium" and history["incident_count"] > 0):
        effective_risk = "high"
    elif risk == "medium":
        effective_risk = "medium"
    else:
        effective_risk = "low"

    # This dict drives the runtime branching in the flow
    return {
        "effective_risk": effective_risk,
        "requires_approval": effective_risk in ("medium", "high"),
        "auto_approved": effective_risk == "low",
    }

# Type-safe input schema — Prefect renders this as a form in the Cloud UI
class ReviewDecision(RunInput):
    approved: bool
    reviewer_name: str
    notes: str = ""

Close the Loop

@task(retries=2, retry_delay_seconds=exponential_backoff(backoff_factor=2), log_prints=True)
def close_github_pr(pr_data: dict, assessment: dict, decision: dict) -> None:
    headers = {"Authorization": f"token {GITHUB_TOKEN}"}
    repo, pr_num = pr_data["repo"], pr_data["pr_number"]
    approved = decision.get("final_approved", False)

    # Post the Cortex AI summary as a review comment on the PR
    comment = (
        f"## {'✅' if approved else '❌'} Prefect PR Review\n\n"
        f"**Risk:** {decision['effective_risk'].upper()}\n"
        f"**Reviewer:** {decision.get('reviewer', 'auto')}\n\n"
        f"### Cortex AI Summary\n{assessment['summary']}\n\n"
        f"---\n*Assessed by Prefect + Snowflake Cortex AI*"
    )
    requests.post(f"https://api.github.com/repos/{repo}/issues/{pr_num}/comments",
                  headers=headers, json={"body": comment})

    # Merge (squash) if approved, close if rejected
    if approved:
        requests.put(f"https://api.github.com/repos/{repo}/pulls/{pr_num}/merge",
                     headers=headers, json={"merge_method": "squash"})
    else:
        requests.patch(f"https://api.github.com/repos/{repo}/pulls/{pr_num}",
                       headers=headers, json={"state": "closed"})

Wiring It Together

@flow(name="pr-review-agent", log_prints=True, persist_result=True)
def pr_review_agent(repo: str, pr_number: int):
    logger = get_run_logger()

    # Get the flow run ID so we can deep-link Slack notifications to Prefect Cloud
    from prefect.runtime import flow_run
    flow_run_id = str(flow_run.id)

    # Open a single Snowflake connection for the entire flow — avoids repeated
    # OAuth browser popups and is more efficient than connecting per-task.
    conn = snowflake.connector.connect(connection_name=SF_CONNECTION)
    # Tag all queries so you can find them in Snowflake's QUERY_HISTORY
    conn.cursor().execute("ALTER SESSION SET QUERY_TAG = 'prefect-pr-review'")

    try:
        # ---- Phase 1: Fetch and Ingest ----
        # Pull PR metadata + diff from GitHub API
        pr_data = fetch_pr_data(repo=repo, pr_number=pr_number)
        # Land the raw data in Snowflake for auditability
        load_pr_to_snowflake(pr_data)

        # ---- Phase 2: AI Assessment ----
        # Use Snowflake Cortex AI to classify risk and generate a summary
        assessment = cortex_assess_pr(pr_data)
        # Query Snowflake for this author's historical PR track record
        history = get_historical_context(pr_data)
        # Combine Cortex output + history into an effective risk decision
        decision = make_decision(pr_data, assessment, history)

        # ---- Phase 3: Human Gate (conditional) ----
        # This is runtime branching — the path depends on what Cortex returned
        if decision["requires_approval"]:
            # Send a rich Slack message with a "Review & Approve" button
            slack_approval_request(pr_data, assessment, decision, flow_run_id)

            # PAUSE the flow — execution literally stops here and waits
            # for a human to submit the ReviewDecision form in Prefect Cloud.
            # timeout=86400 means the flow will fail if no one responds in 24h.
            review = pause_flow_run(
                wait_for_input=ReviewDecision.with_initial_data(
                    approved=False, reviewer_name="",
                ),
                timeout=86400,
            )

            decision["final_approved"] = review.approved
            decision["reviewer"] = review.reviewer_name

            # Write the assessment BEFORE the early return — every decision
            # (approved or rejected) gets an audit trail in Snowflake
            write_assessment(pr_data, assessment, decision,
                             reviewer=review.reviewer_name, reviewer_notes=review.notes)

            if not review.approved:
                # Rejected: close the PR on GitHub + notify Slack, then exit
                close_github_pr(pr_data, assessment, decision)
                slack_completion_notice(pr_data, assessment, decision)
                return

        else:
            # Low risk — auto-approve without human intervention
            decision["final_approved"] = True
            decision["reviewer"] = "auto"
            write_assessment(pr_data, assessment, decision)

        # ---- Phase 4: Close the Loop ----
        # Merge the PR on GitHub and post the Cortex AI review comment
        close_github_pr(pr_data, assessment, decision)
        # Send a final Slack notification with the outcome
        slack_completion_notice(pr_data, assessment, decision)

    finally:
        # Always close the Snowflake connection, even if the flow fails
        conn.close()


if __name__ == "__main__":
    pr_review_agent(repo="your-org/your-repo", pr_number=1)

Slack tasks and write_assessment are defined in the companion repo. They follow the same pattern: tasks call get_connection() internally and use bind parameters for all Snowflake queries.

Run It

GITHUB_TOKEN=your_token \
SNOWFLAKE_CONNECTION_NAME=your_connection \
SLACK_BOT_TOKEN=xoxb-your-token \
SLACK_CHANNEL=your-channel \
python pr_review_pipeline.py

This runs the flow as a local Python process connected to Prefect Cloud. For production, deploy with prefect deploy and run your worker on Snowpark Container Services — your orchestration compute stays inside Snowflake's security perimeter. See the companion repo for a prefect.yaml deployment example.

Deploy It

pull:
  - prefect.deployments.steps.git_clone:
      repository: "{{ $GIT_REPO_URL }}"
      branch: main
      access_token: "{{ $GIT_ACCESS_TOKEN }}"

deployments:
  - name: pr-review-agent
    entrypoint: pr_review_pipeline.py:pr_review_agent
    work_pool:
      name: spcs-pool    # Prefect Worker on Snowpark Container Services
    tags: [github, code-review, cortex-ai]
    parameters:
      repo: your-org/your-repo
      pr_number: 1
    enforce_parameter_schema: true
    description: "Agentic PR review: GitHub → Cortex AI → Human approval → Merge"

# Register the deployment with Prefect Cloud
prefect deploy --all

# Trigger it on-demand from the CLI
prefect deployment run pr-review-agent/pr-review-agent \
  --param repo=your-org/your-repo \
  --param pr_number=42

For production, we recommend Prefect Cloud (authentication, RBAC, audit logs, SSO) with a Prefect Worker running on Snowpark Container Services (SPCS) — keeping your orchestration compute inside Snowflake's security perimeter.

What Just Happened

Cleanup

Skip this if you want to keep your PR assessment history in Snowflake.

DROP DATABASE IF EXISTS PREFECT_QUICKSTART;
DROP WAREHOUSE IF EXISTS PREFECT_WH;

Conclusion

Now Imagine Your Workflows

Resources