Build an AI Document Analyzer API
Build a FastAPI REST API that extracts structured JSON from text using Claude.
Introduction
30 Second Summary
Every day, useful information is buried inside messy text. Job postings, contracts, emails, and articles all contain structured data (names, dates, money, topics) trapped in unstructured paragraphs.
In this project, you will build a REST API that accepts raw text and uses Claude to return clean, structured JSON with extracted entities, document classification, and key points.
What You'll Build
A FastAPI application that sends unstructured documents to Claude and returns validated, structured analysis results through a production-quality endpoint.
By the end of this project, you'll have:
- A working REST API that transforms messy text into structured data, with auto-generated interactive documentation.
- Production patterns implemented: Pydantic input validation, retry logic with exponential backoff, and structured JSON logging.
- Hands-on experience with uv (the modern Python package manager) and the Anthropic SDK for calling Claude programmatically.
- Secret Mission: a batch analysis endpoint that processes multiple documents concurrently using Python's asyncio.
Are there any prerequisites?
You should have working knowledge of Python (functions, classes, dictionaries) and a basic understanding of HTTP requests. You do not need prior experience with FastAPI, uv, or the Anthropic API.
Set Up Your Environment
Before you can build an AI-powered API, you need the right tools in place.
In this step, you will install uv (a modern Python package manager), scaffold your project, and connect to the Anthropic API so your code can talk to Claude.
In this step, get ready to:
- Install uv and verify it works.
- Create the project directory with dependencies.
- Set up your Anthropic API key.
Install uv
uv is the modern Python package manager that replaces pip and poetry. It is 10-100x faster and handles project creation, dependency management, and virtual environments in a single tool.
- 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.
- Install uv by running the installer for your platform:
macOS
curl -LsSf https://astral.sh/uv/install.sh | sh
Windows
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
What does this command do?
This downloads the official uv installer script from Astral (the company behind uv) and runs it.
The installer places the uv binary on your PATH so you can use it from any terminal window.
- The installer may ask you to restart your terminal or run a source command to update your PATH. Follow the instructions it prints.
- Verify uv is installed by running:
uv --version
✔️ I see a version number
You should see something like uv 0.11.21 (or a similar version). You are ready to move on.
ⓧ Command not found
Your terminal cannot find the uv binary. Try these fixes:
- Close your terminal completely and open a new one. The installer updated your PATH, but changes only apply to new sessions.
- On macOS, try running source ~/.zshrc or source ~/.bashrc depending on your shell.
- On Windows, close PowerShell and reopen it as administrator, then re-run the install command.
After restarting, run uv --version again.
Create your project and add dependencies
uv can scaffold an entire Python project in one command. It creates a pyproject.toml, initializes a Git repo, and sets up a virtual environment automatically.
- Move to your Desktop by running this command:
cd ~/Desktop
- Create the project by running:
uv init document-analyzer
What does uv init create?
This command creates a document-analyzer folder on your Desktop containing: .git/, .gitignore, .python-version, README.md, main.py, and pyproject.toml.
The pyproject.toml file is your project's configuration hub. It tracks dependencies, Python version, and metadata.
- Move into your new project folder by running:
cd document-analyzer
- Pin the project to Python 3.11 by running:
uv python pin 3.11
- Add the three dependencies your project needs by running:
uv add fastapi anthropic uvicorn
What are these dependencies?
- fastapi is the web framework for building your API endpoints.
- anthropic is the official Python SDK for communicating with Claude.
- uvicorn is the ASGI server that runs your FastAPI app.
- Confirm the files exist by running:
ls
You should see pyproject.toml, main.py, README.md, and a .python-version file in the output.
Seeing an error about Python 3.11?
If uv add reports that it cannot find Python 3.11, uv will attempt to download it automatically. If the download fails, check your internet connection and try again.
help me fix a Python 3.11 error when running uv add
Set up your Anthropic API key
Your API needs credentials to talk to Claude. The Anthropic Python SDK reads an environment variable called ANTHROPIC_API_KEY automatically, so you do not need to hardcode the key in your source files.
- Open your browser and navigate to console.anthropic.com.
- Create an account (or sign in if you already have one).
- Click API Keys in the left sidebar.
- Click Create Key.
- Give it a name like document-analyzer and click Create Key.
- Copy the key immediately. You will not be able to see it again after closing the dialog.
Why copy the key immediately?
Anthropic only shows your full API key once, at the moment of creation. After you close the dialog, the key is masked and cannot be revealed again. If you lose it, you will need to create a new one.
- Back in your terminal (still inside the document-analyzer folder), export the key as an environment variable:
macOS
export ANTHROPIC_API_KEY=your-api-key-here
Windows
$env:ANTHROPIC_API_KEY="your-api-key-here"
- Replace your-api-key-here with the actual key you copied from the Anthropic console.
Why an environment variable?
Storing secrets in environment variables keeps them out of your source code. This means you can push your project to GitHub without accidentally leaking credentials.
The anthropic SDK reads ANTHROPIC_API_KEY automatically. You never need to pass it explicitly in your code.
Key not working later?
Environment variables only last for the current terminal session. If you close your terminal and reopen it, you will need to export the key again before running your server.
help me troubleshoot my ANTHROPIC_API_KEY not being found
- Open the document-analyzer folder in VS Code. Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
- Type VS Code and press Enter to open it.
- In VS Code, click File in the top menu bar.
- Click Open Folder.
- Navigate to your Desktop, select the document-analyzer folder, and click Open.
Your environment is ready. You have uv managing your project, all three dependencies installed, and your API key available in the terminal. Next up, you will define the data models that shape every request and response your API handles.
Define Your Data Models with Pydantic
Your environment is ready with uv, FastAPI, and the Anthropic SDK installed. Now it's time to define the exact shape of data your API will accept and return.
Without strict data definitions, your API would accept any input and produce unpredictable output. Pydantic models solve this by acting as a contract: they validate every incoming request and guarantee a consistent response structure.
In this step, get ready to:
- Create request and response models that define your API's data contract.
- Build nested entity and classification models for structured extraction results.
- Add field validators and test them in a Python shell.
Create the request model
Every request to your API needs a predictable shape. The AnalysisRequest model defines what callers must send: a block of text to analyze and an optional hint about the document type.
- In VS Code, create a new file called models.py inside your document-analyzer folder.
- Add the imports and request model by pasting this code:
from pydantic import BaseModel, Field, field_validator
class AnalysisRequest(BaseModel):
# The raw text a caller sends for analysis (10-10000 characters)
text: str = Field(..., min_length=10, max_length=10000, description="The document text to analyze")
# Optional hint to improve classification accuracy
document_type: str | None = Field(None, description="Optional hint about the document type")
What does this code do?
The Field(...) with the ellipsis means the field is required. min_length=10 and max_length=10000 enforce character limits at the validation layer, so callers can't send empty strings or massive documents.
The str | None type with Field(None, ...) makes document_type optional. If a caller skips it, the value defaults to None.
- Save models.py.
Seeing a red underline on the import?
If VS Code shows an error on from pydantic import ..., your editor may not be using the project's virtual environment. Press Cmd+Shift+P (macOS) or Ctrl+Shift+P (Windows), type Python: Select Interpreter, and choose the one inside your .venv folder.
Help me fix my Python interpreter in VS Code
Add the entity and classification models
Your API needs to return structured entities (people, organizations, dates) and a document classification. Each of these gets its own model so Pydantic can validate them independently.
- In models.py, add the Entity model below the AnalysisRequest class:
class Entity(BaseModel):
# The extracted entity's name (e.g. "Acme Corp", "$500,000")
name: str = Field(..., description="The entity name")
# Category: person, organization, date, money, or location
type: str = Field(..., description="Entity type: person, organization, date, money, location")
# Surrounding text that gives the entity meaning
context: str = Field(..., description="Brief context where entity appears")
What does this model represent?
Each Entity captures one piece of structured data that Claude extracts from the text. For example, if the input mentions "Sarah joined Google in 2024", you'd get three entities: Sarah (person), Google (organization), and 2024 (date).
- Below the Entity class, add the Classification model:
class Classification(BaseModel):
# What kind of document this is
document_type: str = Field(..., description="Type of document: email, article, contract, job_posting, other")
# The main subject matter
topic: str = Field(..., description="Main topic of the document")
# The writing style and urgency
tone: str = Field(..., description="Tone: formal, informal, urgent, neutral, persuasive")
Why separate models instead of one big dictionary?
Nested models give you type safety at every level. If Claude returns a classification missing the tone field, Pydantic catches it immediately instead of letting bad data propagate through your system.
- Save models.py.
Seeing a syntax error?
Make sure each class definition has proper indentation. All field lines should be indented one level (4 spaces) inside their class. Check that there's a blank line between the end of one class and the start of the next.
Help me fix a syntax error in my models.py
Add the response model and validators
The AnalysisResponse model defines exactly what your API returns. It brings together the entity and classification models you just created, plus a confidence score that Claude rates its own analysis with.
- Below the Classification class in models.py, add the response model:
class AnalysisResponse(BaseModel):
# All entities Claude extracted from the document
entities: list[Entity] = Field(default_factory=list, description="Extracted entities")
# Document type, topic, and tone classification
classification: Classification = Field(..., description="Document classification")
# 3-5 concise takeaways from the document
key_points: list[str] = Field(default_factory=list, description="Key points from the document")
# How confident Claude is in its analysis (0.0 to 1.0)
confidence_score: float = Field(..., ge=0.0, le=1.0, description="Confidence in the analysis")
What do these constraints do?
The default_factory=list means entities and key_points default to empty lists if Claude doesn't extract any. The ge=0.0, le=1.0 on confidence_score enforces that the value must be between 0.0 and 1.0.
If Claude returns 1.5 by mistake, Pydantic rejects it before bad data reaches your callers.
Now add a custom validator to AnalysisRequest that catches whitespace-only input. The min_length constraint allows 10 spaces to pass, but your API should reject meaningless text.
- Scroll back up to the AnalysisRequest class.
- Find the document_type field line.
- Add the following validator method directly below it (still inside the class):
@field_validator("text")
@classmethod
def text_must_not_be_empty(cls, v: str) -> str:
# Reject strings that are only whitespace
if not v.strip():
raise ValueError("Text must contain non-whitespace characters")
return v
What does this validator do?
The @field_validator("text") decorator runs this method every time someone creates an AnalysisRequest. It strips whitespace and checks if anything remains.
A string of 10+ spaces would pass min_length validation, but this custom check catches it.
- Save models.py.
✔️ Awesome, I've got everything!
Great. Double check you've saved your file before moving on to the test.
ⓧ I'd like to double check the full code
Here is the complete models.py file for comparison:
from pydantic import BaseModel, Field, field_validator
class AnalysisRequest(BaseModel):
text: str = Field(..., min_length=10, max_length=10000, description="The document text to analyze")
document_type: str | None = Field(None, description="Optional hint about the document type")
@field_validator("text")
@classmethod
def text_must_not_be_empty(cls, v: str) -> str:
if not v.strip():
raise ValueError("Text must contain non-whitespace characters")
return v
class Entity(BaseModel):
name: str = Field(..., description="The entity name")
type: str = Field(..., description="Entity type: person, organization, date, money, location")
context: str = Field(..., description="Brief context where entity appears")
class Classification(BaseModel):
document_type: str = Field(..., description="Type of document: email, article, contract, job_posting, other")
topic: str = Field(..., description="Main topic of the document")
tone: str = Field(..., description="Tone: formal, informal, urgent, neutral, persuasive")
class AnalysisResponse(BaseModel):
entities: list[Entity] = Field(default_factory=list, description="Extracted entities")
classification: Classification = Field(..., description="Document classification")
key_points: list[str] = Field(default_factory=list, description="Key points from the document")
confidence_score: float = Field(..., ge=0.0, le=1.0, description="Confidence in the analysis")
Now test your models to confirm everything works.
- Test a valid request by running this command in your terminal:
uv run python -c "from models import AnalysisRequest; print(AnalysisRequest(text='This is a sample document for testing the analyzer.'))"
You should see the model printed with text='This is a sample document for testing the analyzer.' and document_type=None.
- Now test that validation rejects short input by running:
uv run python -c "from models import AnalysisRequest; AnalysisRequest(text='short')"
You should see a ValidationError indicating that the string is too short. This confirms Pydantic is rejecting text that doesn't meet the minimum length.
Getting a ModuleNotFoundError?
Make sure you're running the command from inside the document-analyzer directory. The uv run command needs to find your models.py file in the current folder.
Help me fix a ModuleNotFoundError for models
Your data models are locked in. Next up, you'll build the Claude integration layer that takes raw text and produces these exact structures as output.
Build the Claude Integration Layer
Your Pydantic models define the exact shape of data flowing through your API. Now you need the engine that actually produces that structured data from raw text.
In this step, you'll build the module that talks to Claude through the Anthropic SDK. It constructs a prompt, sends text for analysis, retries on transient failures, and parses the AI's JSON response into your validated AnalysisResponse model.
In this step, get ready to:
- Build the core function that sends document text to Claude for analysis.
- Add retry logic to handle transient API failures with exponential backoff.
- Parse Claude's JSON output into your validated Pydantic model.
Create the analyzer module and main function
The analyzer module is the bridge between your API and Claude. It constructs a system prompt with strict JSON formatting instructions, sends the user's text to the model, and delegates error handling to helper functions.
- Create a new file called analyzer.py in your document-analyzer folder.
- Paste the following code into analyzer.py:
import asyncio
import json
import logging
import time
import anthropic
from pydantic import ValidationError
from models import AnalysisResponse
# Set up a logger for this module
logger = logging.getLogger(__name__)
# Create the async Anthropic client (reads ANTHROPIC_API_KEY from environment)
client = anthropic.AsyncAnthropic()
# System prompt instructs Claude to return strictly formatted JSON
SYSTEM_PROMPT = """You are a document analysis engine. Given a document, extract structured information and return ONLY valid JSON matching this exact schema:
{
"entities": [{"name": "string", "type": "person|organization|date|money|location", "context": "string"}],
"classification": {"document_type": "email|article|contract|job_posting|other", "topic": "string", "tone": "formal|informal|urgent|neutral|persuasive"},
"key_points": ["string"],
"confidence_score": 0.0 to 1.0
}
Rules:
- Return ONLY the JSON object, no markdown, no explanation
- confidence_score reflects how confident you are in the analysis (0.0 = uncertain, 1.0 = highly confident)
- Extract ALL entities you can identify
- key_points should be 3-5 concise bullet points
"""
async def analyze_document(text: str, document_type: str | None = None) -> AnalysisResponse:
# Build the user message with optional document type hint
user_message = f"Analyze this document:\n\n{text}"
if document_type:
user_message += f"\n\nHint: This is likely a {document_type}."
# Call Claude with retry logic, then parse the response
response_text = await _call_claude_with_retry(user_message)
return _parse_response(response_text)
What does this code do?
This sets up the foundation of your analyzer module:
- anthropic.AsyncAnthropic() creates an async client that automatically reads your ANTHROPIC_API_KEY environment variable. No need to pass the key manually.
- SYSTEM_PROMPT gives Claude strict formatting rules. By specifying the exact JSON schema and saying "return ONLY the JSON object", you maximize the chance of getting parseable output.
- analyze_document is the public entry point. It builds a user message, calls the retry-wrapped API function, and passes the raw text through the parser.
Why use AsyncAnthropic instead of the sync client?
FastAPI is async-native. When your endpoint awaits an API call to Claude, the server can handle other incoming requests while waiting for Claude's response. A synchronous client would block the entire server thread during each Claude call.
The async client also makes the Secret Mission possible later, where you'll process multiple documents concurrently with asyncio.gather.
- Save analyzer.py.
You won't see a change yet. The next substep builds the retry function that analyze_document calls.
Add retry logic with exponential backoff
API calls to external services can fail for temporary reasons. Rate limits, server overloads, and network hiccups are all transient errors that resolve if you wait and try again.
- In analyzer.py, add the following function below the analyze_document function:
async def _call_claude_with_retry(user_message: str, max_retries: int = 3) -> str:
# Delays between retries: 1s, 2s, 4s (exponential backoff)
delays = [1, 2, 4]
for attempt in range(max_retries):
try:
message = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=SYSTEM_PROMPT,
messages=[{"role": "user", "content": user_message}],
)
return message.content[0].text
except anthropic.RateLimitError as e:
# Rate limited (429) - wait and retry if attempts remain
if attempt < max_retries - 1:
delay = delays[attempt]
logger.warning(f"Rate limited, retrying in {delay}s (attempt {attempt + 1}/{max_retries})")
await asyncio.sleep(delay)
else:
logger.error("Max retries exceeded for rate limit")
raise
except anthropic.APIStatusError as e:
# Server errors (5xx) are transient - retry them
if e.status_code >= 500 and attempt < max_retries - 1:
delay = delays[attempt]
logger.warning(f"Server error {e.status_code}, retrying in {delay}s (attempt {attempt + 1}/{max_retries})")
await asyncio.sleep(delay)
else:
raise
raise RuntimeError("Unexpected retry loop exit")
What does this code do?
This function wraps every Claude API call with automatic retry handling:
- It tries the API call up to 3 times. On the first failure it waits 1 second, then 2 seconds, then 4 seconds. This is exponential backoff.
- anthropic.RateLimitError fires when you've hit the API's request limit (HTTP 429). Waiting a few seconds usually clears it.
- anthropic.APIStatusError with a status code of 500 or above means Anthropic's servers had a temporary problem. These are worth retrying.
- Client errors (4xx other than 429) are NOT retried because they indicate a problem with your request, not a transient failure.
Why write retry logic manually instead of using a library?
Libraries like tenacity exist for this, but writing it manually keeps your dependency count low and teaches you the pattern at the implementation level. In production code you might reach for a library, but understanding exponential backoff from scratch makes you a stronger debugger when retries behave unexpectedly.
- Save analyzer.py.
You won't see a change yet. The next substep adds the response parser to complete the module.
Parse Claude's response into your Pydantic model
LLMs don't always return perfectly clean JSON. Sometimes they wrap output in markdown code fences. Sometimes the structure doesn't match your schema exactly. Your parser needs to handle these gracefully.
- In analyzer.py, add the following function below _call_claude_with_retry:
def _parse_response(response_text: str) -> AnalysisResponse:
try:
# Strip whitespace and remove markdown code fences if present
cleaned = response_text.strip()
if cleaned.startswith("```"):
cleaned = cleaned.split("\n", 1)[1]
cleaned = cleaned.rsplit("```", 1)[0]
# Parse JSON and validate against the Pydantic model
data = json.loads(cleaned)
return AnalysisResponse(**data)
except (json.JSONDecodeError, ValidationError) as e:
logger.error(f"Failed to parse Claude response: {e}")
raise ValueError(f"Failed to parse analysis response: {e}")
What does this code do?
This function takes Claude's raw text output and converts it into your validated AnalysisResponse model:
- It strips markdown code fences (```json ... ```) that LLMs sometimes add despite being told not to.
- json.loads parses the string into a Python dictionary.
- AnalysisResponse(**data) validates every field against your Pydantic schema. If any field is missing or has the wrong type, ValidationError is raised.
- Both JSONDecodeError and ValidationError are caught and re-raised as a ValueError with a clear message. This gives the calling code a single error type to handle.
- Save analyzer.py.
- Verify the module loads correctly by running this command in your terminal:
uv run python -c "from analyzer import analyze_document; print('Import OK')"
You should see Import OK printed with no errors.
Seeing an import error?
- If you see a ModuleNotFoundError for anthropic, make sure you ran uv add anthropic in the previous step. Run uv sync to re-install dependencies.
- If you see a circular import error mentioning "partially initialized module 'anthropic'", check that you do NOT have a file named anthropic.py in your project folder. A local file with that name shadows the installed package.
- If you see a ModuleNotFoundError for models, make sure models.py exists in the same folder as analyzer.py.
Still stuck? Help me debug my analyzer.py import error
✔️ Awesome, I've got everything!
Great. Double check you've saved analyzer.py and move on to the next step.
ⓧ I'd like to double check the full code
Here's the complete analyzer.py file. Make sure yours matches exactly:
import asyncio
import json
import logging
import time
import anthropic
from pydantic import ValidationError
from models import AnalysisResponse
logger = logging.getLogger(__name__)
client = anthropic.AsyncAnthropic()
SYSTEM_PROMPT = """You are a document analysis engine. Given a document, extract structured information and return ONLY valid JSON matching this exact schema:
{
"entities": [{"name": "string", "type": "person|organization|date|money|location", "context": "string"}],
"classification": {"document_type": "email|article|contract|job_posting|other", "topic": "string", "tone": "formal|informal|urgent|neutral|persuasive"},
"key_points": ["string"],
"confidence_score": 0.0 to 1.0
}
Rules:
- Return ONLY the JSON object, no markdown, no explanation
- confidence_score reflects how confident you are in the analysis (0.0 = uncertain, 1.0 = highly confident)
- Extract ALL entities you can identify
- key_points should be 3-5 concise bullet points
"""
async def analyze_document(text: str, document_type: str | None = None) -> AnalysisResponse:
user_message = f"Analyze this document:\n\n{text}"
if document_type:
user_message += f"\n\nHint: This is likely a {document_type}."
response_text = await _call_claude_with_retry(user_message)
return _parse_response(response_text)
async def _call_claude_with_retry(user_message: str, max_retries: int = 3) -> str:
delays = [1, 2, 4]
for attempt in range(max_retries):
try:
message = await client.messages.create(
model="claude-sonnet-4-6",
max_tokens=1024,
system=SYSTEM_PROMPT,
messages=[{"role": "user", "content": user_message}],
)
return message.content[0].text
except anthropic.RateLimitError as e:
if attempt < max_retries - 1:
delay = delays[attempt]
logger.warning(f"Rate limited, retrying in {delay}s (attempt {attempt + 1}/{max_retries})")
await asyncio.sleep(delay)
else:
logger.error("Max retries exceeded for rate limit")
raise
except anthropic.APIStatusError as e:
if e.status_code >= 500 and attempt < max_retries - 1:
delay = delays[attempt]
logger.warning(f"Server error {e.status_code}, retrying in {delay}s (attempt {attempt + 1}/{max_retries})")
await asyncio.sleep(delay)
else:
raise
raise RuntimeError("Unexpected retry loop exit")
def _parse_response(response_text: str) -> AnalysisResponse:
try:
cleaned = response_text.strip()
if cleaned.startswith("```"):
cleaned = cleaned.split("\n", 1)[1]
cleaned = cleaned.rsplit("```", 1)[0]
data = json.loads(cleaned)
return AnalysisResponse(**data)
except (json.JSONDecodeError, ValidationError) as e:
logger.error(f"Failed to parse Claude response: {e}")
raise ValueError(f"Failed to parse analysis response: {e}")
Your Claude integration layer is complete. Next up, you'll wire this analyzer into a FastAPI endpoint and test it with real documents through the interactive API docs.
Wire Up the FastAPI Endpoint
You have a Claude integration layer that takes raw text and returns structured data. But right now, the only way to call it is from Python code.
In this step, you will expose your analyzer as a REST API using FastAPI. This means any application, language, or tool that speaks HTTP can send text to your API and get structured JSON back.
In this step, get ready to:
- Create a FastAPI app with a health check and document analysis endpoint.
- Add structured JSON logging for production observability.
- Run and test your API using the auto-generated interactive docs.
Create main.py with your FastAPI app and endpoints
FastAPI turns Python functions into HTTP endpoints. You decorate a function with a route, and FastAPI handles request parsing, validation, and response serialization automatically.
- Open main.py in VS Code. This file was created by uv init earlier and contains placeholder code.
- Select all the existing content and replace it with the following imports, app instance, and health check endpoint:
import json
import logging
import sys
import time
from fastapi import FastAPI, HTTPException
from analyzer import analyze_document
from models import AnalysisRequest, AnalysisResponse
app = FastAPI(
title="Document Analyzer API",
description="Extracts structured data from unstructured text using Claude",
version="1.0.0",
)
@app.get("/health")
async def health_check() -> dict:
return {"status": "healthy", "service": "document-analyzer"}
What does this code do?
The FastAPI() constructor creates your application instance. The title, description, and version metadata show up in the auto-generated API docs.
The @app.get("/health") decorator turns health_check into a GET endpoint. Load balancers and monitoring tools use health checks to verify your service is alive.
- Below the health check endpoint, add the document analysis endpoint:
@app.post("/analyze", response_model=AnalysisResponse)
async def analyze(request: AnalysisRequest) -> AnalysisResponse:
start_time = time.time()
logger.info(f"Received analysis request: {len(request.text)} characters")
try:
result = await analyze_document(request.text, request.document_type)
duration = time.time() - start_time
logger.info(f"Analysis complete in {duration:.2f}s, confidence: {result.confidence_score}")
return result
except ValueError as e:
logger.error(f"Analysis failed: {e}")
raise HTTPException(status_code=422, detail=str(e))
except Exception as e:
logger.error(f"Unexpected error: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
What does this endpoint do?
- response_model=AnalysisResponse tells FastAPI to validate the return value against your Pydantic model and generate accurate API docs.
- FastAPI automatically parses the JSON request body into an AnalysisRequest object. If the body fails validation (too short, empty text), FastAPI returns a 422 error before your code even runs.
- The try/except block catches ValueError (malformed Claude responses) and unexpected exceptions separately, returning appropriate HTTP status codes for each.
Add structured JSON logging
Production APIs need logs that are machine-parseable. Plain text logs are hard to search and filter at scale. JSON-structured logs let tools like Datadog or CloudWatch query specific fields directly.
- In main.py, find the line that says:
app = FastAPI(
- Add the following logging infrastructure directly above that line:
class JSONFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
log_data = {
"timestamp": self.formatTime(record),
"level": record.levelname,
"message": record.getMessage(),
"module": record.module,
}
if record.exc_info:
log_data["exception"] = self.formatException(record.exc_info)
return json.dumps(log_data)
def setup_logging() -> None:
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
setup_logging()
logger = logging.getLogger(__name__)
What does this logging setup do?
- JSONFormatter overrides Python's default log formatter. Every log message becomes a JSON object with timestamp, level, message, and module fields.
- setup_logging() is called immediately at module load time. This ensures every log message from your app (and the /analyze endpoint) is formatted as JSON from the start.
- logger = logging.getLogger(__name__) creates a module-level logger that the endpoints reference for request tracking and error reporting.
- Save main.py.
Seeing a red squiggly under logger?
Make sure the setup_logging() call and logger = logging.getLogger(__name__) lines sit between the def setup_logging() function and the app = FastAPI(...) line. The logger variable must be defined before the endpoint functions that use it.
help me fix my logger placement in main.py
✔️ Awesome, I've got everything!
Great. Double check you have saved main.py before moving on.
ⓧ I'd like to double check the full code
import json
import logging
import sys
import time
from fastapi import FastAPI, HTTPException
from analyzer import analyze_document
from models import AnalysisRequest, AnalysisResponse
class JSONFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
log_data = {
"timestamp": self.formatTime(record),
"level": record.levelname,
"message": record.getMessage(),
"module": record.module,
}
if record.exc_info:
log_data["exception"] = self.formatException(record.exc_info)
return json.dumps(log_data)
def setup_logging() -> None:
handler = logging.StreamHandler(sys.stdout)
handler.setFormatter(JSONFormatter())
logging.basicConfig(level=logging.INFO, handlers=[handler])
setup_logging()
logger = logging.getLogger(__name__)
app = FastAPI(
title="Document Analyzer API",
description="Extracts structured data from unstructured text using Claude",
version="1.0.0",
)
@app.get("/health")
async def health_check() -> dict:
return {"status": "healthy", "service": "document-analyzer"}
@app.post("/analyze", response_model=AnalysisResponse)
async def analyze(request: AnalysisRequest) -> AnalysisResponse:
start_time = time.time()
logger.info(f"Received analysis request: {len(request.text)} characters")
try:
result = await analyze_document(request.text, request.document_type)
duration = time.time() - start_time
logger.info(f"Analysis complete in {duration:.2f}s, confidence: {result.confidence_score}")
return result
except ValueError as e:
logger.error(f"Analysis failed: {e}")
raise HTTPException(status_code=422, detail=str(e))
except Exception as e:
logger.error(f"Unexpected error: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
Run the server and test your API
Your API is complete. Time to start the server and send it a real document.
- Start the development server by running this command in your terminal:
uv run uvicorn main:app --reload
What does this command do?
- uv run executes the command inside your project's virtual environment, ensuring all your dependencies are available.
- uvicorn is an ASGI server that runs your FastAPI app.
- main:app tells uvicorn to find the app object inside main.py.
- --reload makes the server restart automatically when you save file changes. Perfect for development.
- You should see output indicating the server is running at http://127.0.0.1:8000.
Server not starting?
- If you see ModuleNotFoundError, make sure you are running the command from inside the document-analyzer directory where your pyproject.toml lives.
- If you see an ImportError mentioning analyzer or models, verify that analyzer.py and models.py are saved in the same folder as main.py.
- If port 8000 is already in use, stop the other process or use --port 8001 instead.
help me debug why my uvicorn server won't start
- Open your browser and navigate to http://127.0.0.1:8000/docs.
You should see FastAPI's interactive Swagger UI with your two endpoints listed: GET /health and POST /analyze.
What is this docs page?
FastAPI auto-generates interactive API documentation from your route decorators and Pydantic models. The /docs endpoint uses Swagger UI, which lets you test endpoints directly from the browser without writing a single line of client code.
- Click on the POST /analyze endpoint to expand it.
- Click Try it out.
- Replace the example request body with this sample text:
{
"text": "Dear Mr. Johnson, I am writing to confirm that Acme Corp will proceed with the $2.5 million acquisition of Widget Industries, effective March 15, 2026. Please have your legal team review the attached terms by Friday. Best regards, Sarah Chen, VP of Corporate Development, Acme Corp.",
"document_type": "email"
}
- Click Execute.
After a few seconds, you should see a 200 response with structured JSON containing extracted entities (Mr. Johnson, Acme Corp, Widget Industries, $2.5 million, March 15 2026), a classification, key points, and a confidence score.
- Switch back to your terminal. You should see JSON-formatted log lines showing the request size and processing time.
Getting a 500 error?
- Check your terminal for the error details. If you see a message about the API key, make sure your ANTHROPIC_API_KEY environment variable is still set in the terminal session where the server is running.
- If you see a 422 error instead, the request body failed validation. Make sure your text is at least 10 characters long.
help me debug the 500 error from my /analyze endpoint
Your API is live, logging requests, and returning structured data from Claude. Next up, you will take this further in the Secret Mission by adding batch processing with concurrent requests.
Secret mission
Add Batch Analysis with Asyncio
Your API handles one document at a time. But in production, clients often need to analyze multiple documents in a single request. Can you add a batch endpoint that processes documents concurrently while respecting rate limits?
Clean Up Your Resources
Clean Up Your Resources
This project runs locally and calls the Anthropic API on demand. There are no ongoing costs as long as the server is stopped. You only incur charges when the API processes requests.
Decide whether to keep your resources running, pause them to come back later, or delete them entirely.
Resources you used:
- The document-analyzer project folder on your Desktop.
- The uvicorn development server running at http://127.0.0.1:8000.
- The ANTHROPIC_API_KEY environment variable exported in your terminal session.
- Your Anthropic API key stored in the console at console.anthropic.com.
- The uv package manager installed on your system.
Keep everything running
No action needed. Choose this if you are still actively building on your Document Analyzer API or plan to extend it soon.
- Your uvicorn server will keep running until you close the terminal or press Ctrl+C.
- The ANTHROPIC_API_KEY environment variable only lasts for your current terminal session. If you open a new terminal, you will need to export it again.
- The uv package manager stays installed globally and is useful for future Python projects.
Pause - I'll come back to this later
Shut down the running server to free up your terminal. Your project files stay intact so you can resume later.
- In the terminal where uvicorn is running, press Ctrl+C to stop the server.
- The ANTHROPIC_API_KEY variable disappears when you close your terminal. Next time you return, export it again before starting the server.
- Your document-analyzer folder and all code remain on your Desktop.
Delete - I don't want to use this again
Remove all project resources and start fresh. Follow each step below to fully clean up.
- Stop the uvicorn server by pressing Ctrl+C in the terminal where it is running.
- Delete the project folder by running this command:
cd ~/Desktop
rm -rf document-analyzer
On Windows, use this instead:
cd ~\Desktop
Remove-Item -Recurse -Force document-analyzer
- Revoke your Anthropic API key so it cannot be used again. Go to console.anthropic.com, navigate to your API keys settings, and delete the key you created for this project.
- Unset the environment variable in your current terminal session:
unset ANTHROPIC_API_KEY
- Remove uv from your system. First, clean up any cached data:
uv cache clean
- Then remove the uv and uvx binaries. On macOS or Linux:
rm ~/.local/bin/uv ~/.local/bin/uvx
On Windows PowerShell:
rm $HOME\.local\bin\uv.exe
rm $HOME\.local\bin\uvx.exe
rm $HOME\.local\bin\uvw.exe
Nice Work!
Nice Work!
Nice work! You just built a production-grade AI-powered REST API that transforms messy, unstructured text into clean, structured JSON.
You've learned how to:
- Build a REST API from scratch using FastAPI and uv, the modern Python package manager.
- Integrate Claude via the Anthropic SDK to extract structured data from unstructured text, with Pydantic validation ensuring type-safe responses.
- Implement production patterns including retry logic with exponential backoff, structured JSON logging, and auto-generated interactive API documentation.
- Secret Mission: Add concurrent batch processing with asyncio and rate-limit-aware semaphores.
Ready to quiz yourself?