Building an AI Agent Backend with Python, FastAPI, and LangChain (2026 Guide)

Q: How do I add authentication to the FastAPI backend?

Use python-jose for JWT validation and passlib for password hashing. FastAPI's dependency injection makes it clean — create a get_current_user dependency and add it to any protected endpoint. For API key auth (simpler for service-to-service calls), check the Authorization header in a middleware function.

Q: How do I keep conversation history across sessions?

The current implementation takes chat_history as part of each request — the client is responsible for storing it. For server-side persistence, store conversation history in Redis (with TTL for automatic cleanup) or Postgres, keyed by conversation_id. Retrieve the history at the start of each request and pass it to the agent.

Every serious AI engineering job post in 2026 lists Python.

Not instead of TypeScript — alongside it. The pattern that keeps appearing in both job briefs and client projects is this: a Python backend running the heavy AI logic, exposed as an API, with a Next.js frontend consuming it.

LangChain, LangGraph, and most of the serious ML tooling was built Python-first — the documentation is deeper, the community examples are richer, and async FastAPI makes production AI APIs genuinely fast.

In this article you will build a complete AI agent backend in Python using FastAPI and LangChain — with real tool definitions, streaming responses, async endpoints, and a deployment-ready structure. If you have been following this series and built the Node.js agent from Article #1, you will immediately see the parallels — and the differences that make Python the stronger choice for standalone AI backends.

Why Python + FastAPI for AI Backends

Next.js Frontend

useChat hook
React UI
Streaming consumer

POST /api/agentHTTP

Your AI logic lives here Python FastAPI Backend

LangChain Agent
Tool Definitions
OpenAI Client

OpenAI API

Pinecone

External APIs

The decoupled architecture — frontend and AI backend are independent and deployable separately.

Before writing any code, it is worth understanding the architecture decision.

Language best-fit. React and Next.js are best in TypeScript. LangChain, embeddings, and ML pipelines are best in Python. Use each language where it excels instead of fighting against it.
Independent scaling. Your AI backend can scale up during heavy usage without touching the frontend. You can also swap the frontend entirely — mobile app, Slack bot, CLI — without rewriting the AI logic.
Team separation. Frontend engineers own the Next.js layer. AI engineers own the Python backend. Clear boundary, clean API contract.
Python ecosystem. Libraries like numpy, pandas, scikit-learn, sentence-transformers, and huggingface integrate naturally into a Python backend. Doing this in Node.js is significantly harder.

Prerequisites

Python 3.11+
An OpenAI API key
Basic Python familiarity — you do not need to be an expert
Optional: Docker for local development

Project Setup

mkdir ai-agent-backend
cd ai-agent-backend

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# Install dependencies
pip install fastapi uvicorn langchain langchain-openai \
            langchain-community python-dotenv pydantic \
            sse-starlette httpx

Create your project structure:

ai-agent-backend/
├── app/
│   ├── __init__.py
│   ├── main.py           # FastAPI app entry point
│   ├── agent.py          # LangChain agent logic
│   ├── tools.py          # Tool definitions
│   ├── models.py         # Pydantic request/response models
│   └── config.py         # Environment config
├── .env
├── requirements.txt
└── README.md

Create .env:

OPENAI_API_KEY=sk-your-key-here
MODEL_NAME=gpt-4o
ENVIRONMENT=development

Create app/config.py:

from pydantic_settings import BaseSettings
from functools import lru_cache


class Settings(BaseSettings):
    openai_api_key: str
    model_name: str = "gpt-4o"
    environment: str = "development"
    max_iterations: int = 10
    temperature: float = 0

    class Config:
        env_file = ".env"


# lru_cache ensures Settings is only instantiated once
# — the Python equivalent of a singleton
@lru_cache()
def get_settings() -> Settings:
    return Settings()

Defining Your Pydantic Models

Pydantic is one of the best parts of the Python AI stack — it gives you TypeScript-like type safety with automatic validation and serialization. FastAPI is built on top of it.

Create app/models.py:

from pydantic import BaseModel, Field
from typing import Optional


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


class AgentRequest(BaseModel):
    message: str = Field(..., min_length=1, max_length=4000)
    conversation_id: Optional[str] = None
    chat_history: list[Message] = []

    class Config:
        json_schema_extra = {
            "example": {
                "message": "What is 15% of 2450?",
                "conversation_id": "user-123-session-1",
                "chat_history": []
            }
        }


