Build a Production AI Content API

Build a multi-endpoint AI API with streaming, auth, and validation using FastAPI.

Introduction

30 Second Summary

Every AI product you use daily has a backend API that receives requests, calls an AI model, and streams responses back to you. That backend is real engineering with authentication, validation, and error handling built in.

In this project, you will build a production-quality AI Content API using FastAPI and Google Gemini. You'll create endpoints that summarize text and stream chat responses token-by-token, all protected behind proper authentication.

What You'll Build

When you run it, you get a live interactive API at localhost where you can paste text for an instant AI summary or open a streaming chat connection and watch tokens arrive in real-time.

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

  • A live /summarize endpoint you can call from interactive Swagger UI docs to condense any text into AI-generated key points.
  • A streaming chat endpoint that delivers AI responses token-by-token using Server-Sent Events, the same pattern that powers ChatGPT.
  • Production hardening patterns including API key authentication, input validation, and structured error responses you can demonstrate in interviews.
  • Secret Mission: Rate limiting that detects abuse and rejects excessive requests to protect your API quota.

Are there any prerequisites?

Basic Python knowledge and familiarity with REST APIs is all you need. Django experience is more than sufficient.

You'll install FastAPI, the Gemini SDK, and grab a free API key (no credit card required) in Step 1.

Before We Start

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Environment

You need FastAPI, the Gemini SDK, and an API key before you can build anything.

Getting these running first means every later step gives you instant feedback when you add features to your API.

In this step, get ready to:
  • Create a project directory with a virtual environment and install dependencies.
  • Get your free Gemini API key and create the project files.
  • Start the dev server and confirm Swagger docs load.
Create your project directory and install dependencies

A virtual environment isolates your project's packages from the rest of your system. This prevents version conflicts between different Python projects.

  • Open your terminal. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir ai-content-api
cd ai-content-api
  • Confirm you are in the right place by running:
pwd

You should see a path ending in ai-content-api.

  • Create a virtual environment and activate it by running these commands:

macOS

python3 -m venv venv
source venv/bin/activate

Windows

python -m venv venv
venv\Scripts\activate

Your terminal prompt now shows (venv) at the beginning. This confirms the virtual environment is active.

  • Install the project dependencies by running:
pip install "fastapi[standard]" google-genai python-dotenv

You will see pip downloading and installing packages. This takes about 30 seconds.

  • Verify the packages installed correctly by running:
python -c "import fastapi; import google.genai; print('Ready')"

You should see Ready printed in your terminal.

Not seeing Ready?

Make sure your virtual environment is active. You should see (venv) in your terminal prompt.

If pip install failed, check your internet connection and try the install command again.

Why won't my packages install?

Get your Gemini API key and create project files

Google's Gemini API has a free tier that requires no credit card. You just need to create an API key in Google AI Studio.

  • Open your browser and navigate to aistudio.google.com.
  • Sign in with your Google account if prompted.
  • If you are a new user, accept the Terms of Service. AI Studio automatically creates a default project and API key for you.
  • Navigate to the API Keys page from the dashboard.
  • Click Create API key if you do not already have one listed.
  • Copy your new API key and keep it somewhere safe. You will paste it into a file in a moment.

What is an API key?

An API key is a secret string that identifies your application to Google's servers. It lets Gemini know who is making the request and tracks your usage against the free tier limits.

  • Back in VS Code, open the ai-content-api folder. Click File in the top menu bar.
  • Click Open Folder.
  • Select the ai-content-api folder on your Desktop, then click Open.
  • Create a new file called .env in the root of your project. Click File then New File, and save it as .env.
  • Paste the following into your .env file, replacing your-api-key-here with the Gemini API key you just copied:
GEMINI_API_KEY=your-api-key-here
DEV_API_KEY=dev-secret-key-123

What are these two keys?

GEMINI_API_KEY authenticates your app with Google's AI service. DEV_API_KEY is a key that clients will use to authenticate with YOUR API. You will wire this up in a later step.

  • Save .env.
  • Create another new file called main.py in the same folder. Paste the following code into it:
import os

from dotenv import load_dotenv
from fastapi import FastAPI

from google import genai

load_dotenv()

