ClaimFlow AI: Claims Triage Pipeline

Introduction

30 Second Summary

Claims often arrive as rushed emails with repeated details or missing information. Someone still has to turn that messy inbox into a clear queue for review.

In this project, you will build a nightly claims-triage pipeline that turns fictional emails into structured records for human review. A live FastAPI service will show the source claims plus the processed results.

What You'll Build

A live browser demo will show six messy claims reduced to five structured records with uncertain cases held for human review.

By the end of this project, you'll have:

• A live claims API where you can inspect six synthetic emails through interactive documentation.
• A PostgreSQL cleaning pipeline that reduces six source rows to five unique claims. Visible quality flags preserve evidence of the original problems.
• A nightly cloud demo that shows structured Claude results plus pipeline history. A public GitHub README explains the architecture plus the human-review boundary.
• Secret Mission: Add a dedicated /review-queue endpoint that returns only records needing manual attention with high urgency first.

Are there any prerequisites?

You need a Mac with Python, Git, an editor, plus a GitHub account. The project guides you through the remaining service accounts plus dependencies.

Before We Start

Before any hands-on work, commit to building ClaimFlow AI for insurance operations staff. It organizes synthetic claim information while leaving coverage approval, denial, liability, fraud assessment, and payment decisions to people.

Set Up ClaimFlow and Its Accounts

ClaimFlow will soon move customer-like records between an API, a database, and an AI service. A reproducible local environment keeps that flow consistent as the project grows.

Credentials unlock those services, so they need protection before any data moves. This step pins every dependency. It also keeps private values outside your Git history.

In this step, get ready to:

• Create a local Git repository with an isolated Python environment.
• Pin the Python dependencies for repeatable installs.
• Connect each cloud account with protected local credentials.

Verify Python and create the project

FastAPI requires Python 3.10 or later. A virtual environment gives ClaimFlow its own package installation.

• Press Cmd+Space to open Spotlight on your Mac.
• Type Terminal into Spotlight.
• Press Enter to open Terminal.
• Check your installed Python version by running this command:

python3 --version

What does this check prove?

The command reports the Python version attached to python3. ClaimFlow needs version 3.10 or later because its pinned FastAPI release requires that baseline.

✔️ I see version 3.10 or higher

Your installed version meets ClaimFlow's requirement. Keep this Terminal window open for the project setup.

ⓧ I see an older version

Your current Python version is below the project's minimum. macOS supports a newer Python installation alongside the system-managed version.

• Download a current Python installer from the official Python downloads page.
• Complete the macOS installer.
• Return to Terminal.
• Run the version check above again.
• Continue when the reported version is 3.10 or higher.

ⓧ Command not found

Your shell cannot currently find Python 3. Installing the current macOS release adds the required command.

• Download a current Python installer from the official Python downloads page.
• Complete the macOS installer.
• Return to Terminal.
• Run the version check above again.
• Continue when Terminal reports Python 3.10 or higher.

The local repository gives every later file a predictable home. Initializing it on the main branch also prepares the project for its eventual GitHub repository.

• Create the claimflow-ai folder on your Desktop by running these commands:

cd ~/Desktop
mkdir claimflow-ai
cd claimflow-ai
git init -b main

What does this setup create?

• The first command places the project somewhere easy to find.
• The next two commands create the claimflow-ai folder. They also make it your current Terminal location.
• The final command starts the repository on the main branch.

• Create ClaimFlow's isolated environment by running this command:

python3 -m venv .venv

Why use a virtual environment?

The .venv folder contains a project-specific Python interpreter. Packages installed there stay separate from the dependencies used by your other projects.

• Activate the new environment by running this command:

source .venv/bin/activate

What changes after activation?

Your prompt now starts with (.venv). The python command now uses the interpreter inside ClaimFlow's environment.

• Create the seven project files by running this command:

touch requirements.txt .gitignore .env.example .env main.py pipeline.py README.md

What did this command add?

The command creates files for dependencies, secret protection, application code, pipeline code, and documentation. You fill the setup files next while leaving main.py plus pipeline.py empty for later steps.

• Press Cmd+Space to open Spotlight.
• Type the name of your code editor.
• Press Enter to launch the editor.
• Use the editor's folder-opening control to select claimflow-ai from your Desktop.

You should see requirements.txt, .gitignore, .env.example, .env, main.py, pipeline.py, and README.md in the folder.

Missing a project file?

Confirm that your Terminal prompt shows the claimflow-ai folder before rerunning the file-creation command. A different folder means the files were created somewhere else.

Still stuck? Help me find the ClaimFlow files I created from Terminal.

Pin dependencies and protect secrets

Pinned versions make every installation reproducible. ClaimFlow also uses environment variables to keep database credentials and the API key out of application code.

• Select requirements.txt in your editor.
• Paste these pinned dependencies into the file:

fastapi==0.141.1
uvicorn==0.53.0
psycopg[binary]==3.3.6
anthropic==1.7.0
pydantic==2.13.5
python-dotenv==1.2.3

What do these packages provide?

• FastAPI 0.141.1 provides the web API framework.
• Uvicorn 0.53.0 runs that API locally.
• Psycopg 3.3.6 connects Python to PostgreSQL.
• Anthropic 1.7.0 provides the Claude client.
• Pydantic 2.13.5 validates structured claim fields.
• python-dotenv 1.2.3 loads local environment variables.

• Save requirements.txt.
• Return to the Terminal window from earlier.
• Install the pinned packages into the active environment by running this command:

python -m pip install -r requirements.txt

What does this installation do?

The command reads every pinned line from requirements.txt. It installs those exact releases inside .venv.

When the Terminal prompt returns without an installation error, the environment has all six dependencies.

Did the installation fail?

• Confirm that your Terminal prompt begins with (.venv).
• Confirm that requirements.txt contains the six pinned lines exactly as shown.
• Rerun the installation command after correcting the file.

Need another pair of eyes? Help me diagnose my ClaimFlow dependency installation.

• Select .gitignore in your editor.
• Paste these local-only paths into the file:

.venv/
.env
__pycache__/
*.pyc
.DS_Store

What stays outside Git?

The rules exclude the virtual environment, local credentials, Python cache files, and macOS metadata. The .env rule is the critical barrier that keeps secrets out of future commits.

• Save .gitignore.
• Select .env.example in your editor.
• Paste the safe configuration template into the file:

DATABASE_URL=your-neon-pooled-connection-string
ANTHROPIC_API_KEY=your-api-key-here
SOURCE_API_URL=http://127.0.0.1:8000/source/claims

Why keep an example file?

The example records every required setting without containing a live credential. Future readers can copy its structure without gaining access to your accounts.

• Save .env.example.
• Copy the safe template into the ignored local .env file by running this command:

cp .env.example .env

What does this copy preserve?

Both files now share the same three configuration names. Only .env receives real credentials.

The .env.example file remains safe to publish.

• Confirm that Git ignores .env by running this command:

git status

What should Git report?

You should see the project files listed as untracked. You should not see .env or .venv in that list.

• Select README.md in your editor.
• Paste this project introduction into the file:

# ClaimFlow AI

ClaimFlow AI pulls synthetic claim emails from an API, cleans them with PostgreSQL, extracts structured fields with Claude, and routes uncertain records to human review. It does not approve coverage, assign liability, detect fraud, or authorize payments.

What does the introduction establish?

The README states the pipeline's technical purpose. It also records the human-review boundary before any AI extraction code exists.

• Save README.md.

✔️ Awesome, I've got everything!

• Save every file in the claimflow-ai folder.
• Keep main.py empty for the API step.
• Keep pipeline.py empty for the data-cleaning step.

ⓧ I'd like to double check the full code

Compare each setup file with the references below. The private .env file is excluded because it contains your own credentials.

fastapi==0.141.1
uvicorn==0.53.0
psycopg[binary]==3.3.6
anthropic==1.7.0
pydantic==2.13.5
python-dotenv==1.2.3

This file pins all six project dependencies.

.venv/
.env
__pycache__/
*.pyc
.DS_Store

This file keeps local packages, secrets, caches, and macOS metadata outside Git.

DATABASE_URL=your-neon-pooled-connection-string
ANTHROPIC_API_KEY=your-api-key-here
SOURCE_API_URL=http://127.0.0.1:8000/source/claims

This file documents the three required environment variables without exposing credentials.

# ClaimFlow AI

ClaimFlow AI pulls synthetic claim emails from an API, cleans them with PostgreSQL, extracts structured fields with Claude, and routes uncertain records to human review. It does not approve coverage, assign liability, detect fraud, or authorize payments.

This README introduces the project plus its safety boundary.

• Confirm that main.py contains no text.
• Confirm that pipeline.py contains no text.

Connect the project accounts

Your local foundation is ready. The service accounts now provide the database, AI extraction, and future cloud deployment.

How do the accounts fit together?

Neon hosts the PostgreSQL database that stores raw claims, processed claims, and pipeline runs.

Claude later extracts typed fields from each fictional claim email.

Render deploys the application from GitHub in the final step.

• Open the Neon Console in your browser.
• Complete the Neon sign-in flow.
• Select Projects.
• Click New Project.
• Enter claimflow-ai as the project name.
• Choose the Free plan.
• Click Create Project.

You should see the new claimflow-ai project in the Neon Console. Its Free plan provides the database capacity needed for this synthetic demo.

Why use a pooled connection?

A pooled connection shares a managed group of database connections. This suits a scheduled pipeline because each run can connect without requiring a permanently open process.

The pooled connection string contains database credentials. You place it only in the ignored .env file, so Git never tracks it.

• Click Connect in the Neon project.
• Enable Connection pooling.
• Copy the displayed pooled connection string.
• Return to .env in your editor.
• Replace your-neon-pooled-connection-string with the pooled connection string.
• Save .env.

