DEV Community: zpillsbury

Making a Todo API with FastAPI and MongoDB

zpillsbury — Mon, 13 Jan 2025 15:41:48 +0000

Introduction

This guide walks you through building the most up to date API with FastAPI application with MongoDB. You'll create a Todo API that demonstrates best practices for structuring a FastAPI project, implementing CRUD operations, handling data validation, and organizing code for scalability.

By following this tutorial, you'll learn how to:

Set up a FastAPI project with proper tooling and dependencies
Design and implement data models using Pydantic
Create CRUD endpoints with async functions
Connect to MongoDB using Motor
Implement proper error handling and status codes
Structure your project using routers and utilities
Add logging and monitoring
Configure CORS for frontend integration

Whether you're new to FastAPI or looking to improve your existing skills, this guide provides a practical, hands-on approach to building APIs. The Todo application serves as a foundation that you can build upon for your own projects.

⭐️ The complete source code referenced in this guide is available on GitHub: https://github.com/zpillsbury/to-do
(There will be extra, code, I'll be releasing how to add security and other features next)

Setup

Install uv

Why Use uv?

While not strictly required, the uv package manager offers significantly faster installation speeds compared to traditional tools like pip. It's recommended to integrate uv into your projects moving forward.

Create basic project setup

$ uv add --dev black mypy ruff

This command installs:

black: Code formatter
mypy: Static type checker
ruff: Linter
These tools help maintain code quality, consistency, and readability.

$ uv add --dev black mypy ruff

Add tool settings in pyproject.toml for black, ruff, and mypy:

pyproject.toml

tool.black]
line-length = 88

[tool.ruff]
lint.select = ["E", "F", "I"]
lint.fixable = ["ALL"]
exclude = [".git", ".mypy_cache", ".ruff_cache"]
line-length = 88

[tool.mypy]
plugins = ["pydantic.mypy"]

disallow_any_generics = true
disallow_subclassing_any = true
disallow_untyped_calls = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
no_implicit_optional = true
warn_redundant_casts = true
warn_unused_ignores = true
warn_return_any = true
strict_equality = true
disallow_untyped_decorators = false
ignore_missing_imports = true
implicit_reexport = true

[tool.pydantic-mypy]
init_forbid_extra = true
init_typed = true
warn_required_dynamic_aliases = true
warn_untyped_fields = true

Synchronize the new configurations:

$ uv sync

Installing FastAPI:

$ uv add "fastapi[standard]"

Create an .env file to store your secrets, make sure to put it in your .gitignore. This makes it where any file put inside there won’t be pushed up to github.
You will need to get your mongo uri from your MongoDB.

.env

MONGO_URI=mongodb://root:mySecureDbPassword1@localhost:27017/

Delete the hello.py and create main.py.

Use the Pydantic BaseSettings to load the mongo_uri from your .env folder, we will be using SecretSTR so the information isn’t visible in logs or tracebacks.

main.py

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    mongo_uri: SecretSTR

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")


settings = Settings()

Next we will be making a function get_db to connect to MongoDB using Motor which is an asynchronous drive. In the function we will be calling settings to get the mongo uri.

The ["to-do"] will be the name of your database, you don’t need to create it in mongo, once you create your first todo in your FastAPI it will generate the database and collections you use.

We will be using asynchronous functions to make this API as fast as possible. What makes async so much faster is that functions will be able to be process other tasks while waiting for tasks like a database query or a call to an external api.

If you need a more in depth explanation of the benefits of async over sync click here: https://www.youtube.com/watch?v=8aGhZQkoFbQ

Add app = FASTAPI to be able to call FastAPI.

main.py

from typing import Any

from fastapi import FastAPI
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase

def get_db() -> AsyncIOMotorDatabase[Any]:
    """
    Get MongoDB
    """
    return AsyncIOMotorClient(settings.mongo_uri.get_secret_value())["to-do"]


app = FastAPI()

You should be able to call fastAPI now using the command FastAPI dev main.py.
For now it should be an empty page since we haven’t added any endpoints.

BaseModels

The first thing you need to do to make a good API is to set up your pydantic base models. At this phase you will set up the models that you will be building your API with. This is important to do before you start making your API, since these will be your guidelines when setting up your Create, Read, Update and Delete for your todos.

Read Model

Starting with Todo , this will include all information you want to retrieve from MongoDb. We will use this for our Get functions.