app = FastAPI(
    title="AI Content API",
    description="Production AI API with streaming, auth, and validation",
    version="1.0.0",
)

client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))


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

What does this code do?

  • load_dotenv() reads your .env file and loads the values as environment variables so your code can access them securely.
  • FastAPI(...) creates your API application with metadata that appears in the auto-generated documentation.
  • genai.Client(...) connects to Google's Gemini service using your API key.
  • The /health endpoint is a simple check to confirm the server is running.
  • Save main.py.
Start the dev server and confirm Swagger docs load

FastAPI includes a built-in development server that auto-reloads when you change code. It also generates interactive API documentation automatically.

  • In your terminal (make sure you are still in the ai-content-api folder with (venv) active), start the dev server by running:
fastapi dev main.py

What does this command do?

This command starts a local development server with auto-reload enabled. It detects the FastAPI app in main.py and serves it using Uvicorn. The terminal stays occupied running the server.

You will see output including Uvicorn running on http://127.0.0.1:8000 and Application startup complete. Leave this terminal running for the rest of the project.

Before you open the browser, what do you think you will see at the /docs page? Will it be a blank page, an error, or something else?

  • Open your browser and navigate to http://127.0.0.1:8000/docs.

You should see the Swagger UI page titled "AI Content API" with your /health endpoint listed. FastAPI generated this interactive documentation automatically from your code.

Don't see the Swagger page?

Check your terminal for error messages. If you see ModuleNotFoundError, your virtual environment may not be active. Press Ctrl+C to stop the server, activate the venv again, and re-run the command.

If you see an error about GEMINI_API_KEY, make sure your .env file is saved in the same folder as main.py.

Why isn't my FastAPI server starting?

✔️ Awesome, I've got everything!

Great. Double check you have saved both .env and main.py.

ⓧ I'd like to double check the full code

Here is the full .env file:

GEMINI_API_KEY=your-api-key-here
DEV_API_KEY=dev-secret-key-123

Here is the full main.py file:

import os

from dotenv import load_dotenv
from fastapi import FastAPI

from google import genai

load_dotenv()

app = FastAPI(
    title="AI Content API",
    description="Production AI API with streaming, auth, and validation",
    version="1.0.0",
)

client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))


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

Your environment is ready and your first endpoint is live. Next up, you will build a real AI-powered endpoint that summarizes text using Gemini.

Build Your First AI Endpoint

In the last step, you set up your environment and confirmed that FastAPI and the Gemini client are connected. Now it's time to build something you can actually call.

A /summarize endpoint is the simplest proof that your AI API works end-to-end. You send text in, Pydantic validates it, Gemini processes it, and you get a structured JSON summary back.

In this step, get ready to:
  • Define Pydantic models for validated request and response data.
  • Create an async POST endpoint that calls Gemini.
  • Test the endpoint live in the Swagger UI.
Define your request and response models

Before building the endpoint itself, you need to define what data goes in and what comes out. Pydantic models give FastAPI a schema for automatic validation and documentation.

  • In main.py, find the import line that says from fastapi import FastAPI.
  • Add this new import directly below it:
from pydantic import BaseModel
  • Now scroll down to the blank line above @app.get("/health").
  • Add the following models between the client line and the health endpoint:
# --- Request/Response Models ---

class SummarizeRequest(BaseModel):
    text: str


class SummarizeResponse(BaseModel):
    summary: str
    model: str

What do these models do?

SummarizeRequest defines the shape of incoming data. Any POST request to your endpoint must include a text field, or FastAPI will reject it automatically with a clear error.

SummarizeResponse defines what the endpoint returns. FastAPI uses this to validate outgoing data and to generate accurate documentation in the Swagger UI.

  • Save main.py.
  • Check your terminal where the dev server is running. You should see the server reload with no errors.

Seeing an import error?

If the terminal shows ModuleNotFoundError, make sure you are running the dev server from inside your virtual environment. Reactivate it and restart with fastapi dev main.py.

If you see a SyntaxError, double-check that the class definitions use a colon after the class name and that each field is indented with four spaces.

help me fix a module import error in my FastAPI project

Create the /summarize endpoint

The models define the contract. Now you need the endpoint that fulfills it by calling Gemini and returning the result.

  • In main.py, add this comment directly above the @app.get("/health") line:
# --- Endpoints ---
  • Now scroll to the end of the file, below the health_check function.
  • Add the summarize endpoint by pasting the following code:
@app.post("/summarize", response_model=SummarizeResponse)
async def summarize(body: SummarizeRequest):
    interaction = client.interactions.create(
        model="gemini-2.5-flash",
        system_instruction="You are a summarization assistant. Provide a concise summary of the given text in 2-3 sentences.",
        input=body.text,
    )
    return SummarizeResponse(summary=interaction.output_text, model="gemini-2.5-flash")

What does this endpoint do?

The @app.post("/summarize") decorator registers a POST route. The response_model=SummarizeResponse parameter tells FastAPI to validate the return value and generate docs from it.

Inside the function, client.interactions.create() sends the user's text to Gemini with a system instruction that constrains it to summarization. The response comes back as an interaction object, and interaction.output_text contains the AI-generated summary.

  • Save main.py.
  • Check your terminal. The dev server should reload again with no errors.

Server showing an error after saving?

If you see NameError: name 'SummarizeRequest' is not defined, your models need to be placed above the endpoint in the file. Python reads top-to-bottom, so the classes must come before any code that references them.

help me fix a NameError in my FastAPI summarize endpoint

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and the terminal shows a clean reload.

ⓧ I'd like to double check the full code

Your main.py should have these sections in order. Scroll through your file and confirm each one is present:

  • Imports: os, dotenv, FastAPI, BaseModel, genai.
  • App setup: load_dotenv(), app = FastAPI(...), client = genai.Client(...).
  • Request/Response Models section: SummarizeRequest with a text field, SummarizeResponse with summary and model fields.
  • Endpoints section: /health GET endpoint, then /summarize POST endpoint calling client.interactions.create().
Test the endpoint in the Swagger UI

Your endpoint is live. The Swagger UI at /docs already has an interactive form you can use to send a real request.

Before you test it, what do you think you will see if you send a paragraph of text to /summarize?

  • Switch back to your browser at http://127.0.0.1:8000/docs and refresh the page.
  • You should see a new POST /summarize section listed below the /health endpoint.
  • Click on POST /summarize to expand it.
  • Click Try it out.
  • Replace the request body with this JSON:
{
  "text": "FastAPI is a modern Python web framework that uses type hints for automatic validation and documentation. It is built on top of Starlette and Pydantic, providing both high performance and developer-friendly features. FastAPI supports asynchronous request handling out of the box, making it ideal for I/O-bound applications like AI APIs."
}
  • Click Execute.

You should see a 200 response with a JSON body containing a summary field (a 2-3 sentence AI-generated summary) and a model field showing gemini-2.5-flash.

What is Swagger UI?

Swagger UI is an interactive API documentation page that FastAPI generates automatically from your code. Every endpoint, request model, and response model you define appears here as a testable form.

You did not write any documentation manually. FastAPI reads your Pydantic models and type hints to produce the entire page.

Getting a 500 error or no response?

If you see a 500 Internal Server Error, check your terminal for the traceback. The most common cause is an invalid or missing GEMINI_API_KEY in your .env file.

If the request hangs for more than 30 seconds, your API key may not have free-tier access activated yet. Return to Google AI Studio, confirm your key is listed, and try generating the key again.

help me debug a 500 error on my FastAPI /summarize endpoint

Your first AI endpoint is live and returning real summaries. Next up, you will see exactly why this endpoint is not ready for production by throwing invalid inputs at it.

See Why This Breaks in Production

Your /summarize endpoint works when everything goes right. A valid request comes in, Gemini processes it, and a clean summary comes back.

Production APIs face hostile conditions. Users send garbage input, attackers probe for weaknesses, and upstream services go down without warning. In this step, you will stress-test your endpoint and see exactly how it fails.

In this step, get ready to:
  • Test the endpoint with invalid inputs and observe unstructured error responses.
  • Confirm there is no authentication protecting your API.
  • Trigger a Gemini API error and see the unhelpful 500 response.
Test the endpoint with invalid inputs

Your endpoint has no custom validation beyond Pydantic's default type checking. Let's see what happens when users send data your code does not expect.

