Vision-Powered Quality Control API

Build an AI defect inspection API using GPT-4o vision and FastAPI streaming.

Introduction

30 Second Summary

Every manufactured product you use went through quality inspection. From car parts to circuit boards, someone (or something) checks for scratches, cracks, and defects before it ships.

In this project, you will build an AI-powered quality inspection API that analyzes images of manufactured parts and streams back detailed defect reports in real-time.

What You'll Build

A FastAPI application that accepts images of components, sends them to GPT-4o's vision capabilities for defect analysis, and streams structured inspection reports back via Server-Sent Events.

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

  • A working REST API that accepts manufacturing images and returns AI-generated defect inspection reports.
  • Real-time SSE streaming that delivers analysis results token-by-token as they generate.
  • Built-in observability tracking latency, token usage, and structured logging for every inspection request.
  • Secret Mission: Add a batch inspection endpoint that processes multiple images in parallel.

Are there any prerequisites?

You need an OpenAI API key with billing enabled, and basic Python knowledge. The project walks you through everything else from scratch.

Before We Start

Before We Start...

Set Up the Project

Before building the inspection API, you need a working project environment with the right dependencies installed.

In this step, you will create a Python project using uv, install FastAPI and the OpenAI SDK, verify your API key connects to GPT-4o, and set up the file structure for the rest of the project.

In this step, get ready to:
  • Create the project and install dependencies.
  • Verify your OpenAI API key works with GPT-4o.
  • Create the project file structure.
Create the project and install dependencies