Since we are making a todo app with mongoDB, it will need the id . This is how we can search for a specific record from the mongo database.

We need a title this is what the task is going to be named.

We can add a description for tasks that might needs an explanation.

Finally we need completed this is where we will check off what we have done.

main.py

from pydantic import BaseModel

class Todo(BaseModel):
    id: str
    title: str
    description: str
    completed: bool

Now we have the base model to build the rest of our models off of.

Create Model

The next base model we will make is CreateTodo. This will be all the information the user will enter theirselves.

Since Mongo supplies the id it won’t be needed here.

We will add in the title as a mandatory field, without this the todo can’t be made.

Next we will add description in but we don’t always want to add a description to everything. If you add “take out trash”, you don’t need a description.

This is where Optional comes in. This does exactly what it sounds like making it optional for you to either add a field or not.

Next the completed will be inserted in as automatically False by our function. So the user won’t have to do it themselves every time. This will be shown how to this later on.

Id Model

Next we will need to know the id of the todo we just created. So we will make a TodoId base model .

We will be using this to return the id whenever the user creates a new todo. This is helpful so the user doesn’t have to do a find all or go to the database to find the id of the new entry.

main.py

from typing import Optional

class CreateTodo(BaseModel):
    title: str
    description: Optional[str]


class TodoId(BaseModel):
    id: str

Update Model

The next base model is UpdateTodo. This will be the fields we want our users to be able to update.

If they typed in the wrong title or description they can go update it to change it. More importantly to be able to change completed that way the Todo actually works.

The main problem is if you want to change one of these you don’t want to have to change all of them. This is where Optional comes in handy again, allowing fields in base models to either be changed or not.

Result Model

Now that we have the update model, when we actually update we will want to know if it was a successful update or not. This is where we will create SuccessResult , we will use this to return if the action was successful or not.

Since the Delete end point just needs to know if its succeeded or failed we will be using the same SuccessResult base model for the return.

main.py