Before you test, predict: what do you think will happen if you send an empty string as the text field?

  • Return to the Swagger UI at http://127.0.0.1:8000/docs in your browser.
  • Click the POST /summarize endpoint to expand it.
  • Click Try it out.
  • Replace the request body with this empty-string payload:
{"text": ""}
  • Click Execute.

You get a 500 Internal Server Error. The response body says something like Internal Server Error with no useful details about what went wrong.

Why did this fail?

Pydantic's text: str type check only confirms the value is a string. An empty string passes that check, so the request reaches the Gemini API. Gemini rejects the empty input, raising an unhandled exception that crashes into a raw 500.

Your server gave the client zero information about why the request failed. In production, this makes debugging impossible for API consumers.

Now try sending a request with a completely missing field. Predict: will FastAPI handle this differently?

  • In the same Swagger UI panel, replace the request body with:
{}
  • Click Execute.

This time you get a 422 Unprocessable Entity with a JSON body containing internal details about your data model.

What is this 422 response leaking?

FastAPI returns a detailed error with loc, msg, and type fields that expose your internal model structure. An attacker can use these details to map out exactly what fields your API expects and what types they require.

Confirm there is no authentication

Right now, anyone who knows your server address can call the endpoint. There is no API key, no token, nothing stopping unlimited access.

  • Open a new terminal window (keep the dev server running in the original one).
  • Call your endpoint with no credentials by running this command:
curl -X POST http://127.0.0.1:8000/summarize -H "Content-Type: application/json" -d '{"text": "FastAPI is a modern Python web framework for building APIs."}'

You get a successful response with a summary. No API key was required.

Why does this matter?

Every call to your /summarize endpoint consumes your Gemini API quota. Without authentication, anyone who discovers your URL can drain your free tier credits in minutes by sending automated requests.

Observe how Gemini API errors surface

Your code has no error handling around the Gemini call. Let's see what happens when the upstream service rejects your request.

  • Switch to the terminal where your dev server is running.
  • Stop the server by pressing Ctrl+C.
  • Open .env in VS Code and temporarily change your API key to an invalid value:
GEMINI_API_KEY=definitely-not-a-real-key
  • Save .env.
  • Restart the dev server by running:
fastapi dev main.py

Before you test, predict: what do you think the client will see when Gemini rejects the request because of a bad key?

  • Switch to Swagger UI and send a valid request to POST /summarize with any text.
  • Click Execute.

The client gets a generic 500 Internal Server Error with no indication of what went wrong.

  • Now check the terminal where your dev server is running.

You see a full Python traceback with the Gemini SDK error message printed to the console. This traceback exposes your internal implementation details in the server logs.

What does this reveal?

The client received zero useful information. The server logs dumped a raw traceback. In production, this pattern means your API consumers cannot debug failures, and your logs are full of unstructured noise instead of actionable error data.

Whether the failure comes from a wrong key, a rate limit hit, or a Gemini service outage, the result is always the same: an opaque 500.

  • Stop the dev server with Ctrl+C.
  • Open .env and restore your real Gemini API key.
  • Save .env.
  • Restart the dev server by running:
fastapi dev main.py

Server won't start after restoring the key?

Double-check that you replaced the invalid key with your actual Gemini API key in .env. Make sure there are no extra spaces around the = sign.

Help me debug why my FastAPI server won't start after changing my .env file

Confirm the server is healthy by opening http://127.0.0.1:8000/docs in your browser. You should see the Swagger UI load successfully.

What you just proved

Your API currently has three production-critical gaps: no input validation (empty strings crash the server), no authentication (anyone can consume your quota), and no error handling (upstream failures produce opaque 500s). The next step fixes all three.

You have felt exactly where your API breaks. Next, you will add authentication, input validation, and structured error handling to close every one of these gaps.

Add Authentication and Error Handling

You saw first-hand how your endpoint breaks with invalid inputs, leaks Python tracebacks to callers, and lets anyone with the URL consume your Gemini quota. Those are exactly the gaps that separate a prototype from a production API.

In this step, you'll close all three gaps. You'll add API key authentication to gate access, a global exception handler that returns structured JSON on failures, and input validation that catches bad data before it ever reaches Gemini.

