Part 2: Building FastAPI Applications with Claude

Part of the LLM API Development 101 Series

My First FastAPI + Claude Integration

Built my first chatbot API on a Friday. Worked perfectly in testing - clean code, proper endpoints, fast responses.

Monday morning: Production meltdown. API timing out under load. Users complaining. My manager asking questions.

The problem? Synchronous Claude API calls blocking FastAPI's event loop. Every request waited for Claude's response (2-5 seconds) while holding the connection. With 50 concurrent users, everything ground to a halt.

Fixed it with async/await: Response time dropped from 4 seconds to 400ms. Handled 10x more concurrent users.

Let me show you how to build it right from the start.

FastAPI Basics

Why FastAPI?

I've built APIs with Flask, Django, Express. FastAPI is best for LLM apps because:

1. Native async support - Perfect for I/O-bound LLM calls 2. Automatic validation - Pydantic models prevent bad requests 3. Auto-generated docs - Interactive API documentation 4. High performance - As fast as Node.js and Go 5. Type hints - Better IDE support and fewer bugs

Installation

pip install fastapi uvicorn python-dotenv anthropic

My complete requirements.txt:

fastapi==0.109.0
uvicorn[standard]==0.27.0
python-dotenv==1.0.0
anthropic==0.25.0
pydantic==2.6.0

First FastAPI App

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.get("/health")
async def health():
    return {"status": "healthy"}

Run it:

uvicorn main:app --reload

Visit: http://localhost:8000/docs - Automatic interactive documentation!

Integrating Claude with FastAPI

Basic Integration

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import anthropic
import os
from dotenv import load_dotenv

load_dotenv()

app = FastAPI(title="Claude API Wrapper")

# Initialize Claude client
claude_client = anthropic.Anthropic(
    api_key=os.getenv("ANTHROPIC_API_KEY")
)

class ChatRequest(BaseModel):
    message: str
    max_tokens: int = 1024

class ChatResponse(BaseModel):
    response: str
    input_tokens: int
    output_tokens: int

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """Send message to Claude."""
    
    try:
        # Call Claude API
        message = claude_client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=request.max_tokens,
            messages=[
                {"role": "user", "content": request.message}
            ]
        )
        
        return ChatResponse(
            response=message.content[0].text,
            input_tokens=message.usage.input_tokens,
            output_tokens=message.usage.output_tokens
        )
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Test it:

curl -X POST "http://localhost:8000/chat" \
  -H "Content-Type: application/json" \
  -d '{"message": "What is FastAPI?"}'

This works but isn't production-ready yet.

Async/Await with Claude

Problem: The anthropic SDK doesn't have native async support. Solution: Use asyncio.to_thread() to run blocking calls in thread pool.

Proper Async Integration

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import anthropic
import os
import asyncio
from dotenv import load_dotenv

load_dotenv()

app = FastAPI()

# Initialize client
claude_client = anthropic.Anthropic(
    api_key=os.getenv("ANTHROPIC_API_KEY")
)

class ChatRequest(BaseModel):
    message: str
    max_tokens: int = 1024
    temperature: float = 1.0

class ChatResponse(BaseModel):
    response: str
    input_tokens: int
    output_tokens: int

async def call_claude_async(message: str, max_tokens: int, temperature: float):
    """Async wrapper for Claude API call."""
    
    def _call():
        return claude_ client.messages.create(
            model="claude-3-5-sonnet-20241022",
            max_tokens=max_tokens,
            temperature=temperature,
            messages=[{"role": "user", "content": message}]
        )
    
    # Run in thread pool
    response = await asyncio.to_thread(_call)
    return response

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """Chat with Claude (async)."""
    
    try:
        message = await call_claude_async(
            request.message,
            request.max_tokens,
            request.temperature
        )
        
        return ChatResponse(
            response=message.content[0].text,
            input_tokens=message.usage.input_tokens,
            output_tokens=message.usage.output_tokens
        )
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Now the event loop doesn't block! Can handle many concurrent requests.

Request Validation with Pydantic

Pydantic validates data automatically. Huge time-saver.