class UpdateTodo(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

class SuccessResult(BaseModel):
    success: bool

All Models

main.py

from pydantic import BaseModel
from typing import Optional

class Todo(BaseModel):
    id: str
    title: str
    description: str
    completed: bool

class CreateTodo(BaseModel):
    title: str
    description: Optional[str]

class UpdateTodo(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

class TodoId(BaseModel):
    id: str

class SuccessResult(BaseModel):
    success: bool

CRUD

now that we have the project set up we can move to make the industry standard Create, Read, Update, and Delete(CRUD) functions based off the models.

Create

To create a new Todo, start by defining an endpoint with @app.post("/todo"). This specifies the route for FastAPI and marks it as a POST endpoint, which is used for creating resources.

Next, create an asynchronous function create_to_do that takes new_todo: CreateTodo as a parameter. This parameter is a Pydantic model representing the input data for creating a Todo. The function will return a TodoId model containing the ID of the newly created Todo.

To prepare the data for insertion, use the .model_dump() method to convert the new_todo model into a dictionary. MongoDB requires documents in dictionary format, and you can add default fields, like {"completed": False}, using the dictionary merge operator (|).

Insert the data into the todos collection using the db.todos.insert_one(data) method. If the collection doesn’t already exist, MongoDB will create it automatically. The await keyword ensures that the function can continue handling other tasks while waiting for the database operation to complete.

Finally, retrieve the inserted_id from MongoDB, convert it to a string (as JSON only supports strings and numbers), and return it.

you will notice I broke down each part into separate variables, particularly when you are working with other people its important to make it as readable as possible. This will also help you in the future to be able to find bugs easier.

main.py

@app.post("/todo")
async def create_to_do(new_todo: CreateTodo) -> TodoId:
    """
    Create a new todo
    """
    data = new_todo.model_dump() | {"completed": False}

    result = await db.todos.insert_one(data)

    id = str(result.inserted_id)

    return TodoID(id=id)

go to fastapi dev main.py to try it out, you should now be able to create your first todo. Once you create it, you can check in MongoDB if it was successful. The next step is going to be making it where we can get that information without having to check Mongo ourselves.

Read

Get All Todos

The get_all_todos endpoint retrieves all todo items from the database. you will notice you will be returning a list of Todos.

you will make a variable with an empty list, such as result . We will be using it for the function below, we will be looping for each data entry(todo) in todos then appending it to result .

Once the loop is done you will return result which now is a list of Todos.

main.py

@app.get("/todo")
async def get_all_todos() -> list[Todo]:
    """
    Get all todos
    """
    result = []
    async for todo in db.todos.find():
        result.append(
            Todo(
                id=str(todo.get["_id"]),
                title=todo.get["title"],
                description=todo.get["description"],
                completed=todo.get["completed"],
            )
        )

    return result

Try it out, you should be able to see your todo you made.try make another todo you will be able to get both of them. If you want to only get one you can get todo by id. Which will be covered next.

Get Todo by ID

Since I've explained the process in depth on the last function, I will only hit the main points and new things moving forward.

The get_one_todo endpoint gets a single todo item by its ID.

The provided todo_id string is converted into a MongoDB ObjectId.

We then use find_one method searches for the document with the matching _id.

It returns the Todo if the id matches.

If no document is found, an HTTP 404 error is raised.

Main.py

@app.get("/todo/{todo_id}")
async def get_one_todo(todo_id: str) -> Todo:
    """
    Get todo by ID
    """
    todo_object_id = ObjectId(todo_id)

    todo = await db.todos.find_one({"_id": todo_object_id})
    if not todo:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")

    return Todo(
        id=str(todo.get["_id"]),
        title=todo.get["title"],
        description=todo.get["description"],
        completed=todo.get["completed"],
    )

Try it out, grab one of your ids and see if you can find your todo. If it works great! you will see we can do multiple things with an id .

Update

Update Todo

The update_one_todo endpoint updates the fields of an existing todo item.

The UpdateTodo model allows users to provide only the fields they want to change.

You will use update_one to apply changes.

If successful it shows success=True

If no matching document is found, an HTTP 404 error is raised.

main.py

@app.patch("/todo/{todo_id}")
async def update_one_todo(todo_id: str, todo_update: UpdateTodo) -> SuccessResult:
    """
    Update todo by ID
    """
    todo_object_id = ObjectId(todo_id)

    update_data = todo_update.model_dump(exclude_unset=True)

    update_result = await db.todos.update_one(
        {"_id": todo_object_id}, {"$set": update_data}
    )

    if update_result.matched_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

Try it out, if you see a success and you want to verify it worked. Take the id and put it in the get_one_todo. You should see that the information is updated on your todo.

Delete

Delete Todo

The delete_one_todo endpoint removes a todo item from the database.

MongoDB's delete_one method removes the matching document.

If successful it shows success=True

If no document is deleted, an HTTP 404 error is raised.

main.py

@app.delete("/todo/{todo_id}")
async def delete_one_todo(todo_id: str) -> SuccessResult:
    """
    Delete todo by ID
    """
    todo_object_id = ObjectId(todo_id)

    delete_result = await db.todos.delete_one({"_id": todo_object_id})

    if delete_result.deleted_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

Try it out, if its a success you can take the id to the get_one_todo to verify. It should say Todo not found.

Utilities

Logging

Logging allows for troubleshooting by monitoring your application in real time. This is important to be able to fix and bugs when they occur. We will just use the basic python logging module.

logging configs:

main.py

import logging
import sys

logging.basicConfig(
    stream=sys.stdout,
    level=logging.INFO,
    format="[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s",  # noqa: E501
    datefmt="%d/%b/%Y %H:%M:%S",
)
logger = logging.getLogger("todo")

code breakdown:

stream=sys.stdout :

This makes it where it outputs to you console

level=logging.INFO :

This sets the minimum log level

it goes Debug → Info → Warning → Error → Critical

format="..." :

[%(asctime)s] : The timestamp of the log

%(levelname)s : The severity level

[%(name)s.%(funcName)s:%(lineno)d]

%(name)s : Name of the Logger

%(funcName)s : Name of the function where the logger was created

%(lineno)d : The line where the log was created

%(message)s : The log message

datefmt="%d/%b/%Y %H:%M:%S”

The format the timestamp will be in.

Middleware

Middleware in FastAPI allows you to process requests and responses globally.
CORS Middleware

main.py

from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
    CORSMiddleware,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    allow_origins=[
        "http://localhost:3000",
        "http://localhost:8000",
    ],
)

Here’s a concise breakdown of the CORS middleware setup:

allow_credentials=True
Allows cookies, auth headers, and other credentials to be sent with requests .
allow_methods=["*"]
Permits all HTTP methods (e.g., GET, POST, DELETE) to be used by the frontend .
allow_headers=["*"]
Accepts all headers, providing flexibility for metadata or labels in requests and responses .
allow_origins=
Grants access to specific addresses, such as the backend http://localhost:8000 and the frontend http://localhost:3000 for security and proper integration.

main.py

@app.middleware("http")
async def process_time_log_middleware(request: Request, call_next: F) -> Response:
    """
    Add API process time in response headers and log calls
    """
    start_time = time.time()
    response: Response = await call_next(request)
    process_time = str(round(time.time() - start_time, 3))
    response.headers["X-Process-Time"] = process_time

    logger.info(
        "Method=%s Path=%s StatusCode=%s ProcessTime=%s",
        request.method,
        request.url.path,
        response.status_code,
        process_time,
    )

    return response

This middleware logs request details and tracks processing time for each API call

Captures the current time when a request is received to measure processing duration.
Passes the request to the next handler using call_next and waits for the response.
Determines the time taken for the request by subtracting the start time from the end time and rounding it to three decimal places.
Attaches the processing time as X-Process-Time in the response headers.
Logs the HTTP method, request path, status code, and processing time for better monitoring and debugging.

Organization

At this point your document should be looking like this:

Main.py

from typing import Any, Optional

from bson import ObjectId
from fastapi import FastAPI, HTTPException, status
from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase
from pydantic import BaseModel, SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    mongo_uri: SecretStr
    token_url: str

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

settings = Settings()

def get_db() -> AsyncIOMotorDatabase[Any]:
    """
    Get MongoDB
    """
    return AsyncIOMotorClient(settings.mongo_uri.get_secret_value())["to-do"]

db = get_db()

app = FastAPI(docs_url="/")

class Todo(BaseModel):
    id: str
    title: str
    description: str
    completed: bool

class CreateTodo(BaseModel):
    title: str
    description: str

class UpdateTodo(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

class TodoId(BaseModel):
    id: str

class SuccessResult(BaseModel):
    success: bool

@app.post("/todo")
async def create_to_do(new_todo: CreateTodo) -> TodoId:
    """
    Create a new todo
    """
    data = new_todo.model_dump() | {"completed": False}

    result = await db.todos.insert_one(data)

    id = str(result.inserted_id)

    return TodoId(id=id)

@app.get("/todo")
async def get_all_todos() -> list[Todo]:
    """
    Get all todos
    """
    result = []
    async for todo in db.todos.find():
        result.append(
            Todo(
                id=str(todo.get("_id")),
                title=todo.get("title"),
                description=todo.get("description"),
                completed=todo.get("completed"),
            )
        )

    return result

@app.get("/todo/{todo_id}")
async def get_one_todo(todo_id: str) -> Todo:
    """
    Get todo by ID
    """
    todo_object_id = ObjectId(todo_id)

    todo = await db.todos.find_one({"_id": todo_object_id})
    if not todo:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")

    return Todo(
        id=str(todo.get("_id")),
        title=todo.get("title"),
        description=todo.get("description"),
        completed=todo.get("completed"),
    )

@app.patch("/todo/{todo_id}")
async def update_one_todo(todo_id: str, todo_update: UpdateTodo) -> SuccessResult:
    """
    Update todo by ID
    """
    todo_object_id = ObjectId(todo_id)

    update_data = todo_update.model_dump(exclude_unset=True)

    update_result = await db.todos.update_one(
        {"_id": todo_object_id}, {"$set": update_data}
    )

    if update_result.matched_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

@app.delete("/todo/{todo_id}")
async def delete_one_todo(todo_id: str) -> SuccessResult:
    """
    Delete todo by ID
    """
    todo_object_id = ObjectId(todo_id)

    delete_result = await db.todos.delete_one({"_id": todo_object_id})

    if delete_result.deleted_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

As you can see its getting a bit long and if you wanted to add more end points, security, dockers and so on its only going to get longer and longer. This is where organization is important. The main tools we will be using for this is routers and folders.

Folder Setup

First we will make an app folder to hold everything.

Next we will move our main.py into the app folder.

We will be adding __init__.py to any folders that have files.

Next we will make 2 more folders routers and utils both are inside our app folder.

Utils

Next we will be adding config.py , settings.py, log.py and __init__.py to the utils folder.

Settings.py

You will go into main.py and be moving the basesettings to settings.py

You can copy and paste all the imports as well and use shift + command + p and use ruff to get rid of all unused imports on all files moving forward.

settings.py

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    mongo_uri: SecretStr
    token_url: str

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")

settings = Settings()

Config.py

Make sure to import settings from .settings

config.py

from typing import Any

from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorDatabase

from .settings import settings

def get_db() -> AsyncIOMotorDatabase[Any]:
    """
    Get MongoDB
    """
    return AsyncIOMotorClient(settings.mongo_uri.get_secret_value())["to-do"]

db = get_db()

log.py

log.py

import logging
import sys

logging.basicConfig(
    stream=sys.stdout,
    level=logging.INFO,
    format="[%(asctime)s] %(levelname)s [%(name)s.%(funcName)s:%(lineno)d] %(message)s",  # noqa: E501
    datefmt="%d/%b/%Y %H:%M:%S",
)
logger = logging.getLogger("todo")

Routers

Next we will be adding todos folder in routers

Inside todos folder we will add todo.py , models.py , and __**init**__.py .

Models.py

We now now go into main.py and cut out all the models and and paste them into models.py .

models.py should now look like:

models.py

from pydantic import BaseModel
from typing import Optional

class Todo(BaseModel):
    id: str
    title: str
    description: str
    completed: bool

class CreateTodo(BaseModel):
    title: str
    description: Optional[str]

class UpdateTodo(BaseModel):
    title: Optional[str]
    description: Optional[str]
    completed: Optional[bool]

class TodoId(BaseModel):
    id: str

class SuccessResult(BaseModel):
    success: bool

Todo.py

We will now be moving all the CRUD functions over too todo.py .

You will notice some problems with you code now. Your models you use in the function are no longer there and @app no longer works.

To fix that add from .models import CreateTodo, SuccessResult, Todo, TodoId, UpdateTodo

Add from app.utils.conif import db

We will then add in our router , to do that we just add in router = APIRouter() . We can use a function of prefix= to make it where we don't have to type the path for each function in the router, you will still have to add {todo_id} for functions that need it.

As well as add in tags= which sorts out your APIs into categories. This will be more important when you have more routers.

We then replace @app with @router .

todo.py should look like this:

todo.py

from bson import ObjectId
from fastapi import APIRouter, HTTPException, status

from app.utils.config import db

from .models import CreateTodo, SuccessResult, Todo, TodoId, UpdateTodo

@router.post("/todo")
async def create_to_do(new_todo: CreateTodo) -> TodoId:
    """
    Create a new todo
    """
    data = new_todo.model_dump() | {"completed": False}

    result = await db.todos.insert_one(data)

    id = str(result.inserted_id)

    return TodoId(id=id)

@router.get("/todo")
async def get_all_todos() -> list[Todo]:
    """
    Get all todos
    """
    result = []
    async for todo in db.todos.find():
        result.append(
            Todo(
                id=str(todo.get("_id")),
                title=todo.get("title"),
                description=todo.get("description"),
                completed=todo.get("completed"),
            )
        )

    return result

@router.get("/todo/{todo_id}")
async def get_one_todo(todo_id: str) -> Todo:
    """
    Get todo by ID
    """
    todo_object_id = ObjectId(todo_id)

    todo = await db.todos.find_one({"_id": todo_object_id})
    if not todo:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found")

    return Todo(
        id=str(todo.get("_id")),
        title=todo.get("title"),
        description=todo.get("description"),
        completed=todo.get("completed"),
    )

@router.patch("/todo/{todo_id}")
async def update_one_todo(todo_id: str, todo_update: UpdateTodo) -> SuccessResult:
    """
    Update todo by ID
    """
    todo_object_id = ObjectId(todo_id)

    update_data = todo_update.model_dump(exclude_unset=True)

    update_result = await db.todos.update_one(
        {"_id": todo_object_id}, {"$set": update_data}
    )

    if update_result.matched_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

@router.delete("/todo/{todo_id}")
async def delete_one_todo(todo_id: str) -> SuccessResult:
    """
    Delete todo by ID
    """
    todo_object_id = ObjectId(todo_id)

    delete_result = await db.todos.delete_one({"_id": todo_object_id})

    if delete_result.deleted_count == 0:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND, detail="Todo not found."
        )

    return SuccessResult(success=True)

Main.py

At this point you main.py should only have app = and middleware.

From todos you will import the whole todo file

Then do app.include_router(todo.router)

import time
from typing import Any, Callable, TypeVar

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware


from app.routers.todos import todo

app = FastAPI(    
        title="Todo",
    description="What do I need to do?",
    version="1.0.0",
    docs_url="/",
)

app.add_middleware(
    CORSMiddleware,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
    allow_origins=[
        "http://localhost:3000",
        "http://localhost:8000",
    ],
)

@app.middleware("http")
async def process_time_log_middleware(request: Request, call_next: F) -> Response:
    """
    Add API process time in response headers and log calls
    """
    start_time = time.time()
    response: Response = await call_next(request)
    process_time = str(round(time.time() - start_time, 3))
    response.headers["X-Process-Time"] = process_time

    logger.info(
        "Method=%s Path=%s StatusCode=%s ProcessTime=%s",
        request.method,
        request.url.path,
        response.status_code,
        process_time,
    )

    return response

app.include_router(todo.router)

Conclusion

You've now built a complete FastAPI application that follows modern Python best practices. The Todo API demonstrates several important concepts:

Organization using a modular approach with routers and utilities makes the code maintainable, readable and scalable
Data Validation with Pydantic models ensure type safety and data consistency
Database Integration with MongoDB
Error Handling with status codes and error messages improve API usability
Monitoring with logging middleware which helps track API performance and debug issues
Frontend Integration with CORS configuration e

In the future I will expand on this project with:

Adding user authentication
Creating additional endpoints
Adding test coverage
Containerizing with Docker
Setting up CI/CD pipelines

Remember to keep your dependencies updated and refer to FastAPI's official documentation for the latest best practices and features.

How to make an AI agent with OpenAI, Langgraph, and MongoDB 💡✨

zpillsbury — Fri, 20 Dec 2024 17:12:16 +0000

Large language models (LLMs) such as ChatGPT, Gemini, and Claude have become an essential tool for businesses. Every company is eager to develop its own customized AI tailored to its specific needs or customers. In this exploration, we will focus on creating a personalized assistant that has function calling, stores messages in a database for multi-session conversations, and enables your assistant to perform web searches and summarize its findings. To make it better organized and to allow the addition of new features in the future, we will utilize Langchain, Langgraph and LangSmith, which simplifies the process and enhance functionality. Langchain simplifies the process of streaming, tool calling and lets you use multiple different llms. Langgraph is an organization tool that helps decide which tool needs to be used and lets the agent choose its own path. Langsmith is an observation tool, this will allow you to see the whole process from when you ask the llm a question to when you get your final answer.

⭐️ The complete source code referenced in this guide is available on GitHub https://github.com/zpillsbury/ai-agent

Setup

Add environment variables to .env, you will need your openai_key, tavily_key, and mongo_uri.

📝 .env

OPENAPI_KEY=OPENAI_KEY=sk-proj-XXXXXX
TAVILY_API_KEY=tvly-XXXXXXXXXXXXXXXXXXXXXXXX
MONGO_URI=mongodb+srvXXXXXXXXXXXXXXXXXXXXXXXXXXX

LANGCHAIN_TRACING_V2=true
LANGCHAIN_ENDPOINT=https://api.smith.langchain.com
LANGCHAIN_API_KEY=lsv2_pt_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx_xxxxxxxxxx
LANGCHAIN_PROJECT=project-name

Add langsmith, OpenAI key, Tavily key, and Mongo uri to settings.

📝 app/utilities/settings.py

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    openai_key: SecretStr
    tavily_api_key: str
    mongo_uri: SecretStr


    # LangSmith
    langchain_tracing_v2: bool = True
    langchain_endpoint: str = "https://api.smith.langchain.com"
    langchain_project: str = "ai-agent"
    langchain_api_key: str

    model_config = SettingsConfigDict(env_file=".env", extra="ignore")


settings = Settings()

Setup the OpenAI client using our secret key from the settings.

📝 app/main.py

from langchain_openai import ChatOpenAI

from .utilities.settings import settings

llm = ChatOpenAI(
    openai_api_key=settings.openai_key,
    model_name="gpt-4o-mini",
    max_retries=2,
)

Langgraph

Set up the state so data can be passed and used between nodes. We are adding messages to our chatbot's state using a LLM to generate messages. The add_messages function from langgraph makes it so any messages returned from a node will be appended to the existing messages state.

📝 app/main.py

from typing import Annotated, TypedDict

from langchain_core.messages import BaseMessage
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from langgraph.graph.state import CompiledStateGraph


class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

Nodes are points on graphs and in langgraph nodes are represented with functions. Create a new function chatbot that calls OpenAI using llm.ainvoke sending it the current state of stored messages. OpenAI will return a new AI message. Then return the new state update which includes the AI message. This will be appended to the existing messages thanks to the add_messages helper function.

In langgraph they have synchronous and asynchronous versions of methods. If you need to add an asynchronous version just an a in front of the method. In this case we are using the async version ainvoke instead of the sync version invoke.

📝 app/main.py

async def chatbot(state: State) -> State:
    """
    Chatbot
    """
    response_message = await llm.ainvoke(state["messages"])

    return {"messages": [response_message]}

Add the function with add_node to the graph. You will connect the nodes together with edges, when looking at a picture of a graph these are the lines between each node. The first edge always needs to begin with START. Connect the start of the graph (START) to the chatbot node with add_edge. Then compile the graph with graph_builder.compile, the graph needed to be compiled before you can use it.

📝 app/main.py

async def get_graph() -> CompiledStateGraph:
    """
    Get the graph
    """
    graph_builder = StateGraph(State)

    graph_builder.add_node("chatbot", chatbot)

    graph_builder.add_edge(START, "chatbot")
    graph_builder.add_edge("chatbot", END)

    graph = graph_builder.compile()

    return graph

The Graph currently looks like this:

Create a new async function run_graph using graph.astream (stream if sync) to call the compiled graph, this opens the “chatbot” node which has the value of the function chatbot. It will then insert the system prompt first then the question that the user inputed as the current state. Sending it to openAI where it will return the AI message and your function will append it to the state. We can access the current state in the event variable on each loop of the graph stream. We then extract the value from the event loop. From the value we will extract the content of the last message and it will be returned to the user.

Adding a system prompt is supplying it with information, the current date makes the AI assistant search information more accurate. The LLM is only trained up till a certain time period, making it where it will be more likely to find old article instead of new ones.

📝 app/main.py

from datetime import datetime, timezone
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage

now = datetime.now(timezone.utc)

system_prompt = f"""
You are an AI assistant helping an user.

Current Date: {now}
"""

async def run_graph(question: str) -> None:
    """
    Run the graph
    """
    async for event in graph.astream(
        {
            "messages": [
                SystemMessage(content=system_prompt),
                HumanMessage(content=question),
            ]
        }
    ):
        for value in event.values():
            print(value["messages"][-1].content)

    return None

Create a new async function main this will be adding user input and adding a while statement which will stop the chat in the terminal if the user types (quit, exit, or q). It will use the function run_graph using the user input and go through the process explained previously.

📝 app/main.py

async def main() -> None:
    """
    AI Agent
    """
    while True:
        question = input("q: ")
        if question.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break

        await run_graph(question)

    return None


if __name__ == "__main__":
    anyio.run(main)

Tool Calling

you will be using your using tavily_api_key from .env and .settings here. Using TavilySearchResults lets your ai assistant look things up on the web. We will put them inside tools which will allow you to easily add more tools later. Then set llm_with_tools = llm.bind_tools which gives your ai assistant your list of tools to use. Next change any following place that had llm to llm_with_tools.

📝 app/main.py

from langchain_community.tools.tavily_search import TavilySearchResults


web_search = TavilySearchResults(max_results=2)
tools = [web_search]

llm = ChatOpenAI(
    openai_api_key=settings.openai_key,
    model_name="gpt-4o-mini",
    max_retries=2,
)

llm_with_tools = llm.bind_tools(tools)

async def chatbot(state: State) -> State:
    """
    Chatbot
    """
    response_message = await llm_with_tools.ainvoke(state["messages"])

    return {"messages": [response_message]}

Next add a tool_node, this adds a node that runs the tool calling for AI messaging. Next add add_conditional_edge this will allow the AI assistant to have an edge that can lead to multiple different nodes depending on what it needs. It will first check if it needs tool calling if not it will send it to the END node.

📝 app/main.py

async def get_graph() -> CompiledStateGraph:
    """
    Get the graph
    """
    graph_builder = StateGraph(State)

    graph_builder.add_node("chatbot", chatbot)

    tool_node = ToolNode(tools=tools)
    graph_builder.add_node("tools", tool_node)

    graph_builder.add_edge(START, "chatbot")

    graph_builder.add_conditional_edges("chatbot", tools_condition)
    graph_builder.add_edge("tools", "chatbot")

    graph = graph_builder.compile()

    return graph

Storing Chat Data

You will be using your mongo_uri from .env and .settings. Now add your async_mongodb_client to the top. Then add a check point to save the memory to mongoDB. You will add this inside of your get_graph function, since it needs to be inside an async function to work.

📝 app/main.py

from typing import Annotated, Any, TypedDict
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.mongodb.aio import AsyncMongoDBSaver
from motor.motor_asyncio import AsyncIOMotorClient

async_mongodb_client: AsyncIOMotorClient[Any] = AsyncIOMotorClient(
    settings.mongo_uri.get_secret_value()


async def get_graph() -> CompiledStateGraph:
    """
    Get the graph
    """
    checkpointer = AsyncMongoDBSaver(
        client=async_mongodb_client,
        db_name="ai",
        checkpoint_collection_name="checkpoints",
        writes_collection_name="checkpoint_writes",
    )

    graph_builder = StateGraph(State)

    graph_builder.add_node("chatbot", chatbot)

    tool_node = ToolNode(tools=tools)
    graph_builder.add_node("tools", tool_node)

    graph_builder.add_edge(START, "chatbot")

    graph_builder.add_conditional_edges("chatbot", tools_condition)
    graph_builder.add_edge("tools", "chatbot")

    graph = graph_builder.compile(checkpointer=checkpointer)

    return graph

Now configure your thread_id using MongoDB for storing messages across different chat histories. This setup enables the assistant to remember conversations across multiple sessions and allows you to create separate chat histories by using a different thread_id.

📝 app/main.py

from langchain_core.runnables import RunnableConfig


async def run_graph(config: RunnableConfig, question: str) -> None:
    """
    Run the graph
    """
    graph = await get_graph()

    async for event in graph.astream(
        {
            "messages": [
                SystemMessage(content=system_prompt),
                HumanMessage(content=question),
            ]
        },
        config=config,
        stream_mode="values",
    ):
        event["messages"][-1].pretty_print()

    return None


async def main() -> None:
    """
    AI Agent
    """
    config = RunnableConfig(configurable={"thread_id": 1})

    while True:
        question = input("q: ")
        if question.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break

        await run_graph(config=config, question=question)

    return None

You can run the script and view the output, since we gave it the current date, you can see it gets the most recent information about the team.

$ python3 -m app.main

q: what is the carolina panthers current record?                
================================ Human Message =================================

what is the carolina panthers current record?
================================== Ai Message ==================================
Tool Calls:
  tavily_search_results_json (call_apvK6LYyrTMRunPcoO8enCqD)
 Call ID: call_apvK6LYyrTMRunPcoO8enCqD
  Args:
    query: Carolina Panthers current record December 2024
================================= Tool Message =================================
Name: tavily_search_results_json

[{"url": "https://www.footballdb.com/teams/nfl/carolina-panthers/results", "content": "View the 2024 Carolina Panthers schedule, results and scores for regular season, preseason and postseason NFL games. ... 2024 record: 3-11 (4th in NFC South) Draft; Splits; Transactions; Injuries; ... December 29, 2024 1:00 PM; Carolina (3-11)--Tampa Bay (8-6)--Last Meeting: TB 26 @ CAR 23 (12/1/2024)"}, {"url": "https://champsorchumps.us/team/nfl/carolina-panthers/2024", "content": "The Carolina Panthers currently have a 3-9 record. The 2024 Panthers are 2-5 at home and 1-4 on the road. What division do the 2024 Carolina Panthers play in? The 2024 Carolina Panthers played in the South Division of the National Football Conference."}]
================================== Ai Message ==================================

As of December 2024, the Carolina Panthers have a record of **3 wins and 11 losses** (3-11). They are currently in 4th place in the NFC South division.

You can view the run step by step on your langsmith project.

This is the basics on how to get an AI agent that can be used to decide on its own if it needs tool calling and can look up information on the web. as well as store information for each thread allowing your assistant to remember previous conversations. Since we are using langchain it will be easy to add other functions like streaming, multi-model support and other llms in the future.