In this step, get ready to:
  • Create an authentication dependency that validates an X-API-Key header on every request.
  • Add a global exception handler and structured error responses.
  • Add input validation that rejects empty or oversized text before calling the AI.
Create the authentication dependency

Right now, anyone who discovers your API URL can call it freely. A dependency in FastAPI is a function that runs before your endpoint logic. You'll use one to check every incoming request for a valid API key header.

  • In main.py, find the import line that says:
from fastapi import FastAPI
  • Replace it with these updated imports, and add the new StreamingResponse and field_validator imports below it:
from fastapi import FastAPI, HTTPException, Depends, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, field_validator

What are these new imports?

  • HTTPException lets you return proper HTTP error responses (like 401 or 503) with structured JSON bodies.
  • Depends is FastAPI's dependency injection system. It runs a function before your endpoint and passes the result in.
  • Request gives you access to the raw incoming request, including headers.
  • StreamingResponse and field_validator are for features you'll add later in this step and the next.
  • Now add the VALID_API_KEYS set and the verify_api_key function. Place this code directly below the client = genai.Client(...) line, above the # --- Request/Response Models --- comment:
VALID_API_KEYS = {os.getenv("DEV_API_KEY", "dev-secret-key-123")}


# --- Authentication ---

def verify_api_key(request: Request) -> str:
    # Extract the X-API-Key header from the incoming request
    api_key = request.headers.get("X-API-Key")
    # Reject if missing or not in our allowed set
    if not api_key or api_key not in VALID_API_KEYS:
        raise HTTPException(
            status_code=401,
            detail={"error": "unauthorized", "message": "Invalid or missing API key"},
        )
    return api_key

What does this code do?

VALID_API_KEYS loads the DEV_API_KEY value from your .env file into a Python set. FastAPI will call verify_api_key before any endpoint that depends on it.

If the X-API-Key header is missing or doesn't match a key in the set, the function raises a 401 error with a structured JSON body. The request never reaches your endpoint logic.

  • Now wire the dependency into your endpoint. Find the /summarize function signature:
async def summarize(body: SummarizeRequest):
  • Replace it with this updated signature that adds the auth dependency:
async def summarize(body: SummarizeRequest, api_key: str = Depends(verify_api_key)):
  • Save main.py.
  • Your dev server auto-reloads. Switch to the Swagger UI at http://127.0.0.1:8000/docs.
  • Expand the POST /summarize section and click Try it out.
  • Leave the X-API-Key header empty and click Execute.

You should see a 401 response with the JSON body:

{"detail": {"error": "unauthorized", "message": "Invalid or missing API key"}}

Your endpoint is now locked down. Only requests carrying the correct key get through.

Still seeing a 200 response?

  • Make sure you saved main.py and the server reloaded (check the terminal for a reload message).
  • Confirm your function signature includes api_key: str = Depends(verify_api_key) as the second parameter.
  • help me debug why my /summarize endpoint isn't checking for an API key
Add error handling and response documentation

Authentication is working, but if the Gemini API fails (network issue, quota exceeded), your server still crashes with a raw 500 error. A global exception handler catches any unhandled exception and returns a clean JSON response instead of a traceback.

  • In main.py, add the ErrorResponse model directly below the SummarizeResponse class. Then add the global exception handler below it:
class ErrorResponse(BaseModel):
    error: str
    message: str


# --- Error Handling ---

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    from fastapi.responses import JSONResponse

    # Let FastAPI's own HTTP exceptions pass through unchanged
    if isinstance(exc, HTTPException):
        raise exc
    # Catch everything else and return a structured 503
    return JSONResponse(
        status_code=503,
        content={"error": "service_unavailable", "message": "AI service temporarily unavailable"},
    )

What does this code do?

ErrorResponse is a Pydantic model that documents the shape of your error responses in the Swagger docs. It tells API consumers exactly what fields to expect when something goes wrong.

The global_exception_handler catches any exception that isn't already an HTTPException. Instead of leaking a Python traceback, it returns a clean 503 JSON response.

  • Now replace your entire /summarize endpoint (from the @app.post decorator to the return statement) with this version that adds error response docs and a try/except:
@app.post(
    "/summarize",
    response_model=SummarizeResponse,
    responses={401: {"model": ErrorResponse}, 503: {"model": ErrorResponse}},
)
async def summarize(body: SummarizeRequest, api_key: str = Depends(verify_api_key)):
    try:
        interaction = client.interactions.create(
            model="gemini-2.5-flash",
            system_instruction="You are a summarization assistant. Provide a concise summary of the given text in 2-3 sentences.",
            input=body.text,
        )
        return SummarizeResponse(summary=interaction.output_text, model="gemini-2.5-flash")
    except Exception as exc:
        raise HTTPException(
            status_code=503,
            detail={"error": "ai_service_error", "message": "Failed to generate summary"},
        )

What changed in the endpoint?

  • The responses parameter tells Swagger UI to document the 401 and 503 error shapes. API consumers can see exactly what errors to expect.
  • The try/except block wraps the Gemini call. If the API is down or returns an error, your server returns a structured 503 instead of crashing.
  • Save main.py.
  • Refresh the Swagger UI at http://127.0.0.1:8000/docs.
  • Expand POST /summarize and scroll down to the Responses section.

You should see documented 401 and 503 response schemas alongside the 200 success response.

Server won't reload?

  • Check your terminal for a syntax error. The most common issue is a missing comma between the response_model and responses parameters in the decorator.
  • Make sure the ErrorResponse class is defined above the @app.post decorator that references it.
  • help me fix a syntax error in my FastAPI endpoint decorator
Add input validation to reject bad data early

The last gap is input validation. Without it, someone can send an empty string or a 10MB payload and your API wastes a Gemini call on garbage. Pydantic's field_validator lets you add custom checks that run before your endpoint code touches the data.

  • In main.py, find the SummarizeRequest class. Add the validator method inside the class, directly below the text: str field:
    @field_validator("text")
    @classmethod
    def text_must_not_be_empty(cls, v: str) -> str:
        # Reject empty or whitespace-only strings
        if not v.strip():
            raise ValueError("Text must not be empty")
        # Reject inputs over 10,000 characters
        if len(v) > 10000:
            raise ValueError("Text must be 10,000 characters or fewer")
        return v

What does this validator do?

The @field_validator("text") decorator tells Pydantic to run this method every time a SummarizeRequest is created. It checks two things: the text isn't empty or whitespace-only, and it doesn't exceed 10,000 characters.

If either check fails, Pydantic raises a 422 validation error with your custom message before your endpoint code runs. No wasted Gemini API calls on bad input.

  • Save main.py.
  • In the Swagger UI, click Try it out on POST /summarize.
  • Set the X-API-Key header to dev-secret-key-123 (so auth passes).
  • In the request body, send {"text": " "} (just whitespace).
  • Click Execute.

You should see a 422 response with a validation error message containing "Text must not be empty".

Still getting a 200 with empty text?

  • Make sure the text_must_not_be_empty method is indented inside the SummarizeRequest class, not outside it.
  • Confirm you have both @field_validator("text") and @classmethod decorators, in that order.
  • help me fix my Pydantic field_validator that isn't running

Now let's confirm the full flow works end-to-end. Before you run the next test, predict: what do you think will happen if you send a valid request with the correct API key?

  • In the Swagger UI, set X-API-Key to dev-secret-key-123.
  • Set the request body to {"text": "FastAPI is a modern Python web framework designed for building APIs. It uses Python type hints for automatic data validation and generates interactive documentation."}.
  • Click Execute.

You should see a 200 response with an AI-generated summary. Authentication passed, validation passed, and Gemini returned a result through your hardened endpoint.

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and your server is running without errors.

ⓧ I'd like to double check the full code

Scroll back through your work and confirm these sections are present in order:

  • Updated imports (HTTPException, Depends, Request, StreamingResponse, field_validator).
  • VALID_API_KEYS set below the client initialization.
  • verify_api_key function in the Authentication section.
  • SummarizeRequest with the text_must_not_be_empty field validator.
  • ErrorResponse model below SummarizeResponse.
  • global_exception_handler in the Error Handling section.
  • Updated /summarize endpoint with Depends(verify_api_key), try/except, and responses in the decorator.

Your API now protects its resources, validates input, and fails gracefully. Next up, you'll add the most impressive feature for interviews: real-time streaming chat using Server-Sent Events.

Add Streaming Chat with Server-Sent Events

