Practical MCP with FastMCP & LangChain

Database Setup

All database concern lives in server/database.py. The file has 5 distinct responsibilities:

• Imports and configuration
• Engine/session setup,
• The session dependency,
• The lifespan hook
• The ORM models.

We'll go through each in turn.

Imports and Configuration

Here, we import necessary libraries and load the DATABASE_URL from the environment.

# database.py
import os
from contextlib import contextmanager

from dotenv import load_dotenv
from sqlalchemy import BigInteger, Boolean, Column, Date, DateTime, Integer, String
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

from fastmcp.server.lifespan import lifespan

load_dotenv()
DATABASE_URL = os.getenv("DATABASE_URL")
if not DATABASE_URL:
    raise RuntimeError(
        "DATABASE_URL environment variable is not set. Check your .env file."
    )

load_dotenv() is called here as well as in main.py because database.py is imported at module load time and needs the DATABASE_URL variable immediately to create the engine. The RuntimeError at module level acts as a fast-fail: if the connection string is missing the server refuses to start with a clear message rather than failing silently later with a confusing SQLAlchemy error.

Engine and Session Factory

Here, we set up the SQLAlchemy engine, session factory, and base class for ORM models.

# SQLAlchemy setup: create engine
engine = create_engine(DATABASE_URL)
# SQLAlchemy setup: session factory
SessionLocal = sessionmaker(bind=engine)
# SQLAlchemy setup: base class for ORM models
Base = declarative_base()

create_engine sets up the connection pool using the URL from the .env file. sessionmaker produces a factory: every call to SessionLocal() returns a fresh Session object tied to that engine. declarative_base() creates the parent class for our ORM models — any class that inherits from Base and defines __tablename__ will map to a database table.

Session Dependency

This is the function we pass to Depends(get_db_session) in every tool signature.

@contextmanager
def get_db_session():
    """Provides a SQLAlchemy session and ensures it is closed after use."""
    session = SessionLocal()
    try:
        yield session
    finally:
        session.close()

FastMCP's dependency injection system calls it before the tool runs, yields the session into the db parameter, and guarantees the finally block runs after the tool finishes — whether it succeeded or raised an exception. This means no tool ever needs to open or close a session manually.

Lifespan Hook

As a reminder, the lifespan hook manages the lifecycle of resources that need to be set up when the server starts and cleaned up when it stops.

In our case, we want to ensure the database engine is ready to go when the server is running and that all connections are properly closed when it shuts down. This is a good practice to avoid connection leaks and ensure clean resource management not just for the database but for any other long-lived resources you might add in the future (e.g., external API clients, file handles, etc.).

@lifespan
async def db_lifespan(server):
    """Manage database engine lifecycle."""
    print("Netflix MCP Server starting... Database engine ready.")

    try:
        # Yield empty context — no shared state needed here.
        # If you wanted to share objects with tools you could yield:
        # {"cache": some_cache} and read it via ctx.lifespan_context["cache"]
        yield {}
    finally:
        # Always runs, even on Ctrl+C or unhandled exceptions
        print("Shutting down... Disposing database engine.")
        engine.dispose()
        print("Netflix MCP Server shutdown complete.")

• The @lifespan decorator from FastMCP turns this async generator into the server's startup/shutdown handler.