This next account uses prepaid API credits. Keep auto-reload off. Choose only an amount that leaves room for the later $1 monthly Render Cron Job within the project's $5 maximum.

• Open the Anthropic Console in your browser.
• Complete the Anthropic Console sign-in flow.
• Open Settings.
• Select Billing.
• Turn auto-reload off.
• Click Buy credits.
• Enter an amount that keeps the combined project cost at or below $5.
• Confirm the displayed purchase amount.

Your prepaid balance is capped for this project. Auto-reload cannot add another purchase in the background.

• Open API keys from Anthropic Console settings.
• Click Create key.
• Choose a key name that identifies local development.
• Complete the expiration field shown by the Console.
• Complete the linked account field shown by the Console.
• Copy the secret when the Console shows it once.
• Return to .env in your editor.
• Replace your-api-key-here with the copied secret.

• Save .env.

Keep the API key private

Never paste the key into .env.example, README.md, a screenshot, or a Git commit. Create a replacement key in Anthropic Console if the original secret is exposed.

Need help checking your setup safely? Help me verify that my Anthropic API key is protected without asking me to share it.

• Open the Render Dashboard in your browser.
• Choose GitHub as your sign-in method.
• Authorize Render to use your GitHub identity.
• Return to the Render Dashboard after authorization.

You should see the Render Dashboard while signed in through GitHub. Render can use the repository after you publish it in the deployment step.

Before the final check, what do you expect Python to do when it imports every pinned dependency from the activated environment?

• Return to the Terminal window from earlier.
• Confirm that the prompt still begins with (.venv).
• Verify every ClaimFlow dependency by running this command:

python -c "import fastapi, uvicorn, psycopg, anthropic, pydantic, dotenv; print('ClaimFlow dependencies ready')"

What does this verification prove?

The command imports every installed module in one Python process. The final message prints only after all six imports succeed.

You should see ClaimFlow dependencies ready in Terminal. That is the local foundation complete: ClaimFlow can now import every library it needs.

Missing a dependency?

• Confirm that your Terminal prompt starts with (.venv).
• Rerun the activation command from earlier if the environment is inactive.
• Rerun the dependency installation command after activation.

Still seeing an import error? Help me identify which ClaimFlow dependency is missing.

Your repository, dependencies, protected credentials, and service accounts are ready. Next, you turn the empty application file into a visible API with six deliberately messy claims.

Expose the Messy Claims API

Your ClaimFlow environment is ready. The next goal is to turn the empty main.py file into a browser-visible REST API.

A FastAPI service gives you an immediate view of the source records. That first response also exposes why customer-like data needs inspection before automation.

You will keep Uvicorn running while you build. Its reload mode applies each saved change to the local service.

In this step, get ready to:

• Build the API shell that identifies ClaimFlow AI.
• Expose six synthetic claims through a source endpoint.
• Add placeholder endpoints for processed claims plus pipeline runs.

Build the source API

The application shell loads your local environment before creating the API. Its metadata gives the interactive documentation a clear title plus a description.

• Build the application shell in main.py by replacing the empty file with this code:

import json
import os

import psycopg
from dotenv import load_dotenv
from fastapi import FastAPI

load_dotenv()

app = FastAPI(
    title="ClaimFlow AI",
    version="1.0.0",
    description="Synthetic insurance claim ingestion, cleaning, and human-reviewed AI triage.",
)

What does this setup do?

• The imports prepare the service to read environment values plus connect to the database in later routes.
• The load_dotenv() call loads the protected values from .env.
• The app object holds the routes that FastAPI serves.

• Save main.py.
• Start the local service from the activated virtual environment by running this command:

uvicorn main:app --reload

What does this command do?

Uvicorn serves the app object from main.py. The reload option watches saved code changes during local development.

• Leave this Terminal tab running.
• Open http://127.0.0.1:8000/docs in your browser.

You should see interactive documentation titled ClaimFlow AI. Your local API shell is now running.

Local documentation not loading?

• Confirm that the Terminal tab still has the activated .venv environment.
• Check that main.py is inside the claimflow-ai folder.

Still stuck? Help me diagnose why my local ClaimFlow FastAPI documentation is not loading.

The source collection uses fictional records plus documentation-only email addresses. Each dictionary represents one claim email before any cleaning takes place.

• Add the synthetic source collection below the app = FastAPI( block by pasting this code:

SYNTHETIC_CLAIMS = [
    {"external_id": "CLM-1001", "sender_email": "alex.rivera@example.com", "received_at": "2026-09-18T20:10:00Z", "subject": "Kitchen pipe leak", "body": "Policy POL-H10001. A pipe burst under the kitchen sink on 2026-09-18. Water reached two cabinets. The supply is shut off."},
    {"external_id": "CLM-1002", "sender_email": "jamie.chen@example.com", "received_at": "2026-09-18T21:05:00Z", "subject": "Car accident claim", "body": "My vehicle was hit from behind today. Policy POL-A20002. No injuries were reported. The rear bumper is damaged."},
    {"external_id": "CLM-1002", "sender_email": "Jamie.Chen@Example.com ", "received_at": "2026-09-18T21:22:00Z", "subject": "Update: car accident claim", "body": "Update for policy POL-A20002: the collision happened on 2026-09-18. The vehicle is drivable and no injuries were reported."},
    {"external_id": "CLM-1003", "sender_email": "taylor.morgan@example.com", "received_at": "2026-09-18T22:15:00Z", "subject": "Missing suitcase", "body": "My checked suitcase did not arrive after my flight on 2026-09-17. I do not have my travel policy number with me."},
    {"external_id": "CLM-1004", "sender_email": "maria.santos.example.com", "received_at": "2026-09-18T23:40:00Z", "subject": "Bicycle stolen from garage", "body": "Policy POL-H40004. My locked bicycle was taken from the shared garage on 2026-09-18. I have filed a police report."},
    {"external_id": "CLM-1005", "sender_email": "noah.patel@example.com", "received_at": "yesterday evening", "subject": "Storm damage to parked car", "body": "Policy POL-A50005. A tree branch cracked the windshield while the car was parked. The incident date was 2026-09-18."},
]

What is inside this collection?

• Each claim has an external ID plus the original email fields.
• The records deliberately preserve source inconsistencies.
• The collection stays in memory so the local API can return the same demonstration data on every request.

• Save main.py.
• Return to the running Terminal tab to confirm that Uvicorn reloads the service.

You should see fresh startup activity after the saved change. This confirms that Python loaded the six-record collection successfully.

Service stopped after adding the claims?

• Check that every claim dictionary ends with a comma.
• Confirm that the collection ends with the closing ] character.

Need another pair of eyes? Help me find the syntax problem in my SYNTHETIC_CLAIMS collection.

Routes connect browser requests to Python functions. The root route acts as a small directory while the source route returns the claims as JSON.

• Add the root route plus the source route below the SYNTHETIC_CLAIMS collection by pasting this code:

@app.get("/")
def root():
    return {"service": "ClaimFlow AI", "status": "ready", "documentation": "/docs", "source_claims": "/source/claims", "processed_claims": "/claims", "pipeline_runs": "/runs"}

@app.get("/source/claims")
def source_claims():
    return {"count": len(SYNTHETIC_CLAIMS), "claims": SYNTHETIC_CLAIMS}

How do these routes work?

• The root() function returns the service status plus links to each API path.
• The source_claims() function calculates the record count from SYNTHETIC_CLAIMS.
• FastAPI converts the returned dictionaries into JSON responses.

• Save main.py.
• Open http://127.0.0.1:8000 in your browser.

You should see ClaimFlow AI with a ready status. The response should also list paths for documentation plus source claims plus processed claims plus pipeline runs.

Root response missing a route?

• Confirm that the root decorator contains exactly /.
• Check that the returned dictionary includes source_claims plus processed_claims plus pipeline_runs.

Still missing a route? Help me compare my ClaimFlow root route with the expected response.

Add the processed endpoints

The source endpoint works without a database. The processed endpoints depend on tables that the next pipeline step creates in Neon Postgres.

These routes catch the temporary missing-table condition. That keeps the API readable while the project is still being assembled.

• Add the database helper plus the processed claims route below source_claims() by pasting this code:

def get_database_url():
    return os.getenv("DATABASE_URL")

@app.get("/claims")
def list_claims():
    database_url = get_database_url()
    if not database_url:
        return {"status": "not_configured", "claims": []}
    try:
        with psycopg.connect(database_url) as conn:
            with conn.cursor() as cur:
                cur.execute("""SELECT external_id, sender_email, subject, received_at_raw, source_copy_count, data_quality_issues, policy_number, incident_date, claim_type, urgency, summary, missing_information, needs_human_review, processed_at, run_id FROM processed_claims ORDER BY external_id;""")
                rows = cur.fetchall()
    except Exception as exc:
        return {"status": "not_ready", "detail": type(exc).__name__, "claims": []}
    claims = []
    for row in rows:
        claims.append({"external_id": row[0], "sender_email": row[1], "subject": row[2], "received_at_raw": row[3], "source_copy_count": row[4], "data_quality_issues": json.loads(row[5]), "policy_number": row[6], "incident_date": row[7], "claim_type": row[8], "urgency": row[9], "summary": row[10], "missing_information": json.loads(row[11]), "needs_human_review": row[12], "processed_at": str(row[13]), "run_id": row[14]})
    return {"status": "ready", "count": len(claims), "claims": claims}

What does the claims route do?

• The get_database_url() helper reads the existing connection string from the environment.
• The route queries processed_claims when the table exists.
• The exception branch returns a readable waiting state while the table is absent.
• The final loop converts database rows into API claim objects.

• Save main.py.

Before you visit the route, do you expect processed claims to exist before the pipeline has created its tables?

• Open http://127.0.0.1:8000/claims in your browser.