Your API can summarize text, authenticate requests, and handle errors gracefully. That covers the basics of a production backend.

Now you will add the feature that makes modern AI APIs feel magical: real-time streaming. Instead of waiting for the entire AI response to generate before sending anything back, your /chat endpoint will deliver tokens one-by-one as they are generated. This is the exact pattern that makes ChatGPT feel responsive. You will implement it using Server-Sent Events (SSE), an HTTP standard for pushing real-time data from server to client.

In this step, get ready to:
  • Create a validated ChatRequest model for incoming chat messages.
  • Build a streaming endpoint that delivers AI responses token-by-token via SSE.
  • Test real-time streaming with curl and watch tokens arrive live in your terminal.
Add the ChatRequest model

Before building the streaming endpoint, you need a Pydantic model to validate incoming chat messages. This follows the same pattern you used for SummarizeRequest.

  • In main.py, find the top of the file where your imports are. Add these two new imports directly after import os:
import time
from collections import defaultdict

Why add these imports now?

These imports are not used in this step, but they prepare your file for the rate limiting feature coming in the Secret Mission. Adding them now keeps your imports organized in one place.

  • Now find the ErrorResponse class in your models section. Add the following ChatRequest model directly above it:
class ChatRequest(BaseModel):
    message: str

    @field_validator("message")
    @classmethod
    def message_must_not_be_empty(cls, v: str) -> str:
        if not v.strip():
            raise ValueError("Message must not be empty")
        return v

What does this model do?

The ChatRequest model accepts a message string from the client. The field_validator ensures the message is not empty or whitespace-only. If someone sends a blank message, FastAPI returns a clear validation error instead of wasting a Gemini API call.

  • Save main.py.
  • Check your terminal where the dev server is running. You should see a reload message confirming FastAPI picked up the changes with no errors.

Seeing a syntax error in the terminal?

Make sure your ChatRequest class is placed between SummarizeResponse and ErrorResponse. Check that the @field_validator decorator has @classmethod directly below it.

Ask for help: My FastAPI dev server shows a syntax error after adding ChatRequest. What did I miss?

Create the streaming chat endpoint

This is where the magic happens. Your endpoint will call the Gemini API with stream=True and yield each text chunk as it arrives, formatted as an SSE event.

  • In main.py, scroll to the bottom of the file (after the summarize endpoint). Add the following /chat endpoint:
@app.post("/chat")
async def chat(body: ChatRequest, api_key: str = Depends(verify_api_key)):
    async def event_stream():
        try:
            # Call Gemini with streaming enabled
            stream = client.interactions.create(
                model="gemini-2.5-flash",
                input=body.message,
                stream=True,
            )
            # Yield each text chunk as an SSE data line
            for event in stream:
                if event.event_type == "step.delta":
                    if event.delta.type == "text":
                        yield f"data: {event.delta.text}\n\n"
            yield "data: [DONE]\n\n"
        except Exception:
            yield f"data: [ERROR] AI service unavailable\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

What does this endpoint do?

The event_stream() function is an async generator nested inside the endpoint. It calls Gemini with stream=True, which returns chunks as they are generated instead of waiting for the full response.

Each chunk is formatted as data: {text}\n\n. This is the SSE protocol format that clients expect. When all chunks are sent, it yields data: [DONE] to signal the stream is complete. If anything goes wrong, the client gets an error event instead of a crash.

The StreamingResponse wrapper tells FastAPI to send chunks as they are yielded rather than buffering the entire response. The media_type="text/event-stream" header tells the client this is an SSE stream.

  • Save main.py.
  • Check the dev server terminal. You should see a successful reload with no errors.

Server not reloading or showing an error?

Make sure the /chat endpoint is placed after the /summarize endpoint, not inside it. Check that the return StreamingResponse(...) line is at the same indentation level as async def event_stream() (both inside the chat function, but return is outside the nested function).

Ask for help: My FastAPI server shows an error after adding the /chat streaming endpoint. Can you help me fix the indentation?

✔️ Awesome, I've got everything!

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

ⓧ I'd like to double check the full code