Advanced Request Models

from pydantic import BaseModel, Field, validator
from typing import Optional, List
from enum import Enum

class ClaudeModel(str, Enum):
    """Available Claude models."""
    OPUS = "claude-3-opus-20240229"
    SONNET = "claude-3-5-sonnet-20241022"
    HAIKU = "claude-3-haiku-20240307"

class Message(BaseModel):
    """Chat message."""
    role: str = Field(..., pattern="^(user|assistant)$")
    content: str = Field(..., min_length=1, max_length=10000)

class ChatRequest(BaseModel):
    """Chat request with validation."""
    
    messages: List[Message] = Field(..., min_items=1, max_items=50)
    model: ClaudeModel = ClaudeModel.SONNET
    max_tokens: int = Field(default=1024, ge=1, le=4096)
    temperature: float = Field(default=1.0, ge=0.0, le=1.0)
    system_prompt: Optional[str] = Field(None, max_length=5000)
    
    @validator('messages')
    def validate_message_alternation(cls, messages):
        """Ensure messages alternate between user and assistant."""
        if messages[0].role != "user":
            raise ValueError("First message must be from user")
        return messages

class ChatResponse(BaseModel):
    """Chat response."""
    response: str
    model: str
    usage: dict = Field(..., description="Token usage stats")

FastAPI automatically validates - bad requests get 422 status with detailed errors.

Test validation:

# This will fail - temperature too high
curl -X POST "http://localhost:8000/chat" \
  -H "Content-Type: application/json" \
  -d '{"message": "Hi", "temperature": 2.0}'

Response:

{
  "detail": [
    {
      "loc": ["body", "temperature"],
      "msg": "ensure this value is less than or equal to 1.0",
      "type": "value_error.number.not_le"
    }
  ]
}

Complete Production API

Here's my production-ready FastAPI + Claude application:

"""
Claude API FastAPI Wrapper
Production-ready implementation with proper error handling.
"""
from fastapi import FastAPI, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field, validator
from typing import List, Optional
import anthropic
from anthropic import APIError, RateLimitError, APIConnectionError
import os
import asyncio
import logging
from dotenv import load_dotenv
from enum import Enum

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Load environment
load_dotenv()

# Initialize FastAPI
app = FastAPI(
    title="Claude API Wrapper",
    description="Production-ready Claude API integration",
    version="1.0.0"
)

# CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Configure appropriately in production
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Initialize Claude client
claude_client = anthropic.Anthropic(
    api_key=os.getenv("ANTHROPIC_API_KEY")
)

# Models
class ClaudeModel(str, Enum):
    OPUS = "claude-3-opus-20240229"
    SONNET = "claude-3-5-sonnet-20241022"
    HAIKU = "claude-3-haiku-20240307"

class Message(BaseModel):
    role: str = Field(..., pattern="^(user|assistant)$")
    content: str = Field(..., min_length=1, max_length=100000)

class ChatRequest(BaseModel):
    messages: List[Message] = Field(..., min_items=1, max_items=100)
    model: ClaudeModel = ClaudeModel.SONNET
    max_tokens: int = Field(default=2048, ge=1, le=4096)
    temperature: float = Field(default=1.0, ge=0.0, le=1.0)
    system_prompt: Optional[str] = None
    
    @validator('messages')
    def validate_first_message(cls, messages):
        if messages[0].role != "user":
            raise ValueError("First message must be from user")
        return messages

class Usage(BaseModel):
    input_tokens: int
    output_tokens: int
    total_tokens: int

class ChatResponse(BaseModel):
    response: str
    model: str
    usage: Usage
    finish_reason: str