class AgentResponse(BaseModel):
    output: str
    conversation_id: Optional[str] = None
    tool_calls_made: list[str] = []
    success: bool = True


class ErrorResponse(BaseModel):
    error: str
    detail: Optional[str] = None
    success: bool = False

Why this matters: Every request and response is validated automatically by FastAPI using these models. Pass the wrong type — FastAPI returns a clear 422 error before your agent code even runs. This is the Python equivalent of TypeScript + Zod.

Building Your Tool Definitions

Tools in Python LangChain are cleaner than the Node.js version — the @tool decorator handles the schema automatically by reading the function's type hints and docstring.

Create app/tools.py:

from langchain_core.tools import tool
from datetime import datetime
import httpx
import json


@tool
def calculate(expression: str) -> str:
    """
    Evaluates a mathematical expression and returns the result.
    Use this for any arithmetic, percentages, or number calculations.

    Args:
        expression: A valid mathematical expression like '25 * 4 + 10' or '(1500 * 0.15)'

    Returns:
        The calculated result as a string
    """
    try:
        # Safely evaluate math expressions
        # Only allow numbers and math operators — no builtins
        allowed_chars = set('0123456789+-*/()., ')
        if not all(c in allowed_chars for c in expression):
            return f"Error: Expression contains invalid characters: {expression}"

        result = eval(expression, {"__builtins__": {}})
        return f"Result of {expression} = {result}"
    except ZeroDivisionError:
        return "Error: Division by zero"
    except Exception as e:
        return f"Error evaluating '{expression}': {str(e)}"


@tool
def get_current_datetime() -> str:
    """
    Returns the current date and time in ISO format.
    Use this when the user asks about today's date, current time,
    or when you need to reference the current moment.

    Returns:
        Current datetime as ISO string
    """
    now = datetime.now()
    return json.dumps({
        "datetime": now.isoformat(),
        "date": now.strftime("%B %d, %Y"),
        "time": now.strftime("%I:%M %p"),
        "day_of_week": now.strftime("%A")
    })


@tool
async def search_web(query: str) -> str:
    """
    Searches the web for current information.
    Use this when the user asks about recent events, current data,
    or information that may have changed recently.

    Args:
        query: The search query string

    Returns:
        Search results as a formatted string
    """
    # In production: replace with Tavily, Brave, or SerpAPI
    # Example with Tavily:
    # from tavily import TavilyClient
    # client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
    # results = client.search(query)

    # Mock response for this tutorial
    return json.dumps({
        "query": query,
        "results": [
            {
                "title": f"Result for: {query}",
                "snippet": "Replace this with a real search API (Tavily recommended)",
                "url": "https://example.com"
            }
        ],
        "note": "Connect to Tavily API for real results"
    })


@tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """
    Gets the current weather for a specified city.
    Use when the user asks about weather, temperature, or climate conditions.

    Args:
        city: The city name to get weather for
        unit: Temperature unit - 'celsius' or 'fahrenheit'. Default is celsius.

    Returns:
        Current weather data as JSON string
    """
    # In production: replace with OpenWeatherMap or Tomorrow.io API
    # import httpx
    # response = httpx.get(
    #     f"https://api.openweathermap.org/data/2.5/weather",
    #     params={"q": city, "units": "metric", "appid": os.getenv("WEATHER_API_KEY")}
    # )

    return json.dumps({
        "city": city,
        "temperature": 22 if unit == "celsius" else 71,
        "unit": unit,
        "condition": "Partly cloudy",
        "humidity": "65%",
        "note": "Connect to OpenWeatherMap for real data"
    })


# Export all tools as a list
tools = [calculate, get_current_datetime, search_web, get_weather]

The @tool decorator advantage over Node.js: in LangChain Python, the decorator reads your function's type hints and docstring to automatically generate the JSON schema. No manual schema writing like in Node.js — Python's introspection handles it.

The docstring IS the tool description — write it clearly, because the model reads it.

Building the LangChain Agent

Create app/agent.py:

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.callbacks import AsyncIteratorCallbackHandler
from app.tools import tools
from app.config import get_settings
from typing import AsyncIterator
import asyncio