Your main.py should now contain these sections in order. Scroll through your file and confirm each is present:

  • Imports: os, time, defaultdict, dotenv, fastapi, pydantic, genai.
  • App setup: load_dotenv(), FastAPI() instance, genai.Client, VALID_API_KEYS.
  • Authentication: verify_api_key function.
  • Models: SummarizeRequest, SummarizeResponse, ChatRequest (new), ErrorResponse.
  • Error handling: global_exception_handler.
  • Endpoints: /health, /summarize, /chat (new).
Test streaming with curl

Time to see streaming in action. You will use curl with the -N flag, which disables output buffering so you see each token the moment it arrives.

Before you run this, what do you expect to see in your terminal? Will the entire response appear at once, or will words appear gradually?

  • Open a new terminal window (your dev server is still running in the first one).
  • Send a streaming chat request by running this command:
curl -N -X POST http://127.0.0.1:8000/chat -H "X-API-Key: dev-secret-key-123" -H "Content-Type: application/json" -d '{"message": "Explain async programming"}'

Watch your terminal. You should see data: lines appearing one-by-one as Gemini generates the response. Each line contains a small chunk of text. The stream finishes with data: [DONE].

What is the -N flag doing?

By default, curl buffers output and delivers data in chunks. The -N flag (short for --no-buffer) disables this buffering so you see each SSE event the instant it arrives from the server. Without it, curl might wait and dump everything at once, defeating the point of streaming.

Not seeing tokens stream in?

  • If you get a 401 error, make sure the X-API-Key value matches what is in your .env file (dev-secret-key-123).
  • If you see data: [ERROR] AI service unavailable, your GEMINI_API_KEY may be invalid. Check your .env file has the correct key from Google AI Studio.
  • If the response appears all at once instead of streaming, make sure you included the -N flag in your curl command.

Ask for help: My /chat endpoint is not streaming tokens. How do I debug SSE streaming in FastAPI?

You just built the most impressive feature of your API. A recruiter watching tokens stream in real-time immediately understands you can build the same architecture behind production AI products. Next up, you will put this to the test in a secret challenge.

Secret mission

Add Rate Limiting

Right now, anyone with a valid API key can send unlimited requests and drain your entire Gemini quota in minutes. Can you implement a sliding window rate limiter that blocks abusive traffic and returns a proper 429 Too Many Requests response?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally and uses the Gemini free tier, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • Your project folder on the Desktop (containing main.py, .env, and the virtual environment).
  • The running FastAPI dev server at http://127.0.0.1:8000.
  • Your Gemini API key in Google AI Studio.

Keep everything running

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

  • Your local dev server continues running in the terminal. The Gemini free tier has no charges, so your API key costs nothing to keep.
  • If you close the terminal by accident, restart the server by activating your virtual environment and running fastapi dev main.py from your project folder.

Pause - I'll come back to this later

Shut down the running server to free up your terminal. Your project files stay on disk for next time.

  • Switch to the terminal where the dev server is running.
  • Press Ctrl+C to stop the FastAPI server.
  • Your project folder and .env file remain on your Desktop. When you are ready to resume, activate your virtual environment and run fastapi dev main.py again.

Delete - I don't want to use this again

Remove all project files and revoke your API key. This is a full cleanup.

  • Switch to the terminal where the dev server is running and press Ctrl+C to stop it.
  • Deactivate your virtual environment by running:
deactivate
  • Delete your project folder by running the command for your operating system.

macOS/Linux:

cd ~/Desktop
rm -rf ai-content-api

Windows:

cd ~\Desktop
rmdir /s /q ai-content-api
  • Delete your Gemini API key by visiting aistudio.google.com. Open the API keys section, find the key you created for this project, and delete it.

Nice Work!

Nice Work!

Nice work! You just built a production-quality AI Content API powered by FastAPI and Google Gemini. This is the same backend architecture that powers real AI SaaS products.

You learned how to:

  • Built a multi-endpoint AI API with FastAPI and integrated the Gemini API with async patterns and environment-based configuration.
  • Implemented real-time streaming responses using Server-Sent Events, the same pattern that makes ChatGPT feel responsive.
  • Added production hardening with API key authentication, Pydantic validation, and structured error handling that returns clean JSON responses with proper HTTP status codes.
  • Secret Mission: Implemented rate limiting with a sliding window algorithm to protect against API abuse.

Ready to quiz yourself?