The uv package manager scaffolds a Python project with a pyproject.toml, a starter main.py, and version tracking out of the box.

  • Open Cursor by pressing Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Cursor and press Enter to open it.
  • Open a terminal inside Cursor by pressing Ctrl+` (both macOS and Windows).
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project and move into it by running these commands:
uv init defect-inspector
cd defect-inspector
  • Confirm the project was created by running:
ls

You should see main.py, pyproject.toml, README.md, and .python-version in the output.

What did uv init create?

The uv init command scaffolds a ready-to-go Python project. It creates a pyproject.toml (your project's configuration file), a starter main.py, and a .python-version file that locks your Python version.

  • Install the three packages your project needs by running:
uv add "fastapi[standard]" openai python-dotenv

What does this command do?

This installs three dependencies and adds them to your pyproject.toml:

  • fastapi[standard] installs FastAPI plus uvicorn and other production essentials for running the server.
  • openai is the official Python SDK for calling GPT-4o's vision API.
  • python-dotenv loads environment variables from a .env file so your API key stays out of code.
  • Verify the packages installed correctly by running:
uv run python -c "import fastapi; import openai; print('Ready')"

You should see Ready printed in your terminal.

Not seeing Ready?

If you see a ModuleNotFoundError, the uv add command may not have completed successfully. Run it again and check for error messages about network connectivity or package resolution.

Make sure you are inside the defect-inspector folder when running the command. The uv run command needs the pyproject.toml to know which packages are available.

help me fix my uv add or import error

Now you need to create the .env file that stores your OpenAI API key.

  • In Cursor's Explorer sidebar, click the new file icon at the top of the file list.
  • Name the file .env and press Enter.
  • Paste the following into the file, replacing your-api-key-here with your actual OpenAI API key:
OPENAI_API_KEY=your-api-key-here

Where do I find my OpenAI API key?

Go to platform.openai.com, sign in, and navigate to the API Keys section. If you already have a key saved somewhere, paste it here. If you need to create a new one, click Create new secret key. Copy the key immediately because you will not be able to see it again.

  • Save .env by pressing Cmd+S (macOS) or Ctrl+S (Windows).
Verify OpenAI API access

Before writing any application code, confirm that your API key is valid and that GPT-4o is accessible from your environment.

  • Test your API connection by running this command in the terminal:
uv run python -c "from openai import OpenAI; c = OpenAI(); print(c.models.retrieve('gpt-4o').id)"

You should see gpt-4o printed in your terminal.

What does this command do?

This creates an OpenAI client (which automatically reads OPENAI_API_KEY from your .env file), then asks the API for the gpt-4o model details. If it prints the model ID, your key is working and the model is accessible.

Seeing an authentication error?

If you see an error mentioning an invalid API key, double-check that your .env file has no extra spaces around the = sign and that the key is pasted correctly with no trailing whitespace.

Make sure your OpenAI account has billing enabled. A key without billing access cannot call gpt-4o.

help me fix my OpenAI API key authentication error

Create the project file structure

The API will be split across three files: models.py for data schemas, inspector.py for the vision analysis engine, and main.py for the FastAPI application.

  • Create the two new files by running:
touch models.py inspector.py
  • Confirm the files exist by running:
ls

You should see main.py, models.py, inspector.py, .env, and pyproject.toml in the output.

The main.py file currently has starter code from uv init. You need to clear it so it is ready for your FastAPI code.

  • In Cursor's Explorer sidebar, click main.py to open it.
  • Select all the existing content with Cmd+A (macOS) or Ctrl+A (Windows).
  • Press Delete to remove everything.
  • Save the now-empty file with Cmd+S (macOS) or Ctrl+S (Windows).

Why clear main.py?

The uv init command creates a placeholder main.py with a basic hello-world function. You will replace this with the full FastAPI application in the next step.

Don't see models.py or inspector.py?

Make sure you ran the touch command while inside the defect-inspector folder. Run pwd to check your current directory. It should end with defect-inspector.

help me find my missing project files

✔️ Awesome, I've got everything!

Great. Double check you have saved all files before moving on.

ⓧ I'd like to double check the full code

At this point, models.py, inspector.py, and main.py should all be empty files. Your .env and pyproject.toml should look like this:

OPENAI_API_KEY=your-api-key-here
[project]
name = "defect-inspector"
version = "0.1.0"
description = "AI-powered industrial defect inspection API with streaming responses"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
    "fastapi[standard]>=0.138.0",
    "openai>=2.44.0",
    "python-dotenv>=1.0.0",
]

Version numbers may differ slightly

Your pyproject.toml is generated by uv add. The version numbers after >= reflect whatever was latest when you ran the command. Slightly higher numbers are fine.

Your project environment is fully configured. Next up, you will define the data models and build the vision analysis engine that powers the inspection API.

Define Data Models and Build the Vision Analysis Engine

Your project is scaffolded and your API key is verified. Now it's time to define the data structures that power your inspection pipeline.

Before any image can be analyzed, the API needs a strict contract for what a defect report looks like. You'll create Pydantic models that define the shape of every inspection response, then build the async function that calls GPT-4o's vision capabilities and streams back results token-by-token.

In this step, get ready to:
  • Define structured data models for defect severity, individual defects, and the full inspection report.
  • Write a system prompt that turns GPT-4o into a quality control inspector.
  • Build an async streaming function that sends images to GPT-4o and yields real-time events.
Define Pydantic models for the inspection report

Every response from the inspection API follows a strict schema. Defining that schema upfront means the AI's output is validated automatically and your API consumers always get a predictable response.

  • Open models.py in Cursor. This file is currently empty.
  • Add the imports and the first two models by pasting this code:
from pydantic import BaseModel, Field
from enum import Enum


# Severity levels for classifying defect impact
class SeverityLevel(str, Enum):
    CRITICAL = "critical"
    MAJOR = "major"
    MINOR = "minor"
    NONE = "none"


# Represents a single detected defect with its metadata
class Defect(BaseModel):
    defect_type: str = Field(description="Type of defect detected")
    severity: SeverityLevel = Field(description="Severity classification")
    location: str = Field(description="Location of defect on the part")
    description: str = Field(description="Detailed description of the defect")
    confidence: float = Field(ge=0.0, le=1.0, description="Confidence score")

What does this code do?

SeverityLevel is a string enum with four levels: critical (safety risk), major (quality issue), minor (cosmetic), and none.

Defect defines a single detected defect with its type, severity, location on the part, a description, and a confidence score between 0 and 1. The Field constraints like ge=0.0, le=1.0 enforce valid ranges automatically.

  • Add the remaining two models below the Defect class:

# The full inspection report returned by the API
class InspectionReport(BaseModel):
    overall_status: str = Field(description="PASS, FAIL, or NEEDS_REVIEW")
    defects: list[Defect] = Field(default_factory=list)
    summary: str = Field(description="Brief summary of inspection findings")
    recommendations: list[str] = Field(default_factory=list)


# Tracks performance and cost data for each inspection
class InspectionMetrics(BaseModel):
    part_id: str
    latency_ms: float
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    model: str

What does this code do?

InspectionReport is the main response shape. It contains an overall pass/fail status, a list of Defect objects, a human-readable summary, and action recommendations.

InspectionMetrics tracks observability data for each request: which part was inspected, how long it took, and how many tokens were consumed. You'll use this later to calculate cost per inspection.

  • Save models.py.
  • Verify the models load correctly by running this in your terminal:
uv run python -c "from models import InspectionReport, Defect, SeverityLevel; print('Models loaded')"

You should see Models loaded printed with no errors.

Seeing an ImportError?

Make sure you saved models.py before running the command. Check that the file is at the top level of your defect-inspector folder, not inside a subfolder.

help me debug this ImportError in models.py

✔️ Awesome, I've got everything!

Great. Double check you've saved models.py before moving on.

ⓧ I'd like to double check the full code

from pydantic import BaseModel, Field
from enum import Enum


class SeverityLevel(str, Enum):
    CRITICAL = "critical"
    MAJOR = "major"
    MINOR = "minor"
    NONE = "none"


class Defect(BaseModel):
    defect_type: str = Field(description="Type of defect detected")
    severity: SeverityLevel = Field(description="Severity classification")
    location: str = Field(description="Location of defect on the part")
    description: str = Field(description="Detailed description of the defect")
    confidence: float = Field(ge=0.0, le=1.0, description="Confidence score")


class InspectionReport(BaseModel):
    overall_status: str = Field(description="PASS, FAIL, or NEEDS_REVIEW")
    defects: list[Defect] = Field(default_factory=list)
    summary: str = Field(description="Brief summary of inspection findings")
    recommendations: list[str] = Field(default_factory=list)


class InspectionMetrics(BaseModel):
    part_id: str
    latency_ms: float
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    model: str
Build the OpenAI vision analysis function

The analysis engine lives in inspector.py. It sends an image to GPT-4o, streams back the response token-by-token, then validates the final output against your Pydantic models.

  • Open inspector.py in Cursor. This file is currently empty.
  • Add the imports and system prompt by pasting this code:
import json
import time
from collections.abc import AsyncGenerator

from openai import AsyncOpenAI

from models import InspectionReport, InspectionMetrics

SYSTEM_PROMPT = """You are an expert industrial quality control inspector. Analyze the provided image of a manufactured component for defects.

You must respond with a JSON object in this exact format:
{
    "overall_status": "PASS" | "FAIL" | "NEEDS_REVIEW",
    "defects": [
        {
            "defect_type": "scratch | crack | corrosion | misalignment | discoloration | deformation | contamination | other",
            "severity": "critical | major | minor | none",
            "location": "description of where on the part",
            "description": "detailed description of the defect",
            "confidence": 0.0 to 1.0
        }
    ],
    "summary": "Brief overall assessment",
    "recommendations": ["List of recommended actions"]
}

Guidelines:
- Be thorough but precise in your analysis
- Rate severity: critical = immediate safety/function risk, major = significant quality issue, minor = cosmetic or negligible impact
- If no defects are found, return overall_status as "PASS" with an empty defects array
- Confidence reflects how certain you are about each specific defect
- Include at least one recommendation even for passing parts"""

What does this code do?

The imports bring in AsyncOpenAI for non-blocking API calls and your Pydantic models for response validation.

SYSTEM_PROMPT is the instruction set for GPT-4o. It defines the exact JSON format the model must return, the defect taxonomy (scratch, crack, corrosion, etc.), severity classification guidelines, and edge-case handling rules. By embedding the schema directly in the prompt alongside response_format={"type": "json_object"}, you guarantee valid JSON output every time.

  • Below the SYSTEM_PROMPT, add the async generator function that performs the actual analysis:

async def analyze_defects_stream(
    client: AsyncOpenAI,
    image_base64: str,
    part_id: str,
    part_type: str,
) -> AsyncGenerator[dict, None]:
    """Stream defect analysis results as SSE-compatible events."""
    start_time = time.time()

    # Signal that inspection is starting
    yield {
        "event": "inspection_start",
        "data": {"part_id": part_id, "part_type": part_type, "model": "gpt-4o"},
    }

    collected_content = ""
    prompt_tokens = 0
    completion_tokens = 0

    # Send image to GPT-4o with streaming enabled
    stream = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": (
                            f"Inspect this {part_type} component (Part ID: {part_id}). "
                            "Identify any defects, assess severity, and provide your inspection report."
                        ),
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{image_base64}"
                        },
                    },
                ],
            },
        ],
        response_format={"type": "json_object"},
        stream=True,
        stream_options={"include_usage": True},
    )

    # Yield each token as it arrives from the model
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            token_text = chunk.choices[0].delta.content
            collected_content += token_text
            yield {"event": "token", "data": {"content": token_text}}

        if chunk.usage:
            prompt_tokens = chunk.usage.prompt_tokens
            completion_tokens = chunk.usage.completion_tokens

    elapsed_ms = (time.time() - start_time) * 1000

    # Validate the collected JSON against the Pydantic model
    try:
        report = InspectionReport.model_validate_json(collected_content)
        yield {
            "event": "inspection_complete",
            "data": report.model_dump(),
        }
    except Exception as e:
        yield {
            "event": "error",
            "data": {"message": f"Failed to parse report: {str(e)}"},
        }

    # Emit performance metrics as the final event
    metrics = InspectionMetrics(
        part_id=part_id,
        latency_ms=round(elapsed_ms, 2),
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=prompt_tokens + completion_tokens,
        model="gpt-4o",
    )
    yield {"event": "metrics", "data": metrics.model_dump()}

What does this function do?

analyze_defects_stream is an async generator. Instead of returning one big result, it yields multiple events over time. Each yield becomes a Server-Sent Event that the API streams to the client.

The event flow is: inspection_start (signals the request began), then many token events (one per chunk from GPT-4o), then inspection_complete (the validated report), and finally metrics (latency and token usage).

The stream_options={"include_usage": True} parameter tells OpenAI to include token counts in the final streamed chunk, which feeds your observability metrics.

  • Save inspector.py.
  • Verify the file loads without syntax errors by running:
uv run python -c "from inspector import analyze_defects_stream; print('Inspector loaded')"

You should see Inspector loaded printed. You won't be able to call the function yet because it needs an active AsyncOpenAI client and an image. That comes in the next step when you build the FastAPI endpoints.

Seeing a SyntaxError?

Triple-quoted strings (the SYSTEM_PROMPT) are sensitive to indentation. Make sure the closing """ is on its own line with no leading spaces. Also check that you haven't accidentally indented the function definition inside the prompt string.

help me debug this SyntaxError in inspector.py

✔️ Awesome, I've got everything!

Great. Double check you've saved inspector.py before moving on.

ⓧ I'd like to double check the full code

import json
import time
from collections.abc import AsyncGenerator

from openai import AsyncOpenAI

from models import InspectionReport, InspectionMetrics

SYSTEM_PROMPT = """You are an expert industrial quality control inspector. Analyze the provided image of a manufactured component for defects.

You must respond with a JSON object in this exact format:
{
    "overall_status": "PASS" | "FAIL" | "NEEDS_REVIEW",
    "defects": [
        {
            "defect_type": "scratch | crack | corrosion | misalignment | discoloration | deformation | contamination | other",
            "severity": "critical | major | minor | none",
            "location": "description of where on the part",
            "description": "detailed description of the defect",
            "confidence": 0.0 to 1.0
        }
    ],
    "summary": "Brief overall assessment",
    "recommendations": ["List of recommended actions"]
}

Guidelines:
- Be thorough but precise in your analysis
- Rate severity: critical = immediate safety/function risk, major = significant quality issue, minor = cosmetic or negligible impact
- If no defects are found, return overall_status as "PASS" with an empty defects array
- Confidence reflects how certain you are about each specific defect
- Include at least one recommendation even for passing parts"""


async def analyze_defects_stream(
    client: AsyncOpenAI,
    image_base64: str,
    part_id: str,
    part_type: str,
) -> AsyncGenerator[dict, None]:
    """Stream defect analysis results as SSE-compatible events."""
    start_time = time.time()

    yield {
        "event": "inspection_start",
        "data": {"part_id": part_id, "part_type": part_type, "model": "gpt-4o"},
    }

    collected_content = ""
    prompt_tokens = 0
    completion_tokens = 0

    stream = await client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {
                "role": "user",
                "content": [
                    {
                        "type": "text",
                        "text": (
                            f"Inspect this {part_type} component (Part ID: {part_id}). "
                            "Identify any defects, assess severity, and provide your inspection report."
                        ),
                    },
                    {
                        "type": "image_url",
                        "image_url": {
                            "url": f"data:image/jpeg;base64,{image_base64}"
                        },
                    },
                ],
            },
        ],
        response_format={"type": "json_object"},
        stream=True,
        stream_options={"include_usage": True},
    )

    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            token_text = chunk.choices[0].delta.content
            collected_content += token_text
            yield {"event": "token", "data": {"content": token_text}}

        if chunk.usage:
            prompt_tokens = chunk.usage.prompt_tokens
            completion_tokens = chunk.usage.completion_tokens

    elapsed_ms = (time.time() - start_time) * 1000

    try:
        report = InspectionReport.model_validate_json(collected_content)
        yield {
            "event": "inspection_complete",
            "data": report.model_dump(),
        }
    except Exception as e:
        yield {
            "event": "error",
            "data": {"message": f"Failed to parse report: {str(e)}"},
        }

    metrics = InspectionMetrics(
        part_id=part_id,
        latency_ms=round(elapsed_ms, 2),
        prompt_tokens=prompt_tokens,
        completion_tokens=completion_tokens,
        total_tokens=prompt_tokens + completion_tokens,
        model="gpt-4o",
    )
    yield {"event": "metrics", "data": metrics.model_dump()}
Review the system prompt design

The quality of your inspection results depends entirely on how well the system prompt guides GPT-4o. Your prompt does three critical things that make structured output reliable.

Why does the prompt include the JSON schema?

Even though response_format={"type": "json_object"} guarantees valid JSON, it doesn't guarantee the right JSON structure. The prompt's embedded schema tells GPT-4o exactly which fields to include, what values are valid for each (like the defect type taxonomy), and how to format nested objects.

This creates a double validation: the model tries to match the schema from the prompt, and then Pydantic's model_validate_json catches anything the model gets wrong. If the AI hallucinates an extra field or uses an invalid severity level, Pydantic raises an error that your function handles gracefully with an error event.

What is JSON mode?

JSON mode (response_format={"type": "json_object"}) constrains GPT-4o to only output valid JSON. Without it, the model might wrap its response in markdown code fences or add conversational text around the JSON, which would break model_validate_json.

Your data models and analysis engine are ready. Next up, you'll wire these into FastAPI endpoints that accept image uploads and stream back the inspection results in real-time.

Create the FastAPI Streaming API

You have the data models and the vision analysis engine ready. Now you need an API layer that accepts image uploads and delivers inspection results back to the client.

In this step, you will build the FastAPI application that ties everything together. It initializes the OpenAI client on startup, exposes a streaming endpoint that pushes Server-Sent Events as the analysis progresses, and provides a synchronous endpoint for clients that prefer a single JSON response.

In this step, get ready to:
  • Set up the FastAPI application with lifespan management and a health check endpoint.
  • Build a streaming inspection endpoint that returns SSE events in real-time.
  • Build a synchronous inspection endpoint that returns the full report as JSON.
Set up the FastAPI application with lifespan management

The API server needs to create the AsyncOpenAI client once at startup and share it across all requests. FastAPI's lifespan pattern handles this by running initialization code before the first request arrives and cleanup code when the server shuts down.

  • In Cursor, open main.py (the file uv init created earlier). Replace any existing content with these imports and configuration:
import json
import logging
import base64
from collections.abc import AsyncIterable
from contextlib import asynccontextmanager

from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
from dotenv import load_dotenv

from models import InspectionReport
from inspector import analyze_defects_stream

# Load environment variables from .env
load_dotenv()

# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("defect-inspector")

# Global OpenAI client, initialized during app startup
openai_client: AsyncOpenAI | None = None

What does this code do?

  • load_dotenv() reads your .env file so the OPENAI_API_KEY environment variable is available to the OpenAI SDK.
  • logging.basicConfig(...) sets up structured log output with timestamps, which makes it easy to see when each inspection request completes.
  • openai_client is declared as a global. The lifespan function (coming next) will assign it a real client instance at startup.
  • Below the imports, add the lifespan context manager, the FastAPI app instance, and a health check endpoint:
@asynccontextmanager
async def lifespan(app: FastAPI):
    global openai_client
    openai_client = AsyncOpenAI()
    logger.info("OpenAI client initialized")
    yield
    logger.info("Shutting down")


app = FastAPI(
    title="AI Industrial Defect Inspector",
    description="Vision-powered manufacturing quality inspection API with streaming analysis",
    version="0.1.0",
    lifespan=lifespan,
)


@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "defect-inspector"}

What does this code do?

  • The lifespan function is an async context manager. Code before yield runs at startup (creating the client). Code after yield runs at shutdown.
  • Passing lifespan=lifespan to FastAPI(...) wires the lifecycle. The OpenAI client is created once and reused across every request.
  • The /health endpoint gives you a quick way to confirm the server is running without triggering any OpenAI calls.
  • Save main.py.
  • Start the dev server by running this command:
uv run fastapi dev main.py
  • Open your browser and visit http://localhost:8000/health. You should see:
{"status": "healthy", "service": "defect-inspector"}

Not seeing the health response?

  • Check that your terminal shows Uvicorn running on http://127.0.0.1:8000 in the output. If it shows an import error, make sure models.py and inspector.py are saved and in the same folder as main.py.
  • If you see a ModuleNotFoundError, confirm you ran the command from inside the defect-inspector directory.
  • help me debug my FastAPI server startup error
Build the streaming inspection endpoint

The streaming endpoint accepts an image upload, base64-encodes it, and streams back SSE events as GPT-4o analyzes the image. Clients receive tokens in real-time rather than waiting for the full response.

  • In main.py, add the streaming endpoint below the health_check function:
@app.post("/inspect/stream")
async def inspect_stream(
    file: UploadFile = File(..., description="Image of the part to inspect"),
    part_id: str = Form(default="UNKNOWN", description="Identifier for the part"),
    part_type: str = Form(default="general", description="Type of part being inspected"),
):
    # Validate the uploaded file is an image
    if not file.content_type or not file.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="Uploaded file must be an image")

    # Read and base64-encode the image for the OpenAI API
    image_bytes = await file.read()
    image_base64 = base64.b64encode(image_bytes).decode("utf-8")

    logger.info(
        f"Inspection request: part_id={part_id}, part_type={part_type}, "
        f"size={len(image_bytes)} bytes"
    )

    # Inner async generator that formats events as SSE strings
    async def event_generator() -> AsyncIterable[str]:
        async for event in analyze_defects_stream(
            client=openai_client,
            image_base64=image_base64,
            part_id=part_id,
            part_type=part_type,
        ):
            sse_data = json.dumps(event["data"])
            yield f"event: {event['event']}\ndata: {sse_data}\n\n"

            if event["event"] == "metrics":
                logger.info(
                    f"Inspection complete: part_id={part_id} "
                    f"latency={event['data']['latency_ms']}ms "
                    f"tokens={event['data']['total_tokens']}"
                )

    # Return the SSE stream with appropriate headers
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )

What does this code do?

  • The endpoint accepts multipart form data: the image file, a part_id, and part_type string.
  • It validates the upload is an image, then encodes the raw bytes as base64 for the OpenAI vision API.
  • event_generator() wraps the analysis stream. Each event from analyze_defects_stream gets formatted into the SSE wire protocol: event: <type>\ndata: <json>\n\n.
  • StreamingResponse with media_type="text/event-stream" tells the client this is an SSE connection. The Cache-Control and X-Accel-Buffering headers prevent proxies from buffering the stream.

Why SSE over WebSockets?

SSE is a better fit here because the communication is one-directional. The server pushes analysis results to the client. There is no need for the client to send messages back during the inspection. SSE works over standard HTTP, requires no protocol upgrade, and is simpler to debug with curl.

  • Save main.py. The dev server auto-reloads.
  • Visit http://localhost:8000/docs in your browser. You should see the POST /inspect/stream endpoint listed alongside /health.

Endpoint not appearing in /docs?

  • Check your terminal for a traceback. A syntax error or indentation issue prevents the server from reloading. Fix the error and save again.
  • Make sure the function is decorated with @app.post("/inspect/stream") and is not accidentally nested inside another function.
  • help me fix my streaming endpoint
Build the synchronous inspection endpoint

Some clients prefer receiving the complete inspection report in a single JSON response rather than processing a stream. This endpoint reuses the same analyze_defects_stream function but collects the full result before responding.

  • In main.py, add the synchronous endpoint below the inspect_stream function:
@app.post("/inspect", response_model=InspectionReport)
async def inspect(
    file: UploadFile = File(..., description="Image of the part to inspect"),
    part_id: str = Form(default="UNKNOWN", description="Identifier for the part"),
    part_type: str = Form(default="general", description="Type of part being inspected"),
):
    # Validate the uploaded file is an image
    if not file.content_type or not file.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="Uploaded file must be an image")

    # Read and base64-encode the image
    image_bytes = await file.read()
    image_base64 = base64.b64encode(image_bytes).decode("utf-8")

    logger.info(
        f"Inspection request (sync): part_id={part_id}, part_type={part_type}"
    )

    # Consume the analysis stream and capture the final report
    report = None
    async for event in analyze_defects_stream(
        client=openai_client,
        image_base64=image_base64,
        part_id=part_id,
        part_type=part_type,
    ):
        if event["event"] == "inspection_complete":
            report = InspectionReport.model_validate(event["data"])
        if event["event"] == "metrics":
            logger.info(
                f"Inspection complete: part_id={part_id} "
                f"latency={event['data']['latency_ms']}ms "
                f"tokens={event['data']['total_tokens']}"
            )

    # Return an error if analysis did not produce a report
    if report is None:
        raise HTTPException(
            status_code=500, detail="Inspection failed to produce a report"
        )

    return report

What does this code do?

  • The response_model=InspectionReport declaration tells FastAPI to validate and serialize the response using your Pydantic model. It also generates accurate schema documentation in /docs.
  • Instead of yielding events, this endpoint iterates through them silently. It captures the inspection_complete event's data as the final report.
  • If the analysis fails to parse (an error event was yielded instead), report remains None and the endpoint returns a 500 error.
  • Save main.py.
  • Refresh http://localhost:8000/docs in your browser. You should now see three endpoints: GET /health, POST /inspect/stream, and POST /inspect.

Seeing a validation error in the terminal?

  • Make sure the from models import InspectionReport line at the top of main.py matches the class name exactly as defined in models.py.
  • Check that response_model=InspectionReport is passed in the @app.post("/inspect", ...) decorator, not inside the function body.
  • help me fix my synchronous endpoint

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and the dev server is running without errors.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point:

import json
import logging
import base64
from collections.abc import AsyncIterable
from contextlib import asynccontextmanager

from fastapi import FastAPI, UploadFile, File, Form, HTTPException
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
from dotenv import load_dotenv

from models import InspectionReport
from inspector import analyze_defects_stream

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("defect-inspector")

openai_client: AsyncOpenAI | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global openai_client
    openai_client = AsyncOpenAI()
    logger.info("OpenAI client initialized")
    yield
    logger.info("Shutting down")


app = FastAPI(
    title="AI Industrial Defect Inspector",
    description="Vision-powered manufacturing quality inspection API with streaming analysis",
    version="0.1.0",
    lifespan=lifespan,
)


@app.get("/health")
async def health_check():
    return {"status": "healthy", "service": "defect-inspector"}


@app.post("/inspect/stream")
async def inspect_stream(
    file: UploadFile = File(..., description="Image of the part to inspect"),
    part_id: str = Form(default="UNKNOWN", description="Identifier for the part"),
    part_type: str = Form(default="general", description="Type of part being inspected"),
):
    if not file.content_type or not file.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="Uploaded file must be an image")

    image_bytes = await file.read()
    image_base64 = base64.b64encode(image_bytes).decode("utf-8")

    logger.info(
        f"Inspection request: part_id={part_id}, part_type={part_type}, "
        f"size={len(image_bytes)} bytes"
    )

    async def event_generator() -> AsyncIterable[str]:
        async for event in analyze_defects_stream(
            client=openai_client,
            image_base64=image_base64,
            part_id=part_id,
            part_type=part_type,
        ):
            sse_data = json.dumps(event["data"])
            yield f"event: {event['event']}\ndata: {sse_data}\n\n"

            if event["event"] == "metrics":
                logger.info(
                    f"Inspection complete: part_id={part_id} "
                    f"latency={event['data']['latency_ms']}ms "
                    f"tokens={event['data']['total_tokens']}"
                )

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",
        },
    )


@app.post("/inspect", response_model=InspectionReport)
async def inspect(
    file: UploadFile = File(..., description="Image of the part to inspect"),
    part_id: str = Form(default="UNKNOWN", description="Identifier for the part"),
    part_type: str = Form(default="general", description="Type of part being inspected"),
):
    if not file.content_type or not file.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="Uploaded file must be an image")

    image_bytes = await file.read()
    image_base64 = base64.b64encode(image_bytes).decode("utf-8")

    logger.info(
        f"Inspection request (sync): part_id={part_id}, part_type={part_type}"
    )

    report = None
    async for event in analyze_defects_stream(
        client=openai_client,
        image_base64=image_base64,
        part_id=part_id,
        part_type=part_type,
    ):
        if event["event"] == "inspection_complete":
            report = InspectionReport.model_validate(event["data"])
        if event["event"] == "metrics":
            logger.info(
                f"Inspection complete: part_id={part_id} "
                f"latency={event['data']['latency_ms']}ms "
                f"tokens={event['data']['total_tokens']}"
            )

    if report is None:
        raise HTTPException(
            status_code=500, detail="Inspection failed to produce a report"
        )

    return report

Your API is live and ready to inspect. Next up, you will send a real image to both endpoints and watch the defect analysis stream back in your terminal.

Test and Add Observability

Your API is fully built with both streaming and synchronous endpoints. Now it is time to see it in action with real images and observe the built-in metrics your code produces.

In a production quality control system, you need to know how long each inspection takes, how many tokens it consumes, and what it costs. Your API already tracks all of this. You just need to send it an image and watch the data flow.

In this step, get ready to:
  • Test the streaming endpoint and watch SSE events arrive in real-time.
  • Test the synchronous endpoint and verify the structured JSON report.
  • Read the observability logs and calculate cost per inspection.
Test the streaming endpoint with a sample image

To test the API, you need any image of a manufactured part. A photo of a scratched surface, a PCB, a weld, or even a close-up of any household object with visible wear will work.

  • Find or take a photo of any manufactured object (a circuit board, a metal surface, a scratched phone case, anything with a visible surface).
  • Save the image as test_image.jpg inside your defect-inspector folder.

No manufactured parts handy?

Search online for "PCB defect" or "scratched metal surface" and save one of the results. The model inspects whatever image you provide, so even a stock photo works for testing.

Your dev server is still running in your first terminal. You need a second terminal to run curl commands against it.

  • Open a new terminal tab or window.
  • Navigate to your defect-inspector folder by running:
cd ~/Desktop/defect-inspector
  • Send your test image to the streaming endpoint by running this command:
curl -X POST http://localhost:8000/inspect/stream -F "file=@test_image.jpg" -F "part_id=PCB-001" -F "part_type=circuit board" -N

What does this command do?

The -F flags send multipart form data: your image file, a part ID, and a part type. The -N flag disables curl's output buffering so you see SSE events as they arrive instead of waiting for the full response.

You should see events streaming in one by one. First an inspection_start event, then many token events as GPT-4o generates the JSON analysis, then an inspection_complete event with the parsed report, and finally a metrics event.

Seeing "Connection refused"?

Make sure your dev server is still running in the other terminal. If you stopped it, restart it with uv run fastapi dev main.py.

If you see a 400 error about the file not being an image, check that test_image.jpg exists in the same folder where you ran the curl command.

Help me debug my curl request to the streaming endpoint.

Test the synchronous endpoint and verify structured output

The synchronous endpoint returns the same analysis as a single JSON response instead of a stream. This is useful for clients that just need the final report without real-time updates.

  • Send the same image to the synchronous endpoint by running:
curl -X POST http://localhost:8000/inspect -F "file=@test_image.jpg" -F "part_id=PCB-001" -F "part_type=circuit board"

This time you get a single JSON object back after the full analysis completes. Verify it contains overall_status, a defects array, summary, and recommendations.

What should I see?

The response is a structured InspectionReport matching your Pydantic model. The overall_status will be PASS, FAIL, or NEEDS_REVIEW. Each defect in the array has a type, severity, location, description, and confidence score between 0 and 1.

Getting a 500 error?

A 500 error usually means the OpenAI API call failed. Check your server terminal for the error details. The most common causes are an invalid or expired API key, or insufficient account balance.

Help me figure out why my /inspect endpoint is returning a 500 error.

Observe the metrics and calculate inspection cost

Every inspection request your API processes gets logged with structured metrics. This is the observability layer that tells you how your service is performing in production.

  • Switch back to the terminal where your dev server is running.
  • Look at the log output. You should see structured lines showing the inspection requests you just made.

Each completed inspection logs a line with latency_ms (how long the full round-trip took) and total_tokens (how many tokens GPT-4o consumed).

How do I calculate cost per inspection?

GPT-4o costs $2.50 per 1M input tokens and $10.00 per 1M output tokens. Use the token counts from your metrics event to calculate your cost per inspection:

Cost = (prompt_tokens * $2.50 + completion_tokens * $10.00) / 1,000,000.

A typical inspection uses around 1,500 input tokens and 300 output tokens, costing roughly $0.007 per call.

  • Find the prompt_tokens and completion_tokens values from either the SSE metrics event output or the server logs.
  • Calculate your cost per inspection using the formula above.

Why track these metrics?

In a production inference service, latency and token usage directly affect user experience and cost. If latency spikes, something is wrong with your prompts or images. If token usage is unexpectedly high, you might be sending images that are too large.

These logs give you the data to spot problems before your users (or your bill) do.

Secret mission

Add Batch Inspection with Parallel Processing

Your single-image endpoint works great, but real factories don't inspect one part at a time. Build a batch endpoint that processes multiple images in parallel using asyncio.gather, then computes aggregated quality statistics across the entire batch.

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing cloud costs. The OpenAI API only charges per request, meaning once you stop sending images for inspection, no further charges accrue.

Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The defect-inspector project folder on your Desktop (contains main.py, models.py, inspector.py, .env, pyproject.toml).
  • The FastAPI dev server running in your terminal.
  • OpenAI API usage (pay-per-request, no ongoing charge when idle).

Keep everything running

No action needed. Choose this if you are still actively building or experimenting with the API.

  • Your project files remain on your Desktop.
  • The FastAPI dev server keeps running in your terminal. You can send more inspection requests anytime.
  • Your OpenAI API key in .env stays active. You are only charged when you make requests.

Pause - I'll come back to this later

Shut down the running server to free up your terminal. Your files stay on disk so you can pick up where you left off.

  • Switch to the terminal where the FastAPI server is running.
  • Press Ctrl+C to stop the dev server.
  • Your project files remain in the defect-inspector folder on your Desktop.
  • To resume later, open a terminal in that folder and run uv run fastapi dev main.py again.

Delete - I don't want to use this again

Remove all project files from your machine. This cannot be undone.

  • Switch to the terminal where the FastAPI server is running and press Ctrl+C to stop it.
  • Delete the project folder by running the following command (macOS/Linux):
cd ~/Desktop
rm -rf defect-inspector

On Windows, run this instead:

cd ~\Desktop
Remove-Item -Recurse -Force defect-inspector
  • Confirm the folder is gone by running ls (macOS/Linux) or dir (Windows) on your Desktop. You should no longer see defect-inspector in the output.
  • Your OpenAI API key remains active in your OpenAI account. If you want to revoke it, visit the API keys page at platform.openai.com and delete the key from there.

Nice Work!

Nice Work!

Nice work! You just built a production-style AI vision inspection API from scratch.

You now have a working service that takes manufacturing images, streams GPT-4o's defect analysis in real-time, and returns structured inspection reports. This is the same pattern companies use for automated quality control in Industry 4.0 pipelines.

You learned how to:

  • Build a FastAPI REST API with Server-Sent Events streaming, delivering GPT-4o's vision analysis token-by-token as it generates.
  • Create structured Pydantic models and use JSON mode to guarantee parseable structured output from an LLM.
  • Add built-in observability with latency tracking, token usage metrics, and structured logging for every inspection request.
  • Secret Mission: Built a batch endpoint that processes multiple images in parallel using asyncio.gather and returns aggregated quality statistics.

Ready to quiz yourself?