You should see a not_ready response with an empty claims list. This temporary result proves that the route is handling the unfinished database state.

Claims route returning a server error?

• Confirm that the database call sits inside the try block.
• Check that the except Exception as exc block aligns with try.

Need help with the route structure? Help me debug why my /claims route raises an unhandled error.

• Add the pipeline runs route below list_claims() by pasting this code:

@app.get("/runs")
def list_runs():
    database_url = get_database_url()
    if not database_url:
        return {"status": "not_configured", "runs": []}
    try:
        with psycopg.connect(database_url) as conn:
            with conn.cursor() as cur:
                cur.execute("""SELECT run_id, completed_at, source_count, clean_count, review_count, status FROM pipeline_runs ORDER BY completed_at DESC LIMIT 10;""")
                rows = cur.fetchall()
    except Exception as exc:
        return {"status": "not_ready", "detail": type(exc).__name__, "runs": []}
    runs = [{"run_id": row[0], "completed_at": str(row[1]), "source_count": row[2], "clean_count": row[3], "review_count": row[4], "status": row[5]} for row in rows]
    return {"status": "ready", "count": len(runs), "runs": runs}

What does the runs route do?

• The route reads the ten most recent entries from pipeline_runs.
• The temporary response keeps the endpoint available before that table exists.
• Once the pipeline runs, the response exposes counts plus completion status for each run.

• Save main.py.
• Open http://127.0.0.1:8000/runs in your browser.

You should see a not_ready response with an empty runs list. The pipeline creates the missing table in the next step.

Runs route not loading?

• Confirm that the decorator contains /runs.
• Check that the query references pipeline_runs.
• Verify that the final response uses the runs variable.

Still stuck? Help me find why my /runs endpoint is not returning its waiting state.

✔️ Awesome, I've got everything!

Your source plus placeholder routes are in place. Double check that main.py is saved.

ⓧ I'd like to double check the full code

Compare your complete main.py file with this reference.

import json
import os

import psycopg
from dotenv import load_dotenv
from fastapi import FastAPI

load_dotenv()

app = FastAPI(
    title="ClaimFlow AI",
    version="1.0.0",
    description="Synthetic insurance claim ingestion, cleaning, and human-reviewed AI triage.",
)

SYNTHETIC_CLAIMS = [
    {"external_id": "CLM-1001", "sender_email": "alex.rivera@example.com", "received_at": "2026-09-18T20:10:00Z", "subject": "Kitchen pipe leak", "body": "Policy POL-H10001. A pipe burst under the kitchen sink on 2026-09-18. Water reached two cabinets. The supply is shut off."},
    {"external_id": "CLM-1002", "sender_email": "jamie.chen@example.com", "received_at": "2026-09-18T21:05:00Z", "subject": "Car accident claim", "body": "My vehicle was hit from behind today. Policy POL-A20002. No injuries were reported. The rear bumper is damaged."},
    {"external_id": "CLM-1002", "sender_email": "Jamie.Chen@Example.com ", "received_at": "2026-09-18T21:22:00Z", "subject": "Update: car accident claim", "body": "Update for policy POL-A20002: the collision happened on 2026-09-18. The vehicle is drivable and no injuries were reported."},
    {"external_id": "CLM-1003", "sender_email": "taylor.morgan@example.com", "received_at": "2026-09-18T22:15:00Z", "subject": "Missing suitcase", "body": "My checked suitcase did not arrive after my flight on 2026-09-17. I do not have my travel policy number with me."},
    {"external_id": "CLM-1004", "sender_email": "maria.santos.example.com", "received_at": "2026-09-18T23:40:00Z", "subject": "Bicycle stolen from garage", "body": "Policy POL-H40004. My locked bicycle was taken from the shared garage on 2026-09-18. I have filed a police report."},
    {"external_id": "CLM-1005", "sender_email": "noah.patel@example.com", "received_at": "yesterday evening", "subject": "Storm damage to parked car", "body": "Policy POL-A50005. A tree branch cracked the windshield while the car was parked. The incident date was 2026-09-18."},
]

@app.get("/")
def root():
    return {"service": "ClaimFlow AI", "status": "ready", "documentation": "/docs", "source_claims": "/source/claims", "processed_claims": "/claims", "pipeline_runs": "/runs"}

@app.get("/source/claims")
def source_claims():
    return {"count": len(SYNTHETIC_CLAIMS), "claims": SYNTHETIC_CLAIMS}

def get_database_url():
    return os.getenv("DATABASE_URL")

@app.get("/claims")
def list_claims():
    database_url = get_database_url()
    if not database_url:
        return {"status": "not_configured", "claims": []}
    try:
        with psycopg.connect(database_url) as conn:
            with conn.cursor() as cur:
                cur.execute("""SELECT external_id, sender_email, subject, received_at_raw, source_copy_count, data_quality_issues, policy_number, incident_date, claim_type, urgency, summary, missing_information, needs_human_review, processed_at, run_id FROM processed_claims ORDER BY external_id;""")
                rows = cur.fetchall()
    except Exception as exc:
        return {"status": "not_ready", "detail": type(exc).__name__, "claims": []}
    claims = []
    for row in rows:
        claims.append({"external_id": row[0], "sender_email": row[1], "subject": row[2], "received_at_raw": row[3], "source_copy_count": row[4], "data_quality_issues": json.loads(row[5]), "policy_number": row[6], "incident_date": row[7], "claim_type": row[8], "urgency": row[9], "summary": row[10], "missing_information": json.loads(row[11]), "needs_human_review": row[12], "processed_at": str(row[13]), "run_id": row[14]})
    return {"status": "ready", "count": len(claims), "claims": claims}

@app.get("/runs")
def list_runs():
    database_url = get_database_url()
    if not database_url:
        return {"status": "not_configured", "runs": []}
    try:
        with psycopg.connect(database_url) as conn:
            with conn.cursor() as cur:
                cur.execute("""SELECT run_id, completed_at, source_count, clean_count, review_count, status FROM pipeline_runs ORDER BY completed_at DESC LIMIT 10;""")
                rows = cur.fetchall()
    except Exception as exc:
        return {"status": "not_ready", "detail": type(exc).__name__, "runs": []}
    runs = [{"run_id": row[0], "completed_at": str(row[1]), "source_count": row[2], "clean_count": row[3], "review_count": row[4], "status": row[5]} for row in rows]
    return {"status": "ready", "count": len(runs), "runs": runs}

This reference includes the complete source collection plus all five route functions. Its order matches the finished file.

Inspect the messy response

The interactive documentation lets you send requests without another client. This is where the source data reveals the defects that the cleaning pipeline must handle.

• Return to http://127.0.0.1:8000/docs in your browser.
• Select GET /claims in the interactive documentation.
• Use the request control to send the claims request.
• Select GET /runs in the interactive documentation.
• Use the request control to send the runs request.

Both processed endpoints should return a clear waiting state. This confirms that the API remains usable before the Neon tables exist.

Before you send the source request, how many records do you expect to see? Look closely for repeated identities plus inconsistent field formats.

• Select GET /source/claims in the interactive documentation.
• Use the request control to send the source request.

You should see six records. Two of those records use the external ID CLM-1002.

• Compare the two CLM-1002 records.
• Find the malformed sender value maria.santos.example.com in CLM-1004.
• Read the body of CLM-1003 to identify the missing policy number.
• Find the received_at value yesterday evening in CLM-1005.

Why keep the defects visible?

The API preserves what the source system sent. That evidence lets the next pipeline distinguish ingestion from cleaning.

The duplicate count plus malformed fields become measurable quality signals. The pipeline can route uncertain records to people instead of silently trusting them.

That first API is live. You can now show six raw records plus the exact quality problems hidden inside them.

Next up, you will move those records into PostgreSQL. You will use window functions to reduce the six source rows to five unique claims.

Deduplicate Claims with PostgreSQL

Your source API now exposes six claim records. The duplicate ID and malformed values show why the raw feed cannot move directly into triage.

This step moves the records into Neon. You will use PostgreSQL window functions to preserve five unique claims while keeping evidence of every source defect.

In this step, get ready to:

• Fetch the six source records from the running ClaimFlow API.
• Create the three database tables and stage the raw records with parameterized inserts.
• Deduplicate the records and report their data quality issues.

Build the database contract

The pipeline needs a stable database shape before it can move any records. The raw table preserves incoming evidence while the other tables prepare for processed claims and run history.

The cleaning query uses two window functions. One counts every copy of an external ID. The other ranks each copy so the newest row survives.

• In the existing empty pipeline.py file, add the imports and pipeline settings by copying this code:

import json
import os
import sys
import urllib.request
import uuid

import psycopg
from anthropic import Anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()
MODEL_ID = "claude-sonnet-5"
DEFAULT_SOURCE_API_URL = "http://127.0.0.1:8000/source/claims"
ALLOWED_CLAIM_TYPES = {"auto", "property", "travel", "liability", "other"}
ALLOWED_URGENCY_LEVELS = {"low", "medium", "high"}

What does this setup do?

• The imports provide database access, HTTP requests, JSON handling, environment loading, and unique run IDs.
• The source URL points to the local endpoint that Uvicorn is already serving.
• The model and category constants prepare the same file for the structured triage step.

• Save pipeline.py.
• Confirm your editor recognizes pipeline.py as a Python file.

Seeing import warnings?

Confirm that your existing virtual environment is still active. The environment contains the pinned packages installed in Step 1.

If an import remains highlighted, make sure your editor is using the interpreter inside .venv.

Still stuck? Help me check the Python interpreter selected for this project.

• Below the constants in pipeline.py, add the table schemas and cleaning query by copying this code:

SCHEMA_STATEMENTS = (
    """CREATE TABLE IF NOT EXISTS raw_claims (id BIGSERIAL PRIMARY KEY, external_id TEXT NOT NULL, sender_email TEXT NOT NULL, received_at_raw TEXT NOT NULL, subject TEXT NOT NULL, body TEXT NOT NULL, run_id TEXT NOT NULL, ingested_at TIMESTAMPTZ NOT NULL DEFAULT NOW());""",
    """CREATE TABLE IF NOT EXISTS processed_claims (external_id TEXT PRIMARY KEY, sender_email TEXT NOT NULL, subject TEXT NOT NULL, received_at_raw TEXT NOT NULL, source_copy_count INTEGER NOT NULL, data_quality_issues TEXT NOT NULL, policy_number TEXT NOT NULL, incident_date TEXT NOT NULL, claim_type TEXT NOT NULL, urgency TEXT NOT NULL, summary TEXT NOT NULL, missing_information TEXT NOT NULL, needs_human_review BOOLEAN NOT NULL, processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), run_id TEXT NOT NULL);""",
    """CREATE TABLE IF NOT EXISTS pipeline_runs (run_id TEXT PRIMARY KEY, completed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), source_count INTEGER NOT NULL, clean_count INTEGER NOT NULL, review_count INTEGER NOT NULL, status TEXT NOT NULL);""",
)

CLEAN_QUERY = """
WITH ranked AS (
    SELECT external_id, sender_email, subject, body, received_at_raw,
           COUNT(*) OVER (PARTITION BY external_id) AS source_copy_count,
           ROW_NUMBER() OVER (PARTITION BY external_id ORDER BY id DESC) AS duplicate_rank
    FROM raw_claims
)
SELECT external_id, sender_email, subject, body, received_at_raw, source_copy_count
FROM ranked WHERE duplicate_rank = 1 ORDER BY external_id;
"""

How does the database contract work?

• The raw_claims table stores each source row before cleaning.
• The processed_claims table has the columns needed by the later triage stage.
• The pipeline_runs table records metrics for each completed pipeline run.
• The CLEAN_QUERY common table expression calculates duplicate evidence before selecting one row per claim ID.

• Save pipeline.py.
• Confirm the closing triple quotes for CLEAN_QUERY appear beneath the final query line.

Seeing broken SQL highlighting?

Check that each schema statement starts and ends with three double quotes. Confirm that the surrounding tuple ends with a closing parenthesis.

Make sure CLEAN_QUERY also starts and ends with three double quotes.

Still stuck? Help me compare the quote boundaries in my schema and cleaning query.

Stage and clean the claims

The database contract defines where the records belong. The next functions read the configured source URL and replace the staging data for a predictable demo run.

A missing database connection must stop the pipeline before it touches data. The environment guard makes that failure immediate and specific.

• Below CLEAN_QUERY, add the environment guard by copying this function:

def require_environment_variable(name):
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value

What does this guard protect?

The helper retrieves a required setting from the environment. It stops the run with a named error when that setting is empty.

• Save pipeline.py.
• Confirm the return value line remains indented inside require_environment_variable().

Seeing an indentation warning?

Keep every line after the function definition indented by four spaces. Keep the error line indented inside the condition.

Still stuck? Help me correct the indentation in this environment helper.

The source service returns a JSON object containing the claims list. The fetch helper opens that endpoint and returns only the records the database needs.

• Below require_environment_variable(), add the source fetcher by copying this function:

def fetch_source_claims(source_url):
    with urllib.request.urlopen(source_url, timeout=120) as response:
        payload = json.load(response)
    return payload["claims"]

How does the fetcher work?

• The URL opener requests the running /source/claims endpoint.
• The timeout prevents the request from waiting forever when the source service cannot respond.
• The JSON loader converts the response into Python data.

• Save pipeline.py.
• Confirm return payload["claims"] aligns with the function body.

Is the source fetcher highlighted?

Check that the response line is indented inside the context manager. Keep the return line inside the function but outside the context manager.

Still stuck? Help me check the indentation in my source fetcher.

The staging function performs the database work. It creates the tables, replaces the raw rows, executes the cleaning query, and attaches quality flags to the retained claims.

The insert uses parameterized SQL placeholders. Psycopg sends record values separately from the SQL statement.

• Below fetch_source_claims(), add the staging and cleaning function by copying this code:

def stage_and_clean_claims(database_url, source_claims, run_id):
    with psycopg.connect(database_url) as conn:
        with conn.cursor() as cur:
            for statement in SCHEMA_STATEMENTS:
                cur.execute(statement)
            cur.execute("DELETE FROM raw_claims;")
            cur.executemany("""INSERT INTO raw_claims (external_id, sender_email, received_at_raw, subject, body, run_id) VALUES (%s, %s, %s, %s, %s, %s);""", [(claim["external_id"], claim["sender_email"], claim["received_at"], claim["subject"], claim["body"], run_id) for claim in source_claims])
            cur.execute(CLEAN_QUERY)
            rows = cur.fetchall()
    clean_claims = []
    for row in rows:
        normalized_email = row[1].strip().lower()
        issues = []
        if row[5] > 1:
            issues.append("duplicate_external_id")
        if normalized_email.count("@") != 1:
            issues.append("invalid_sender_email")
        if "T" not in row[4] or not row[4].endswith("Z"):
            issues.append("invalid_received_at")
        clean_claims.append({"external_id": row[0], "sender_email": normalized_email, "subject": row[2], "body": row[3], "received_at_raw": row[4], "source_copy_count": row[5], "data_quality_issues": issues})
    return clean_claims

What happens during staging?

• The connection context commits the database changes after successful execution.
• The raw table is cleared before every demo run so old source rows cannot inflate the duplicate count.
• The parameterized insert writes all six source records without joining their values into the SQL text.
• The cleaning loop normalizes sender casing and records each detected source defect.

• Save pipeline.py.
• Confirm the function ends with return clean_claims.

Seeing a long line warning?

The parameterized insert and final dictionary are intentionally long lines from the project source. Keep their quotes and parentheses unchanged.

If your editor reports a syntax error, check the closing parenthesis on cur.executemany().

Still stuck? Help me locate an unmatched quote or parenthesis in this staging function.

Run the clean preview

The cleaned records now contain the evidence needed for a useful preview. The final functions print that evidence and connect the clean-only command to the pipeline.

• Below stage_and_clean_claims(), add the preview printer by copying this function:

def print_clean_preview(source_claims, clean_claims):
    print(f"Clean preview: {len(source_claims)} source rows -> {len(clean_claims)} unique claims")
    for claim in clean_claims:
        print(json.dumps({"external_id": claim["external_id"], "source_copy_count": claim["source_copy_count"], "data_quality_issues": claim["data_quality_issues"]}))

What does the preview reveal?

The first line compares the raw source count with the unique claim count. Each following line keeps the claim ID beside its copy count and quality flags.

• Save pipeline.py.
• Confirm the print() call inside the loop is indented beneath for claim in clean_claims.

Is the loop indentation uneven?

Keep the first print line directly inside the function. Indent the second print line one level deeper than the loop.

Still stuck? Help me correct the indentation in the preview function.

• Below print_clean_preview(), add the clean-only pipeline path by copying this code:

def run_pipeline(clean_only=False):
    database_url = require_environment_variable("DATABASE_URL")
    source_url = os.getenv("SOURCE_API_URL", DEFAULT_SOURCE_API_URL)
    run_id = str(uuid.uuid4())
    source_claims = fetch_source_claims(source_url)
    clean_claims = stage_and_clean_claims(database_url, source_claims, run_id)
    if clean_only:
        print_clean_preview(source_claims, clean_claims)
        return

if __name__ == "__main__":
    run_pipeline(clean_only="--clean-only" in sys.argv[1:])

How does the clean-only path connect?

• The run loads the Neon connection string from the existing environment configuration.
• A fresh run ID identifies the staged rows from this execution.
• The pipeline fetches the source claims before passing them into the staging function.
• The command-line check prints the clean preview when the clean-only flag is present.

• Save pipeline.py.
• Confirm the final line remains outside the run_pipeline() function.

Does the file end inside the function?

The main guard starts at the left edge of the file. Its final line is indented by four spaces.

Still stuck? Help me place the main guard at the correct indentation level.

✔️ Awesome, I've got everything!

Great. Double check that you saved pipeline.py before running the clean preview.

ⓧ I'd like to double check the full code

Compare your pipeline.py file with this complete clean-only version.

import json
import os
import sys
import urllib.request
import uuid

import psycopg
from anthropic import Anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()
MODEL_ID = "claude-sonnet-5"
DEFAULT_SOURCE_API_URL = "http://127.0.0.1:8000/source/claims"
ALLOWED_CLAIM_TYPES = {"auto", "property", "travel", "liability", "other"}
ALLOWED_URGENCY_LEVELS = {"low", "medium", "high"}

SCHEMA_STATEMENTS = (
    """CREATE TABLE IF NOT EXISTS raw_claims (id BIGSERIAL PRIMARY KEY, external_id TEXT NOT NULL, sender_email TEXT NOT NULL, received_at_raw TEXT NOT NULL, subject TEXT NOT NULL, body TEXT NOT NULL, run_id TEXT NOT NULL, ingested_at TIMESTAMPTZ NOT NULL DEFAULT NOW());""",
    """CREATE TABLE IF NOT EXISTS processed_claims (external_id TEXT PRIMARY KEY, sender_email TEXT NOT NULL, subject TEXT NOT NULL, received_at_raw TEXT NOT NULL, source_copy_count INTEGER NOT NULL, data_quality_issues TEXT NOT NULL, policy_number TEXT NOT NULL, incident_date TEXT NOT NULL, claim_type TEXT NOT NULL, urgency TEXT NOT NULL, summary TEXT NOT NULL, missing_information TEXT NOT NULL, needs_human_review BOOLEAN NOT NULL, processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), run_id TEXT NOT NULL);""",
    """CREATE TABLE IF NOT EXISTS pipeline_runs (run_id TEXT PRIMARY KEY, completed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), source_count INTEGER NOT NULL, clean_count INTEGER NOT NULL, review_count INTEGER NOT NULL, status TEXT NOT NULL);""",
)