# Helper functions
async def call_claude_with_retry(
    messages: List[dict],
    model: str,
    max_tokens: int,
    temperature: float,
    system_prompt: Optional[str],
    max_retries: int = 3
):
    """Call Claude API with retry logic."""
    
    def _call():
        params = {
            "model": model,
            "max_tokens": max_tokens,
            "temperature": temperature,
            "messages": messages
        }
        
        if system_prompt:
            params["system"] = system_prompt
        
        return claude_client.messages.create(**params)
    
    for attempt in range(max_retries):
        try:
            response = await asyncio.to_thread(_call)
            return response
            
        except RateLimitError:
            if attempt == max_retries - 1:
                raise HTTPException(
                    status_code=status.HTTP_429_TOO_MANY_REQUESTS,
                    detail="Rate limit exceeded. Please try again later."
                )
            
            wait_time = 2 ** attempt
            logger.warning(f"Rate limited. Retrying in {wait_time}s...")
            await asyncio.sleep(wait_time)
            
        except APIConnectionError:
            if attempt == max_retries - 1:
                raise HTTPException(
                    status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
                    detail="Could not connect to Claude API"
                )
            
            wait_time = 2 ** attempt
            logger.warning(f"Connection error. Retrying in {wait_time}s...")
            await asyncio.sleep(wait_time)
            
        except APIError as e:
            logger.error(f"Claude API error: {e}")
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Claude API error: {str(e)}"
            )
        
        except Exception as e:
            logger.error(f"Unexpected error: {e}")
            raise HTTPException(
                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail=f"Internal error: {str(e)}"
            )

# Endpoints
@app.get("/")
async def root():
    """Root endpoint."""
    return {
        "service": "Claude API Wrapper",
        "version": "1.0.0",
        "status": "healthy"
    }

@app.get("/health")
async def health():
    """Health check."""
    return {"status": "healthy"}

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """
    Chat with Claude.
    
    - **messages**: List of conversation messages
    - **model**: Claude model to use
    - **max_tokens**: Maximum response length
    - **temperature**: Response randomness (0-1)
    - **system_prompt**: Optional system prompt
    """
    
    # Convert Pydantic models to dicts
    messages = [msg.dict() for msg in request.messages]
    
    # Log request
    logger.info(
        f"Chat request - Model: {request.model}, "
        f"Messages: {len(messages)}, "
        f"Max tokens: {request.max_tokens}"
    )
    
    # Call Claude
    response = await call_claude_with_retry(
        messages=messages,
        model=request.model.value,
        max_tokens=request.max_tokens,
        temperature=request.temperature,
        system_prompt=request.system_prompt
    )
    
    # Extract response
    assistant_message = response.content[0].text
    
    # Log usage
    logger.info(
        f"Response - Input tokens: {response.usage.input_tokens}, "
        f"Output tokens: {response.usage.output_tokens}"
    )
    
    return ChatResponse(
        response=assistant_message,
        model=request.model.value,
        usage=Usage(
            input_tokens=response.usage.input_tokens,
            output_tokens=response.usage.output_tokens,
            total_tokens=response.usage.input_tokens + response.usage.output_tokens
        ),
        finish_reason=response.stop_reason or "end_turn"
    )

@app.post("/simple-chat")
async def simple_chat(message: str, max_tokens: int = 1024):
    """Simple chat endpoint (for testing)."""
    
    try:
        def _call():
            return claude_client.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=max_tokens,
                messages=[{"role": "user", "content": message}]
            )
        
        response = await asyncio.to_thread(_call)
        
        return {
            "response": response.content[0].text,
            "tokens": {
                "input": response.usage.input_tokens,
                "output": response.usage.output_tokens
            }
        }
        
    except Exception as e:
        logger.error(f"Error: {e}")
        raise HTTPException(status_code=500, detail=str(e))

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

This is production-grade:

• ✅ Async/await for performance

• ✅ Request validation

• ✅ Error handling with retries

• ✅ Logging

• ✅ CORS support

• ✅ Auto-generated documentation

Testing the API

Using curl

# Simple chat
curl -X POST "http://localhost:8000/simple-chat?message=Hello&max_tokens=100"

# Full chat
curl -X POST "http://localhost:8000/chat" \
  -H "Content-Type: application/json" \
  -d '{
    "messages": [
      {"role": "user", "content": "What is FastAPI?"}
    ],
    "model": "claude-3-5-sonnet-20241022",
    "max_tokens": 500,
    "temperature": 0.7
  }'