def build_agent_executor() -> AgentExecutor:
    """
    Builds and returns a configured AgentExecutor.
    Called once at startup and reused across requests.
    """
    settings = get_settings()

    # Initialize the LLM
    llm = ChatOpenAI(
        model=settings.model_name,
        temperature=settings.temperature,
        openai_api_key=settings.openai_api_key,
        streaming=True,  # Enable for streaming endpoints
    )

    # System prompt — controls agent personality and behavior
    system_prompt = """You are a helpful AI assistant with access to tools.

You can:
- Perform mathematical calculations
- Get the current date and time
- Search the web for current information
- Get weather data for any city

Guidelines:
- Think step by step before using a tool
- Use tools when you need real data — not for things you already know
- Be concise and accurate in your final response
- If you cannot complete a task with available tools, say so clearly"""

    prompt = ChatPromptTemplate.from_messages([
        ("system", system_prompt),
        MessagesPlaceholder(variable_name="chat_history"),
        ("human", "{input}"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ])

    # Create the agent
    agent = create_tool_calling_agent(llm, tools, prompt)

    # Wrap in executor with safety limits
    executor = AgentExecutor(
        agent=agent,
        tools=tools,
        verbose=settings.environment == "development",
        max_iterations=settings.max_iterations,
        handle_parsing_errors=True,
        return_intermediate_steps=True,  # Lets us track which tools were called
    )

    return executor


def build_chat_history(messages: list) -> list:
    """
    Converts API message objects to LangChain message objects.
    Python equivalent of the TypeScript HumanMessage/AIMessage mapping.
    """
    history = []
    for msg in messages:
        if msg.role == "user":
            history.append(HumanMessage(content=msg.content))
        elif msg.role == "assistant":
            history.append(AIMessage(content=msg.content))
    return history


async def stream_agent_response(
    message: str,
    chat_history: list,
    executor: AgentExecutor
) -> AsyncIterator[str]:
    """
    Streams the agent response token by token.
    Used by the streaming endpoint.
    """
    callback = AsyncIteratorCallbackHandler()

    # Run the agent with the streaming callback
    task = asyncio.create_task(
        executor.ainvoke(
            {
                "input": message,
                "chat_history": build_chat_history(chat_history),
            },
            config={"callbacks": [callback]}
        )
    )

    # Yield tokens as they arrive
    async for token in callback.aiter():
        yield token

    # Ensure the task completes
    await task

Building the FastAPI Application

POST/api/agent

Request ValidatePydantic Agent runs JSON response

For simple queries, testing

POST/api/agent/stream

Request Validate Agent streams Server-Sent Events

For production chat UI (connects to Next.js useChat)

GET/health

Request Status + model info

For deployment monitoring

Three endpoints — one for standard responses, one for streaming, one for health monitoring.

Create app/main.py:

from fastapi import FastAPI, HTTPException, Depends
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from sse_starlette.sse import EventSourceResponse
from app.models import AgentRequest, AgentResponse, ErrorResponse
from app.agent import build_agent_executor, build_chat_history, stream_agent_response
from app.config import get_settings, Settings
from langchain.agents import AgentExecutor
import json
import asyncio
import time


# ─── App initialization ───────────────────────────────────────────

app = FastAPI(
    title="AI Agent Backend",
    description="Production AI agent backend built with FastAPI and LangChain",
    version="1.0.0",
)

# CORS — allow your Next.js frontend to call this API
app.add_middleware(
    CORSMiddleware,
    allow_origins=[
        "http://localhost:3000",       # Next.js dev
        "https://yourapp.vercel.app",  # Next.js production
    ],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


# ─── Agent singleton ──────────────────────────────────────────────

# Build the agent once at startup — not on every request
agent_executor: AgentExecutor | None = None


@app.on_event("startup")
async def startup_event():
    global agent_executor
    print("Initializing AI agent...")
    agent_executor = build_agent_executor()
    print("Agent ready")


def get_agent() -> AgentExecutor:
    if agent_executor is None:
        raise HTTPException(status_code=503, detail="Agent not initialized")
    return agent_executor


# ─── Health check endpoint ────────────────────────────────────────

@app.get("/health")
async def health_check(settings: Settings = Depends(get_settings)):
    return {
        "status": "healthy",
        "model": settings.model_name,
        "environment": settings.environment,
        "timestamp": time.time()
    }


# ─── Standard agent endpoint ──────────────────────────────────────

@app.post(
    "/api/agent",
    response_model=AgentResponse,
    responses={500: {"model": ErrorResponse}}
)
async def run_agent(
    request: AgentRequest,
    executor: AgentExecutor = Depends(get_agent),
    settings: Settings = Depends(get_settings),
):
    """
    Runs the AI agent and returns the complete response.
    Use this for simple queries or when streaming is not needed.
    """
    try:
        chat_history = build_chat_history(request.chat_history)

        result = await executor.ainvoke({
            "input": request.message,
            "chat_history": chat_history,
        })

        # Extract which tools were called
        tool_calls_made = []
        if "intermediate_steps" in result:
            for action, _ in result["intermediate_steps"]:
                tool_calls_made.append(action.tool)

        return AgentResponse(
            output=result["output"],
            conversation_id=request.conversation_id,
            tool_calls_made=tool_calls_made,
            success=True,
        )

    except asyncio.TimeoutError:
        raise HTTPException(
            status_code=504,
            detail="Agent timed out. Please try a simpler query."
        )
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Agent error: {str(e)}"
        )


# ─── Streaming agent endpoint ─────────────────────────────────────

@app.post("/api/agent/stream")
async def stream_agent(
    request: AgentRequest,
    executor: AgentExecutor = Depends(get_agent),
):
    """
    Streams the AI agent response using Server-Sent Events (SSE).
    This is the endpoint your Next.js useChat hook connects to.
    Compatible with the Vercel AI SDK data stream protocol.
    """
    async def event_generator():
        try:
            async for token in stream_agent_response(
                message=request.message,
                chat_history=request.chat_history,
                executor=executor
            ):
                yield {
                    "data": json.dumps({"token": token}),
                    "event": "token",
                }

            # Signal stream completion
            yield {
                "data": json.dumps({"done": True}),
                "event": "done",
            }

        except Exception as e:
            yield {
                "data": json.dumps({"error": str(e)}),
                "event": "error",
            }

    return EventSourceResponse(event_generator())

Connecting to Your Next.js Frontend

This is the bridge that makes the whole architecture work. Your Next.js frontend from Article #4 can call this Python backend directly.

Next.js TypeScript

useChat hook

POST /api/proxy/agent

Next.js Route Handlerproxies to the Python backend

HTTP / SSE stream CORS configured — both run on different domains

Python FastAPI

POST /api/agent/stream

Pydantic validation

LangChain AgentExecutor

OpenAI GPT-4o

SSE token stream back

The frontend never talks to OpenAI directly — all AI logic stays in the Python backend.

In your Next.js project, create a proxy route that forwards requests to the Python backend. This keeps your OpenAI key server-side and lets you add auth middleware cleanly:

// app/api/agent/route.ts (Next.js)
import { NextRequest } from "next/server";

// Your Python backend URL
const PYTHON_BACKEND_URL = process.env.PYTHON_BACKEND_URL || "http://localhost:8000";

export async function POST(req: NextRequest) {
  const body = await req.json();

  // Forward the request to your Python FastAPI backend
  const response = await fetch(`${PYTHON_BACKEND_URL}/api/agent/stream`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      // Add auth headers here if your backend requires them
    },
    body: JSON.stringify(body),
  });

  // Stream the SSE response back to the browser
  return new Response(response.body, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      "Connection": "keep-alive",
    },
  });
}