CLEAN_QUERY = """
WITH ranked AS (
    SELECT external_id, sender_email, subject, body, received_at_raw,
           COUNT(*) OVER (PARTITION BY external_id) AS source_copy_count,
           ROW_NUMBER() OVER (PARTITION BY external_id ORDER BY id DESC) AS duplicate_rank
    FROM raw_claims
)
SELECT external_id, sender_email, subject, body, received_at_raw, source_copy_count
FROM ranked WHERE duplicate_rank = 1 ORDER BY external_id;
"""

def require_environment_variable(name):
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value

def fetch_source_claims(source_url):
    with urllib.request.urlopen(source_url, timeout=120) as response:
        payload = json.load(response)
    return payload["claims"]

def stage_and_clean_claims(database_url, source_claims, run_id):
    with psycopg.connect(database_url) as conn:
        with conn.cursor() as cur:
            for statement in SCHEMA_STATEMENTS:
                cur.execute(statement)
            cur.execute("DELETE FROM raw_claims;")
            cur.executemany("""INSERT INTO raw_claims (external_id, sender_email, received_at_raw, subject, body, run_id) VALUES (%s, %s, %s, %s, %s, %s);""", [(claim["external_id"], claim["sender_email"], claim["received_at"], claim["subject"], claim["body"], run_id) for claim in source_claims])
            cur.execute(CLEAN_QUERY)
            rows = cur.fetchall()
    clean_claims = []
    for row in rows:
        normalized_email = row[1].strip().lower()
        issues = []
        if row[5] > 1:
            issues.append("duplicate_external_id")
        if normalized_email.count("@") != 1:
            issues.append("invalid_sender_email")
        if "T" not in row[4] or not row[4].endswith("Z"):
            issues.append("invalid_received_at")
        clean_claims.append({"external_id": row[0], "sender_email": normalized_email, "subject": row[2], "body": row[3], "received_at_raw": row[4], "source_copy_count": row[5], "data_quality_issues": issues})
    return clean_claims

def print_clean_preview(source_claims, clean_claims):
    print(f"Clean preview: {len(source_claims)} source rows -> {len(clean_claims)} unique claims")
    for claim in clean_claims:
        print(json.dumps({"external_id": claim["external_id"], "source_copy_count": claim["source_copy_count"], "data_quality_issues": claim["data_quality_issues"]}))

def run_pipeline(clean_only=False):
    database_url = require_environment_variable("DATABASE_URL")
    source_url = os.getenv("SOURCE_API_URL", DEFAULT_SOURCE_API_URL)
    run_id = str(uuid.uuid4())
    source_claims = fetch_source_claims(source_url)
    clean_claims = stage_and_clean_claims(database_url, source_claims, run_id)
    if clean_only:
        print_clean_preview(source_claims, clean_claims)
        return

if __name__ == "__main__":
    run_pipeline(clean_only="--clean-only" in sys.argv[1:])

Before you run the pipeline, do you expect the two CLM-1002 records to remain separate?

• Switch to a second Terminal tab while Uvicorn keeps running in the first tab.
• Activate the existing .venv environment using the same command from Step 1.
• Run the clean-only pipeline with this command:

python pipeline.py --clean-only

What should you see?

The first line reads Clean preview: 6 source rows -> 5 unique claims. That confirms the duplicate source ID became one retained claim.

The retained CLM-1002 record has a source_copy_count of 2. Its quality issues include duplicate_external_id.

The CLM-1004 record reports invalid_sender_email. The CLM-1005 record reports invalid_received_at.

Clean preview did not complete?

If the request cannot connect, confirm Uvicorn is still running in the first Terminal tab. Confirm the source claims endpoint still responds in your browser.

If the database connection fails, check that your local .env contains the pooled Neon connection string from Step 1. Keep the credential private.

Still stuck? Help me diagnose my clean-only pipeline failure.

That is the data-cleaning core working. Your pipeline now preserves five unique claims while exposing the defects that still need attention. Next, you will add structured Claude triage and deterministic human-review rules.

Add Claude Triage and Human Review

Your PostgreSQL cleaner now reduces six source rows to five records. Important claim details still sit inside free-form email text.

In this step, Claude maps that text into a Pydantic model. Deterministic Python rules send uncertain records to people for review.

In this step, get ready to:

• Define a typed schema for the extracted claim fields.
• Add deterministic guardrails for human review.
• Persist the completed triage run in Neon.

Define the triage schema

A structured output asks Claude to return fields that match a schema. Pydantic validates that shape before the pipeline uses it.

The schema becomes the contract for response.parsed_output. Each completed response must provide every annotated field.

• In pipeline.py, find the existing from dotenv import load_dotenv import.
• Add the Anthropic import immediately above that line.
• Add the BaseModel import immediately below that line.
• Compare the updated import group with this code:

from anthropic import Anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

What do these imports provide?

• Anthropic creates the client that sends each cleaned claim to Claude.
• BaseModel defines the validated structure expected from each response.
• load_dotenv continues loading the protected credentials from .env.

• Find the end of CLEAN_QUERY in pipeline.py.
• Add this schema below CLEAN_QUERY by copying the code:

class ClaimTriage(BaseModel):
    policy_number: str
    incident_date: str
    claim_type: str
    urgency: str
    summary: str
    missing_information: list[str]

What does this schema validate?

• policy_number plus incident_date capture the identifiers needed to understand the event.
• claim_type plus urgency organize the operational triage.
• summary gives a concise description of the claim.
• missing_information records details that a person still needs to collect.

• Save pipeline.py.
• Confirm that ClaimTriage contains six annotated fields.

Seeing unresolved imports?

Return to the activated virtual environment from earlier. Confirm that your editor is using its Python interpreter.

Still stuck? Help me fix unresolved Anthropic or Pydantic imports in my ClaimFlow virtual environment.

Add extraction and review guardrails

The extraction prompt gives Claude a narrow role. It allows information extraction plus urgency triage.

Python checks the returned categories against ALLOWED_CLAIM_TYPES plus ALLOWED_URGENCY_LEVELS. Any source defect or unresolved field activates human review.

• Find the end of stage_and_clean_claims() in pipeline.py.
• Add triage_claim() immediately below it by copying the code:

def triage_claim(client, claim):
    prompt = f"""You are extracting fields from a fictional insurance claim email for a software demo.
Do not approve or deny coverage. Do not assign liability. Do not detect fraud. Do not recommend or authorize payment. Extract information and identify what a human must review.
Use one of these lowercase claim types: auto, property, travel, liability, other.
Use one of these lowercase urgency levels: low, medium, high.
Use the string unknown when the policy number or incident date is not present.
Keep the summary to one sentence. List every important missing detail in missing_information.

Claim record:
{json.dumps(claim, indent=2)}"""
    response = client.messages.parse(model=MODEL_ID, max_tokens=1024, messages=[{"role": "user", "content": prompt}], output_format=ClaimTriage)
    triage = response.parsed_output
    if triage is None:
        raise RuntimeError("Claude returned no parsed output")
    data = triage.model_dump()
    policy_number = data["policy_number"].strip() or "unknown"
    incident_date = data["incident_date"].strip() or "unknown"
    claim_type = data["claim_type"].strip().lower()
    urgency = data["urgency"].strip().lower()
    missing_information = [item.strip() for item in data["missing_information"] if item.strip()]
    if claim_type not in ALLOWED_CLAIM_TYPES:
        claim_type = "other"
        missing_information.append("claim type requires confirmation")
    if urgency not in ALLOWED_URGENCY_LEVELS:
        urgency = "medium"
        missing_information.append("urgency requires confirmation")
    needs_human_review = bool(claim["data_quality_issues"] or missing_information or policy_number.lower() == "unknown" or incident_date.lower() == "unknown")
    return {**claim, "policy_number": policy_number, "incident_date": incident_date, "claim_type": claim_type, "urgency": urgency, "summary": data["summary"].strip(), "missing_information": missing_information, "needs_human_review": needs_human_review}

How does the triage stay constrained?

• client.messages.parse() sends the cleaned claim to claude-sonnet-5 with a limit of 1024 output tokens.
• output_format=ClaimTriage asks the SDK to return a validated Pydantic value.
• model_dump() converts the validated result into a dictionary.
• needs_human_review becomes true when a source defect or unresolved field remains.

Where is the human boundary?

The prompt prohibits coverage decisions. It also prohibits liability assignments.

The prompt prohibits fraud detection plus payment authorization. Deterministic code keeps questionable records visible to people.

• Save pipeline.py.
• Confirm that triage_claim() ends by returning needs_human_review with the extracted fields.

Seeing a syntax warning in the prompt?

Check that the prompt starts plus ends with three double quotes. Keep the json.dumps(claim, indent=2) expression inside the braces.

Still stuck? Help me compare my triage_claim function with the required ClaimFlow structure.

Save and inspect the triage run

The enriched claims need durable storage in Neon. The same transaction records metrics that connect the source count to the cleaned result.

• Add save_processed_claims() immediately below triage_claim() by copying the code:

def save_processed_claims(database_url, processed_claims, run_id, source_count):
    review_count = sum(1 for claim in processed_claims if claim["needs_human_review"])
    with psycopg.connect(database_url) as conn:
        with conn.cursor() as cur:
            cur.execute("DELETE FROM processed_claims;")
            cur.executemany("""INSERT INTO processed_claims (external_id, sender_email, subject, received_at_raw, source_copy_count, data_quality_issues, policy_number, incident_date, claim_type, urgency, summary, missing_information, needs_human_review, run_id) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);""", [(claim["external_id"], claim["sender_email"], claim["subject"], claim["received_at_raw"], claim["source_copy_count"], json.dumps(claim["data_quality_issues"]), claim["policy_number"], claim["incident_date"], claim["claim_type"], claim["urgency"], claim["summary"], json.dumps(claim["missing_information"]), claim["needs_human_review"], run_id) for claim in processed_claims])
            cur.execute("""INSERT INTO pipeline_runs (run_id, source_count, clean_count, review_count, status) VALUES (%s, %s, %s, %s, %s);""", (run_id, source_count, len(processed_claims), review_count, "successful"))
    return review_count

What does the database write preserve?

• review_count counts the claims that require manual attention.
• executemany() writes the five processed claims with parameterized placeholders.
• pipeline_runs receives the source count plus clean count plus review count.
• successful records the completed run status.

• Find the existing run_pipeline() function near the bottom of pipeline.py.
• Replace that function with this completed version:

def run_pipeline(clean_only=False):
    database_url = require_environment_variable("DATABASE_URL")
    source_url = os.getenv("SOURCE_API_URL", DEFAULT_SOURCE_API_URL)
    run_id = str(uuid.uuid4())
    source_claims = fetch_source_claims(source_url)
    clean_claims = stage_and_clean_claims(database_url, source_claims, run_id)
    if clean_only:
        print_clean_preview(source_claims, clean_claims)
        return
    require_environment_variable("ANTHROPIC_API_KEY")
    client = Anthropic()
    processed_claims = [triage_claim(client, claim) for claim in clean_claims]
    review_count = save_processed_claims(database_url, processed_claims, run_id, len(source_claims))
    print(f"Pipeline complete: {len(source_claims)} source rows -> {len(processed_claims)} clean claims; {review_count} need human review.")

How does the full run flow?

• run_pipeline() still supports the earlier clean-only path.
• Anthropic() reads the protected API key from the environment.
• triage_claim() enriches each of the five cleaned claims.
• save_processed_claims() stores the results plus one run-history record.

• Save pipeline.py.

✔️ Awesome, I've got everything!

Great. Your schema plus extraction logic plus review rules plus database write are in place.

ⓧ I'd like to double check the full code

This reference shows the cumulative pipeline.py file for this checkpoint.

import json
import os
import sys
import urllib.request
import uuid

import psycopg
from anthropic import Anthropic
from dotenv import load_dotenv
from pydantic import BaseModel

load_dotenv()
MODEL_ID = "claude-sonnet-5"
DEFAULT_SOURCE_API_URL = "http://127.0.0.1:8000/source/claims"
ALLOWED_CLAIM_TYPES = {"auto", "property", "travel", "liability", "other"}
ALLOWED_URGENCY_LEVELS = {"low", "medium", "high"}

SCHEMA_STATEMENTS = (
    """CREATE TABLE IF NOT EXISTS raw_claims (id BIGSERIAL PRIMARY KEY, external_id TEXT NOT NULL, sender_email TEXT NOT NULL, received_at_raw TEXT NOT NULL, subject TEXT NOT NULL, body TEXT NOT NULL, run_id TEXT NOT NULL, ingested_at TIMESTAMPTZ NOT NULL DEFAULT NOW());""",
    """CREATE TABLE IF NOT EXISTS processed_claims (external_id TEXT PRIMARY KEY, sender_email TEXT NOT NULL, subject TEXT NOT NULL, received_at_raw TEXT NOT NULL, source_copy_count INTEGER NOT NULL, data_quality_issues TEXT NOT NULL, policy_number TEXT NOT NULL, incident_date TEXT NOT NULL, claim_type TEXT NOT NULL, urgency TEXT NOT NULL, summary TEXT NOT NULL, missing_information TEXT NOT NULL, needs_human_review BOOLEAN NOT NULL, processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), run_id TEXT NOT NULL);""",
    """CREATE TABLE IF NOT EXISTS pipeline_runs (run_id TEXT PRIMARY KEY, completed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), source_count INTEGER NOT NULL, clean_count INTEGER NOT NULL, review_count INTEGER NOT NULL, status TEXT NOT NULL);""",
)

CLEAN_QUERY = """
WITH ranked AS (
    SELECT external_id, sender_email, subject, body, received_at_raw,
           COUNT(*) OVER (PARTITION BY external_id) AS source_copy_count,
           ROW_NUMBER() OVER (PARTITION BY external_id ORDER BY id DESC) AS duplicate_rank
    FROM raw_claims
)
SELECT external_id, sender_email, subject, body, received_at_raw, source_copy_count
FROM ranked WHERE duplicate_rank = 1 ORDER BY external_id;
"""

class ClaimTriage(BaseModel):
    policy_number: str
    incident_date: str
    claim_type: str
    urgency: str
    summary: str
    missing_information: list[str]

def require_environment_variable(name):
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Missing required environment variable: {name}")
    return value

def fetch_source_claims(source_url):
    with urllib.request.urlopen(source_url, timeout=120) as response:
        payload = json.load(response)
    return payload["claims"]

def stage_and_clean_claims(database_url, source_claims, run_id):
    with psycopg.connect(database_url) as conn:
        with conn.cursor() as cur:
            for statement in SCHEMA_STATEMENTS:
                cur.execute(statement)
            cur.execute("DELETE FROM raw_claims;")
            cur.executemany("""INSERT INTO raw_claims (external_id, sender_email, received_at_raw, subject, body, run_id) VALUES (%s, %s, %s, %s, %s, %s);""", [(claim["external_id"], claim["sender_email"], claim["received_at"], claim["subject"], claim["body"], run_id) for claim in source_claims])
            cur.execute(CLEAN_QUERY)
            rows = cur.fetchall()
    clean_claims = []
    for row in rows:
        normalized_email = row[1].strip().lower()
        issues = []
        if row[5] > 1:
            issues.append("duplicate_external_id")
        if normalized_email.count("@") != 1:
            issues.append("invalid_sender_email")
        if "T" not in row[4] or not row[4].endswith("Z"):
            issues.append("invalid_received_at")
        clean_claims.append({"external_id": row[0], "sender_email": normalized_email, "subject": row[2], "body": row[3], "received_at_raw": row[4], "source_copy_count": row[5], "data_quality_issues": issues})
    return clean_claims

def triage_claim(client, claim):
    prompt = f"""You are extracting fields from a fictional insurance claim email for a software demo.
Do not approve or deny coverage. Do not assign liability. Do not detect fraud. Do not recommend or authorize payment. Extract information and identify what a human must review.
Use one of these lowercase claim types: auto, property, travel, liability, other.
Use one of these lowercase urgency levels: low, medium, high.
Use the string unknown when the policy number or incident date is not present.
Keep the summary to one sentence. List every important missing detail in missing_information.

Claim record:
{json.dumps(claim, indent=2)}"""
    response = client.messages.parse(model=MODEL_ID, max_tokens=1024, messages=[{"role": "user", "content": prompt}], output_format=ClaimTriage)
    triage = response.parsed_output
    if triage is None:
        raise RuntimeError("Claude returned no parsed output")
    data = triage.model_dump()
    policy_number = data["policy_number"].strip() or "unknown"
    incident_date = data["incident_date"].strip() or "unknown"
    claim_type = data["claim_type"].strip().lower()
    urgency = data["urgency"].strip().lower()
    missing_information = [item.strip() for item in data["missing_information"] if item.strip()]
    if claim_type not in ALLOWED_CLAIM_TYPES:
        claim_type = "other"
        missing_information.append("claim type requires confirmation")
    if urgency not in ALLOWED_URGENCY_LEVELS:
        urgency = "medium"
        missing_information.append("urgency requires confirmation")
    needs_human_review = bool(claim["data_quality_issues"] or missing_information or policy_number.lower() == "unknown" or incident_date.lower() == "unknown")
    return {**claim, "policy_number": policy_number, "incident_date": incident_date, "claim_type": claim_type, "urgency": urgency, "summary": data["summary"].strip(), "missing_information": missing_information, "needs_human_review": needs_human_review}

def save_processed_claims(database_url, processed_claims, run_id, source_count):
    review_count = sum(1 for claim in processed_claims if claim["needs_human_review"])
    with psycopg.connect(database_url) as conn:
        with conn.cursor() as cur:
            cur.execute("DELETE FROM processed_claims;")
            cur.executemany("""INSERT INTO processed_claims (external_id, sender_email, subject, received_at_raw, source_copy_count, data_quality_issues, policy_number, incident_date, claim_type, urgency, summary, missing_information, needs_human_review, run_id) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s);""", [(claim["external_id"], claim["sender_email"], claim["subject"], claim["received_at_raw"], claim["source_copy_count"], json.dumps(claim["data_quality_issues"]), claim["policy_number"], claim["incident_date"], claim["claim_type"], claim["urgency"], claim["summary"], json.dumps(claim["missing_information"]), claim["needs_human_review"], run_id) for claim in processed_claims])
            cur.execute("""INSERT INTO pipeline_runs (run_id, source_count, clean_count, review_count, status) VALUES (%s, %s, %s, %s, %s);""", (run_id, source_count, len(processed_claims), review_count, "successful"))
    return review_count

def print_clean_preview(source_claims, clean_claims):
    print(f"Clean preview: {len(source_claims)} source rows -> {len(clean_claims)} unique claims")
    for claim in clean_claims:
        print(json.dumps({"external_id": claim["external_id"], "source_copy_count": claim["source_copy_count"], "data_quality_issues": claim["data_quality_issues"]}))