Using Python Requests

import requests

# Simple chat
response = requests.post(
    "http://localhost:8000/simple-chat",
    params={"message": "Hello!", "max_tokens": 100}
)
print(response.json())

# Full chat
response = requests.post(
    "http://localhost:8000/chat",
    json={
        "messages": [
            {"role": "user", "content": "Explain async/await in Python"}
        ],
        "model": "claude-3-5-sonnet-20241022",
        "max_tokens": 1000,
        "temperature": 0.7,
        "system_prompt": "You are a Python expert. Explain concepts clearly."
    }
)
print(response.json())

Interactive Documentation

FastAPI auto-generates docs! Visit:

• Swagger UI: http://localhost:8000/docs

• ReDoc: http://localhost:8000/redoc

You can test all endpoints directly in the browser.

Rate Limiting

Protect your API from abuse.

from fastapi import FastAPI, Request, HTTPException
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

# Initialize limiter
limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post("/chat")
@limiter.limit("10/minute")  # 10 requests per minute per IP
async def chat(request: Request, chat_request: ChatRequest):
    """Rate-limited chat endpoint."""
    # ... implementation
    pass

Install dependencies:

pip install slowapi

I use this in production - prevents one user from burning through API quota.

Environment Configuration

Proper configuration management:

from pydantic_settings import BaseSettings
from functools import lru_cache

class Settings(BaseSettings):
    """Application settings."""
    
    # API keys
    anthropic_api_key: str
    
    # Model defaults
    default_model: str = "claude-3-5-sonnet-20241022"
    default_max_tokens: int = 2048
    default_temperature: float = 1.0
    
    # Rate limiting
    rate_limit_requests: int = 100
    rate_limit_period: str = "minute"
    
    # Server
    host: str = "0.0.0.0"
    port: int = 8000
    debug: bool = False
    
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

@lru_cache()
def get_settings():
    """Get cached settings."""
    return Settings()

# Usage
settings = get_settings()
client = anthropic.Anthropic(api_key=settings.anthropic_api_key)

.env file:

ANTHROPIC_API_KEY=sk-ant-api03-...
DEFAULT_MODEL=claude-3-5-sonnet-20241022
DEFAULT_MAX_TOKENS=2048
DEFAULT_TEMPERATURE=1.0
DEBUG=false

Best Practices

From my production deployments:

1. Always use async/await:

# Good
async def chat(request: ChatRequest):
    response = await call_claude_async(...)
    
# Bad - blocks event loop
def chat(request: ChatRequest):
    response = claude_client.messages.create(...)

2. Validate all inputs:

class ChatRequest(BaseModel):
    message: str = Field(..., min_length=1, max_length=10000)
    max_tokens: int = Field(..., ge=1, le=4096)

3. Implement retry logic:

for attempt in range(max_retries):
    try:
        return await call_api()
    except RateLimitError:
        await asyncio.sleep(2 ** attempt)

4. Log everything:

logger.info(f"Request: {request_id}")
logger.error(f"Error: {error}")

5. Use environment variables:

api_key = os.getenv("ANTHROPIC_API_KEY")

Common Mistakes

Mistakes I made:

1. Blocking calls in async functions ❌

async def chat():
    # Blocks event loop!
    response = claude_client.messages.create(...)

2. No request validation ❌

# Allows any input - dangerous
@app.post("/chat")
async def chat(message: str):
    ...

3. Poor error handling ❌

# Generic errors - hard to debug
except Exception as e:
    return {"error": "Something went wrong"}

4. Hardcoded configuration ❌

# Bad - can't change without code changes
max_tokens = 1024

5. No logging ❌

# Can't debug production issues

What's Next?

You now have a production-ready FastAPI application with Claude integration. In Part 3, we'll add streaming responses for real-time user experience and explore advanced Claude features.

Next: Part 3 - Streaming Responses and Advanced Features

---

Previous: Part 1 - Introduction to LLM APIs and Claude Series Home: LLM API Development 101

This article is part of the LLM API Development 101 series. All examples use Python 3 and FastAPI based on real production applications.