Add the backend URL to your Next.js .env.local:

PYTHON_BACKEND_URL=http://localhost:8000        # Development
# PYTHON_BACKEND_URL=https://your-api.railway.app  # Production

Running the Backend Locally

# Start the FastAPI server with hot reload
uvicorn app.main:app --reload --port 8000

Your API is now running at http://localhost:8000. FastAPI automatically generates interactive docs at http://localhost:8000/docs — one of the best features of FastAPI and something Node.js/Express does not give you out of the box.

Test your agent endpoint:

# Test the standard endpoint
curl -X POST http://localhost:8000/api/agent \
  -H "Content-Type: application/json" \
  -d '{
    "message": "What is 15% of 2450? And what day of the week is it today?",
    "chat_history": []
  }'

Expected response:

{
  "output": "15% of 2450 is 367.5. Today is Sunday.",
  "conversation_id": null,
  "tool_calls_made": ["calculate", "get_current_datetime"],
  "success": true
}

Adding Rate Limiting

Never deploy an AI endpoint without rate limiting. In Python, slowapi is the FastAPI equivalent of Upstash Rate Limit:

pip install slowapi

# app/main.py — add rate limiting
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

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

# Apply to your endpoint
@app.post("/api/agent", response_model=AgentResponse)
@limiter.limit("20/minute")  # 20 requests per minute per IP
async def run_agent(request: Request, body: AgentRequest, ...):
    ...

Deployment on Railway

Railway is the cleanest option for deploying Python FastAPI backends in 2026 — Docker support, environment variables, and automatic HTTPS out of the box.

Create a Dockerfile:

FROM python:3.11-slim

WORKDIR /app