def run_pipeline(clean_only=False):
    database_url = require_environment_variable("DATABASE_URL")
    source_url = os.getenv("SOURCE_API_URL", DEFAULT_SOURCE_API_URL)
    run_id = str(uuid.uuid4())
    source_claims = fetch_source_claims(source_url)
    clean_claims = stage_and_clean_claims(database_url, source_claims, run_id)
    if clean_only:
        print_clean_preview(source_claims, clean_claims)
        return
    require_environment_variable("ANTHROPIC_API_KEY")
    client = Anthropic()
    processed_claims = [triage_claim(client, claim) for claim in clean_claims]
    review_count = save_processed_claims(database_url, processed_claims, run_id, len(source_claims))
    print(f"Pipeline complete: {len(source_claims)} source rows -> {len(processed_claims)} clean claims; {review_count} need human review.")

if __name__ == "__main__":
    run_pipeline(clean_only="--clean-only" in sys.argv[1:])

The function order keeps each dependency defined before the pipeline invokes it.

Before you run the full pipeline, which records do you expect to enter human review? Make your prediction now.

• Run the completed pipeline in the activated second Terminal:

python pipeline.py

What does this run prove?

The pipeline fetches the six source records. It cleans them into five unique claims.

Claude extracts typed fields from each claim. Neon stores the enriched results plus the completed run metrics.

You'll see a completion line showing six source rows plus five clean claims plus a human-review count.

Did the full pipeline stop?

• Confirm that the local source API from earlier is still running.
• Confirm that .env contains your saved database connection string.
• Confirm that .env contains your saved Claude API key.

Still stuck? Help me diagnose why my full ClaimFlow pipeline failed.

• Return to the interactive FastAPI documentation tab from earlier.
• Run GET /claims.

You'll see a ready response containing five enriched claims. Each claim includes typed extraction fields plus SQL quality evidence plus its human-review status.

• Run GET /runs.

You'll see a successful run record with a source count of 6 plus a clean count of 5 plus the recorded review count.

That completed run closes the loop from unstructured email text to a reviewable claim record. Claude supplies typed extraction while Python preserves human oversight.

Your local triage pipeline now extracts structured fields without handing high-impact decisions to the model. Next up, you'll deploy the API plus schedule the pipeline in the cloud.

Deploy and Schedule the Pipeline

Your completed triage run proves the workflow locally. A customer-facing integration needs an API that stays reachable after your Mac sleeps.

A GitHub repository becomes the source for a Render deployment. A Cron Job runs the same pipeline every night.

In this step, get ready to:

• Publish the project documentation in a public GitHub repository.
• Deploy the ClaimFlow API as a Render Web Service.
• Run the pipeline nightly through a Render Cron Job.

Publish ClaimFlow to GitHub

A public repository gives Render a stable source to deploy. Your README.md also explains what ClaimFlow does before anyone opens the code.

Your existing README introduces the project. The remaining sections turn it into a reproducible demo with architecture details and safety boundaries.

• In README.md, add the live demo links and architecture summary below the existing introduction by pasting this block:

## Live demo

- API documentation: `https://YOUR-SERVICE.onrender.com/docs`
- Raw source records: `https://YOUR-SERVICE.onrender.com/source/claims`
- Processed claims: `https://YOUR-SERVICE.onrender.com/claims`
- Pipeline runs: `https://YOUR-SERVICE.onrender.com/runs`

## Architecture

```mermaid
flowchart LR
    A[Render Web Service] -->|Synthetic claims JSON| B[Render Cron Job]
    B -->|Raw rows| C[(Neon Postgres)]
    C -->|Deduplicated rows| B
    B -->|Structured extraction| D[Claude API]
    B -->|Processed claims and run metrics| C
    C -->|Recruiter-facing JSON| A
```

## Skills demonstrated

- Python API integration and JSON ingestion
- PostgreSQL parameterized inserts and window-function deduplication
- Claude structured outputs validated by Pydantic
- Deterministic human-review rules
- GitHub delivery, Render deployment, and nightly scheduling

What Does This Section Show?

• The live demo links give readers direct paths to the documentation and pipeline results.
• The architecture diagram shows how the web service and nightly job exchange data through Neon Postgres.
• The skills list connects each project component to a practical engineering capability.

• Save README.md.
• Render README.md with your editor's Markdown preview.

You'll see four demo links followed by a left-to-right architecture diagram. The five project skills appear below the diagram.

Diagram Missing From the Preview?

• Check that the opening ```mermaid fence sits directly above flowchart LR.
• Check that the closing ``` fence sits below the final architecture connection.

Still stuck? Help me fix the Mermaid diagram in my ClaimFlow README.

• Add the local setup and deployment configuration below the skills list by pasting this block:

## Local setup

```bash
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
cp .env.example .env
uvicorn main:app --reload
```

Add a Neon pooled connection string and Claude API key to `.env`. In a second terminal, activate the environment and run `python pipeline.py`. Open `http://127.0.0.1:8000/docs` to explore the result.

## Required environment variables

- `DATABASE_URL`: Neon pooled Postgres connection string
- `ANTHROPIC_API_KEY`: Claude Console API key
- `SOURCE_API_URL`: Full URL of the `/source/claims` endpoint

## Render configuration

Web Service build: `pip install -r requirements.txt`

Web Service start: `uvicorn main:app --host 0.0.0.0 --port $PORT`

Cron Job build: `pip install -r requirements.txt`

Cron Job command: `python pipeline.py`

Cron Job schedule: `0 3 * * *` at 03:00 UTC

Why Document the Configuration?

The setup section gives another developer a repeatable route from a fresh clone to a running API. The Render section records the exact cloud configuration used by this project.

• Save README.md.
• Refresh your editor's Markdown preview.

You'll see a five-command setup block followed by three required environment variables. The Render configuration lists separate web-service and Cron Job settings.

Setup Commands Displaying as Plain Text?

• Check that ```bash appears immediately above the first setup command.
• Check that the closing ``` appears immediately below the Uvicorn command.

Still stuck? Help me repair the command block in my ClaimFlow README.

• Finish README.md by adding the cost and safety section below the Cron Job schedule:

## Cost and safety

The Render web service and Neon project use free plans. The Render Cron Job has a $1 monthly minimum. Claude usage is metered against prepaid credits. Keep auto-reload off. All records are fictional and use documentation-only email domains.

Why End With Safety?

This section makes the project's operating cost visible. It also confirms that ClaimFlow uses fictional records within a human-reviewed workflow.

• Save README.md.
• Refresh the Markdown preview one final time.

You'll see a Cost and safety heading at the bottom. Its paragraph covers Render pricing and prepaid Claude usage.

Cost Section in the Wrong Place?

• Move the Cost and safety heading below the Cron Job schedule line.
• Keep one blank line between the heading and its paragraph.

Still stuck? Help me check the final section order in my ClaimFlow README.

✔️ Awesome, I've got everything!

Your README now documents the demo links and architecture. It also covers setup and deployment safety.

ⓧ I'd like to double check the full code

• Compare your complete README.md with this reference:

# ClaimFlow AI

ClaimFlow AI pulls synthetic claim emails from an API, cleans them with PostgreSQL, extracts structured fields with Claude, and routes uncertain records to human review. It does not approve coverage, assign liability, detect fraud, or authorize payments.

## Live demo

- API documentation: `https://YOUR-SERVICE.onrender.com/docs`
- Raw source records: `https://YOUR-SERVICE.onrender.com/source/claims`
- Processed claims: `https://YOUR-SERVICE.onrender.com/claims`
- Pipeline runs: `https://YOUR-SERVICE.onrender.com/runs`

## Architecture

```mermaid
flowchart LR
    A[Render Web Service] -->|Synthetic claims JSON| B[Render Cron Job]
    B -->|Raw rows| C[(Neon Postgres)]
    C -->|Deduplicated rows| B
    B -->|Structured extraction| D[Claude API]
    B -->|Processed claims and run metrics| C
    C -->|Recruiter-facing JSON| A
```

## Skills demonstrated

- Python API integration and JSON ingestion
- PostgreSQL parameterized inserts and window-function deduplication
- Claude structured outputs validated by Pydantic
- Deterministic human-review rules
- GitHub delivery, Render deployment, and nightly scheduling

## Local setup

```bash
python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt
cp .env.example .env
uvicorn main:app --reload
```

Add a Neon pooled connection string and Claude API key to `.env`. In a second terminal, activate the environment and run `python pipeline.py`. Open `http://127.0.0.1:8000/docs` to explore the result.

## Required environment variables

- `DATABASE_URL`: Neon pooled Postgres connection string
- `ANTHROPIC_API_KEY`: Claude Console API key
- `SOURCE_API_URL`: Full URL of the `/source/claims` endpoint

## Render configuration

Web Service build: `pip install -r requirements.txt`

Web Service start: `uvicorn main:app --host 0.0.0.0 --port $PORT`

Cron Job build: `pip install -r requirements.txt`

Cron Job command: `python pipeline.py`

Cron Job schedule: `0 3 * * *` at 03:00 UTC

## Cost and safety

The Render web service and Neon project use free plans. The Render Cron Job has a $1 monthly minimum. Claude usage is metered against prepaid credits. Keep auto-reload off. All records are fictional and use documentation-only email domains.

Git captures the finished local project before the cloud deployment begins. Your ignored .env file keeps its credentials outside the commit.

• Return to the ClaimFlow Terminal from earlier.
• Stage and commit the project by running these commands:

git add .
git commit -m "Build ClaimFlow AI pipeline"

What Do These Commands Save?

• The first command stages the project files that Git is allowed to track.
• The second command records that staged snapshot with a descriptive message.

• Confirm Terminal reports a new commit containing the ClaimFlow files.

Commit Missing Your Project Files?

• Check that the Terminal is still inside the claimflow-ai folder.
• Check that you saved README.md before creating the commit.

Still stuck? Help me diagnose why my ClaimFlow Git commit is missing files.