# Install dependencies first (Docker layer caching)
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application code
COPY . .

# Expose the port FastAPI runs on
EXPOSE 8000

# Production command — no --reload in production
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Create requirements.txt:

fastapi==0.111.0
uvicorn==0.30.1
langchain==0.2.6
langchain-openai==0.1.14
langchain-community==0.2.6
python-dotenv==1.0.1
pydantic==2.7.4
pydantic-settings==2.3.4
sse-starlette==2.1.0
slowapi==0.1.9
httpx==0.27.0

Deploy to Railway:

# Install Railway CLI
npm install -g @railway/cli

# Login and deploy
railway login
railway init
railway up

Set your environment variables in the Railway dashboard:

OPENAI_API_KEY=sk-...
MODEL_NAME=gpt-4o
ENVIRONMENT=production

Railway gives you a URL like https://ai-agent-backend-production.up.railway.app — add this to your Next.js PYTHON_BACKEND_URL environment variable on Vercel.

Python vs Node.js LangChain — Key Differences

If you built the agent from Article #1, here is what changes in Python:

	Node.js (Article #1)	Python (This article)
Tool definition	Manual JSON schema + function	`@tool` decorator reads type hints
Async	Native async/await	`async def` + `await`
Type safety	TypeScript + Zod	Python type hints + Pydantic
Agent creation	`createReactAgent`	`create_tool_calling_agent`
Documentation depth	Good	Significantly deeper
ML library support	Limited	Full ecosystem
API framework	Express / Next.js routes	FastAPI (auto docs!)

The logic is identical. The Python version is more concise thanks to decorators and type hints doing more work automatically.

What to Build Next

You now have a production-ready Python AI agent backend. The natural progression:

Add RAG to the Python backend — connect LangChain's Python retriever to your Pinecone index from Article #2 so the agent can search your documents.
Switch to LangGraph — upgrade from a linear agent to a stateful graph for complex multi-step tasks.
Add authentication — protect your FastAPI endpoints with JWT using python-jose before going to production.
Connect the streaming frontend — wire the SSE stream from this backend to the useChat hook from Article #4.

The full source code is on GitHub.

Frequently asked questions

Why FastAPI over Flask or Django for AI backends?

FastAPI is built for async Python from the ground up — critical for AI backends where you are waiting on LLM API calls constantly. Flask is synchronous by default and blocks during long AI calls. Django is a full web framework with too much overhead for a focused API service. FastAPI also auto-generates OpenAPI docs, has native Pydantic integration, and is significantly faster than both alternatives in benchmarks.

Do I need to know Python well to follow this?

Basic Python familiarity is enough. If you know TypeScript, Python will feel familiar — the syntax is different but the concepts (async/await, type hints, classes) map directly. The main adjustment is whitespace-based indentation and the @decorator pattern.

Can this backend handle multiple users simultaneously?

Yes. FastAPI with uvicorn runs multiple async workers, so concurrent requests are handled without blocking. For high traffic, run multiple uvicorn workers: uvicorn app.main:app --workers 4. For very high load, add a Redis-based queue (Celery or ARQ) to handle agent tasks asynchronously.

How do I add authentication to the FastAPI backend?

Use python-jose for JWT validation and passlib for password hashing. FastAPI's dependency injection makes it clean — create a get_current_user dependency and add it to any protected endpoint. For API key auth (simpler for service-to-service calls), check the Authorization header in a middleware function.

Is Railway better than Fly.io for Python AI backends?

Railway is simpler to set up and better for teams that want managed infrastructure. Fly.io gives more control — you can choose specific regions, configure persistent volumes, and run WebSocket servers cleanly. For a straightforward FastAPI backend, Railway is the faster path to production. For WebSocket-heavy or globally distributed AI services, Fly.io is worth the extra setup.

How do I keep conversation history across sessions?

The current implementation takes chat_history as part of each request — the client is responsible for storing it. For server-side persistence, store conversation history in Redis (with TTL for automatic cleanup) or Postgres, keyed by conversation_id. Retrieve the history at the start of each request and pass it to the agent.

Where to go next

You've now built the same agent twice — once in Node.js, once in Python. These pick up from here:

Python and TypeScript are not competing choices here — they are the right tool for two different halves of the same product. Keep the AI logic in FastAPI where LangChain is native and the ecosystem runs deepest, keep the UI in Next.js where React is native, and let a clean HTTP boundary hold the whole thing together.

Building an AI Agent Backend with Python, FastAPI, and LangChain