• Return to GitHub from earlier.
• Use the new-repository form to create a repository named claimflow-ai.
• Choose public visibility for the repository.
• Leave automatic starter files disabled so the repository stays empty.
• Finish creating the repository.
• Copy the generated command that adds this GitHub repository as the remote for an existing project.
• Run the copied command in the ClaimFlow Terminal.

GitHub may ask you to complete authentication before accepting the first push. Follow that prompt without pasting credentials into your project files.

• Push the committed project to the public repository by running this command:

git push -u origin main

What Does This Push Do?

The push uploads your local main branch to GitHub. The tracking link lets later pushes use the same remote branch.

• Complete the GitHub authentication flow if your browser requests it.
• Open the public claimflow-ai repository in your browser.

You'll see main.py and pipeline.py in the repository. GitHub renders the completed README below the file list.

Push Rejected by GitHub?

• Check that the copied remote command came from the empty claimflow-ai repository.
• Complete any pending GitHub authentication prompt in your browser.
• Confirm that the repository was created without starter files.

Still stuck? Help me diagnose why my ClaimFlow repository push was rejected.

Deploy the live API

A FastAPI Web Service keeps the documentation and claim endpoints available from a public URL. Render builds that service directly from your GitHub repository.

• Return to the Render Dashboard from earlier.
• Start creating a new Web Service.
• Select the public claimflow-ai repository as the deployment source.
• Set Language to Python 3.
• Set Build Command to pip install -r requirements.txt.
• Set Start Command to uvicorn main:app --host 0.0.0.0 --port $PORT.

How Does Render Start ClaimFlow?

The build command installs the pinned packages from requirements.txt. The start command launches the API on the port supplied by Render.

• Select the Free instance type.
• Add DATABASE_URL using the Neon pooled connection string stored in your local .env file.
• Finish creating the Web Service.

The first deployment can take a few minutes while Render installs every dependency. A quiet build log during installation does not mean the deployment has stopped.

• Wait for the deployment log to confirm that the service is live.
• Record the live service URL here: https://your-service.onrender.com.

Before you load the deployment, which ClaimFlow routes do you expect the interactive documentation to list?

• Open https://your-service.onrender.com/docs in your browser.

You'll see the interactive documentation with the root and source routes. The processed claims and pipeline runs routes appear on the same page.

• Open https://your-service.onrender.com/source/claims in a new browser tab.

You'll see six source records. Two records use CLM-1002 as their external claim ID.

• Open https://your-service.onrender.com/claims in a new browser tab.
• Open https://your-service.onrender.com/runs in a new browser tab.

The claims response has a ready status with five enriched claims. The runs response includes the successful local pipeline record stored in Neon.

Live API Not Starting?

• Compare the Build Command with pip install -r requirements.txt.
• Compare the Start Command with uvicorn main:app --host 0.0.0.0 --port $PORT.
• Check that DATABASE_URL contains the Neon pooled connection string.

Still stuck? Help me diagnose my ClaimFlow Render Web Service deployment.

Schedule the nightly pipeline

The Web Service exposes ClaimFlow's data. The Cron Job turns the pipeline into a nightly process with its own run history.

Why Use a Render Cron Job?

GitHub Actions could schedule a repository workflow. A Render Cron Job keeps scheduled execution and run logs beside the deployed API.

Render also prevents overlapping runs of the same Cron Job. That protects the replace-and-save pipeline from running twice at once.

Before You Schedule It

Creating this Cron Job starts a $1 monthly minimum. Its pipeline calls also use your prepaid Claude credits.

• Return to the Render Dashboard.
• Start creating a new Cron Job.
• Select the public claimflow-ai repository as the source.
• Set Build Command to pip install -r requirements.txt.
• Set Command to python pipeline.py.
• Set Schedule to 0 3 * * *.

How Does the Schedule Work?

Render evaluates the five-field schedule in UTC. The value 0 3 * * * starts one run each day at 03:00 UTC.

Environment variables keep deployment settings outside the repository. The Cron Job needs both service credentials and the public source endpoint.

• Add DATABASE_URL using the Neon pooled connection string stored securely from earlier.
• Add ANTHROPIC_API_KEY using the Claude API key stored securely from earlier.
• Set SOURCE_API_URL to https://your-service.onrender.com/source/claims.
• Finish creating the Cron Job.

The first manual run may take a few minutes while Render prepares the environment and Claude processes five claims. A short wait here is expected.

Before you trigger the job, do you expect its source count and clean count to match the local run?

• Click Trigger Run on the Cron Job page.
• Open Runs to follow the manual execution.

You'll see Pipeline complete: 6 source rows -> 5 clean claims in the run logs. That line proves the cloud job fetched and deduplicated the deployed source records.

• Refresh https://your-service.onrender.com/claims in your browser.
• Refresh https://your-service.onrender.com/runs in your browser.

You'll see five enriched claims on the live API. The newest run records six source rows and five clean claims.

• Confirm that the Cron Job's Schedule field still shows 0 3 * * *.

Render now schedules future pipeline runs for 03:00 UTC. The Runs history provides an observable record of each execution.

Manual Run Did Not Complete?

• Check that SOURCE_API_URL ends with /source/claims.
• Check that DATABASE_URL contains the pooled Neon connection string.
• Check that ANTHROPIC_API_KEY contains the active Claude key.

Still stuck? Help me diagnose my failed ClaimFlow Cron Job run.

That's the cloud handoff complete. ClaimFlow now serves live results with a nightly schedule set for 03:00 UTC.

Secret mission

Add a Human Review Queue

Your processed claims currently share one combined feed. Build a dedicated review queue that filters for manual attention and places urgent claims first.

Clean Up Your Resources

Clean Up Your Resources

Choose whether to keep ClaimFlow running, pause it for later, or delete it. The Render Web Service and Neon project use free plans.

Cost warning

The Render Cron Job has a $1 monthly minimum. Claude API calls consume prepaid credits.

You control both costs from their service settings. Choose Pause or Delete if you do not want the nightly pipeline to keep spending.

Resources you used:

• Your public GitHub repository containing the ClaimFlow code and README.
• Your Render Web Service hosting the live API.
• Your Render Cron Job scheduled for 03:00 UTC.
• Your Neon project holding staged claims, processed claims, and run history.
• Your Claude Console personal API key.
• Your local .env file containing the Neon connection string and Claude key.

Keep everything running

No cleanup is required. Choose this option if you want the live API and nightly pipeline to remain available.

• Review the Cron Job's Runs page after its 03:00 UTC schedule.
• Monitor your prepaid credit balance from the Claude Console Billing page.
• Keep auto-reload off to prevent automatic credit purchases.

Your public endpoints remain available for future demonstrations. The pipeline continues processing its synthetic claims each night.

Pause - I'll come back to this later

Remove the scheduled job to stop future nightly runs and its recurring minimum. Keep the free web service, Neon project, GitHub repository, and local files ready for your return.

Pause the scheduled pipeline:

• Open the ClaimFlow Cron Job from the Render Dashboard.
• Use the service deletion control to remove the Cron Job.
• Open Settings in Claude Console.
• Select API keys.
• Open the three-dot menu for the personal API key you created for ClaimFlow.
• Select Disable.

You should no longer see the Cron Job among your active Render services. The disabled Claude key cannot authenticate new pipeline calls.

Your live API and stored data remain available. You can create a new Cron Job and enable the key when you return.

Delete - I don't want to use this again

Deleting the whole stack is a big step, so choose this path only when you no longer need the live demo or its data.

Delete the Render services:

• Open the ClaimFlow Web Service from the Render Dashboard.
• Open the Web Service's settings page.
• Use the service deletion control on that page to remove the Web Service.
• Return to the Render Dashboard.
• Open the ClaimFlow Cron Job.
• Open the Cron Job's settings page.
• Use the service deletion control on that page to remove the Cron Job.

You should no longer see either service in the Render Dashboard. The scheduled 03:00 UTC runs now stop.

Delete the Neon project:

• Select the ClaimFlow project in the Neon Console.
• Select Settings.
• Select Delete.
• Complete the confirmation shown by Neon.

The project leaves your active project list. Neon provides a seven-day recovery window through its API or CLI.

Delete the Claude API key:

• Open Settings in Claude Console.
• Select API keys.
• Open the three-dot menu for the personal API key you created for ClaimFlow.
• Select Delete API Key.
• Complete the permanent deletion confirmation.

The deleted key can no longer authenticate Claude API calls.

Delete the GitHub repository:

• Open the settings page for your public ClaimFlow repository on GitHub.
• Use the permanent repository deletion control at the bottom of that page.
• Complete the confirmation shown by GitHub.

The repository and its public README should no longer load on GitHub.

Remove the local secrets file:

• Return to the local ClaimFlow project in your code editor.
• Select .env in the file explorer.
• Move .env to Trash.
• Confirm that .env no longer appears beside .env.example.

Your local secrets are removed. The reusable example file remains available without containing live credentials.

Nice Work!

Nice Work!

You did it! ClaimFlow AI now runs as a live nightly claims-triage pipeline.

Your service turns six messy synthetic records into five structured claims. Uncertain records stay visible for human review.

You've learned how to:

• Built a live FastAPI integration that moves six synthetic JSON records into Neon Postgres. Used PostgreSQL window functions to keep five unique claims while preserving duplicate evidence.
• Used Claude structured outputs validated by Pydantic to extract predictable claim fields. Added deterministic guardrails that keep high-impact insurance decisions with people.
• Deployed ClaimFlow from GitHub as a Render Web Service. Scheduled nightly processing with a Render Cron Job. Exposed run metrics through live endpoints.
• Completed the Secret Mission by adding an urgency-ordered human review queue. The live /review-queue endpoint now returns only claims that need manual attention.

Ready to quiz yourself?