Guide · ORM Integration

MCP Tools for SQLAlchemy — async engine setup, session-per-tool-call, selectinload vs joinedload

Three SQLAlchemy behaviours regularly trip MCP server authors: a module-level Session object shared across tool calls leaks transaction state between invocations — SQLAlchemy sessions are not thread-safe and carry an identity map that grows unbounded across calls; an AsyncSession that is not closed after each tool call holds a connection from the pool until garbage-collected, silently exhausting the pool under concurrent agent invocations; lazy="select" (SQLAlchemy's default relationship loading strategy) fires a new SQL query for each relationship access — a tool returning 50 users with their posts issues 51 queries, and because Python async code can interleave these subqueries with other tool calls, they arrive at the database as unordered concurrent requests that each consume a pool connection; and pool_pre_ping=True is required for MCP servers that idle between agent runs — without it, a connection borrowed from the pool after a period of inactivity may be stale (TCP connection dropped by the database or a NAT gateway), and the first query on that connection raises an OperationalError: SSL connection has been closed unexpectedly that looks like a database outage but is actually a pool hygiene problem.

TL;DR

Create one AsyncEngine at startup with pool_pre_ping=True. Use async_sessionmaker to produce one AsyncSession per tool call, closing it in a finally block. Load relationships with selectinload() (best for one-to-many) or joinedload() (best for many-to-one). Wrap multi-step writes in async with session.begin(). Health probe: await session.execute(text("SELECT 1")).

Async engine and session factory — one session per tool invocation

SQLAlchemy 2.x async support is built on asyncio via the sqlalchemy.ext.asyncio module. The engine manages the connection pool; the session carries the unit-of-work identity map. Creating a new session per tool call keeps the identity map small, ensures no transaction state bleeds between calls, and returns the pool connection promptly.

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy import text
import os

# Create engine once at module level — shared across all tool calls
engine = create_async_engine(
    os.environ["DATABASE_URL"],  # postgresql+asyncpg://user:pass@host/db
    pool_size=20,          # max persistent connections in the pool
    max_overflow=5,        # extra connections allowed when pool_size is exhausted
    pool_timeout=5,        # seconds to wait for a pool slot before raising
    pool_recycle=1800,     # recycle connections after 30 minutes (avoids server timeout)
    pool_pre_ping=True,    # test connection on checkout — catches stale connections after idle
    echo=False,            # set True for SQL logging during development only
)

# Session factory — call it to produce a new AsyncSession for each tool call
AsyncSessionLocal = async_sessionmaker(
    bind=engine,
    expire_on_commit=False,  # Don't expire attributes after commit — avoids lazy loads post-commit
    autoflush=False,         # Explicit flush gives control over when SQL is emitted
    autocommit=False,
)

class Base(DeclarativeBase):
    pass

# Context manager helper — ensures session is closed after every tool call
from contextlib import asynccontextmanager
from typing import AsyncGenerator

@asynccontextmanager
async def get_session() -> AsyncGenerator[AsyncSession, None]:
    session = AsyncSessionLocal()
    try:
        yield session
    except Exception:
        await session.rollback()
        raise
    finally:
        await session.close()  # Returns connection to the pool

# Usage in an MCP tool handler
async def handle_get_user(user_id: str) -> dict:
    async with get_session() as session:
        result = await session.execute(
            select(User).where(User.id == user_id)
        )
        user = result.scalar_one_or_none()
        if user is None:
            return {"error": "user not found"}
        return {"id": user.id, "name": user.name, "email": user.email}

# Startup — verify engine connectivity before MCP server starts
async def startup():
    async with engine.connect() as conn:
        await conn.execute(text("SELECT 1"))
    print("Database engine ready")

# Shutdown — dispose pool gracefully
async def shutdown():
    await engine.dispose()

expire_on_commit=False is important for MCP tools. The default expire_on_commit=True marks all attributes as expired after a commit, meaning any attribute access after the session commits (e.g., while serializing the return value) triggers a new lazy SQL query. With an already-closed session, this raises MissingGreenlet or DetachedInstanceError. Setting expire_on_commit=False means committed instances retain their values for serialization without hitting the database again.

Relationship loading strategies — selectinload vs joinedload for MCP tool results

SQLAlchemy's default relationship loading is lazy="select": the related objects are fetched in a separate query when you first access the attribute. In async code, this lazy access raises MissingGreenlet: greenlet_spawn has not been called if you try it outside an async context (e.g., during serialization). You must explicitly load relationships before the session closes.

from sqlalchemy import select
from sqlalchemy.orm import selectinload, joinedload, contains_eager, load_only
from models import User, Post, Comment

# BAD: lazy access raises MissingGreenlet in async sessions
async def get_user_bad(session: AsyncSession, user_id: str):
    result = await session.execute(select(User).where(User.id == user_id))
    user = result.scalar_one()
    # This line raises MissingGreenlet — lazy load triggered outside greenlet context
    posts = user.posts
    return posts

# GOOD: selectinload — separate IN-clause query, best for one-to-many
# SELECT * FROM posts WHERE user_id IN (id1, id2, ...)
async def get_users_with_posts(session: AsyncSession) -> list:
    result = await session.execute(
        select(User)
        .options(selectinload(User.posts))   # One extra query for all posts
        .limit(50)
        .order_by(User.created_at.desc())
    )
    users = result.scalars().all()
    return [
        {"id": u.id, "name": u.name, "posts": [{"id": p.id, "title": p.title} for p in u.posts]}
        for u in users
    ]

# joinedload — single JOIN query, best for many-to-one (avoids duplicate rows for one-to-many)
# SELECT users.*, posts.* FROM users LEFT JOIN posts ON posts.user_id = users.id
async def get_posts_with_author(session: AsyncSession) -> list:
    result = await session.execute(
        select(Post)
        .options(joinedload(Post.author))    # JOIN — good for single parent record per post
        .limit(50)
    )
    posts = result.unique().scalars().all()  # .unique() required after joinedload to deduplicate
    return [{"id": p.id, "title": p.title, "author": p.author.name} for p in posts]

# Deep nesting: chain options() calls
async def get_posts_deep(session: AsyncSession) -> list:
    result = await session.execute(
        select(Post)
        .options(
            joinedload(Post.author).load_only(User.id, User.name),  # Limit author columns
            selectinload(Post.comments).selectinload(Comment.author),
        )
        .limit(20)
    )
    return result.unique().scalars().all()

# load_only — select specific columns to avoid fetching large TEXT/BLOB fields
async def get_user_list(session: AsyncSession) -> list:
    result = await session.execute(
        select(User).options(load_only(User.id, User.name, User.created_at))
    )
    return result.scalars().all()

# contains_eager — use when you've already joined in the query (avoids double-join)
async def get_active_posts(session: AsyncSession) -> list:
    result = await session.execute(
        select(Post)
        .join(Post.author)
        .where(User.is_active == True)
        .options(contains_eager(Post.author))  # Reuse the join, don't add another
    )

Use selectinload for one-to-many relationships (one user has many posts) and joinedload for many-to-one (one post has one author). A JOIN for one-to-many multiplies rows in the result set — 50 users with 10 posts each returns 500 rows — whereas selectinload always returns exactly N rows for N users plus one additional query.

Transaction management in async tool handlers

SQLAlchemy 2.x's preferred transaction pattern uses async with session.begin() as a context manager: it begins the transaction, commits on normal exit, and rolls back on exception. This is safer than manually calling session.commit() because it handles the rollback case automatically.

from sqlalchemy import update, select
from sqlalchemy.ext.asyncio import AsyncSession
from models import User, Transfer, AuditLog
from decimal import Decimal

# Atomic multi-step write — commit or rollback as a unit
async def transfer_credits(
    session: AsyncSession,
    from_user_id: str,
    to_user_id: str,
    amount: Decimal,
) -> dict:
    async with session.begin():  # Begins transaction; commits on exit, rolls back on exception
        # SELECT ... FOR UPDATE — locks rows to prevent concurrent double-spends
        result = await session.execute(
            select(User)
            .where(User.id.in_([from_user_id, to_user_id]))
            .with_for_update()
        )
        users = {u.id: u for u in result.scalars().all()}

        from_user = users.get(from_user_id)
        to_user   = users.get(to_user_id)

        if not from_user or not to_user:
            raise ValueError("User not found")
        if from_user.credits < amount:
            raise ValueError(f"Insufficient credits: {from_user.credits} available")

        from_user.credits -= amount
        to_user.credits   += amount
        session.add_all([from_user, to_user])

        # Audit record in the same transaction
        session.add(AuditLog(
            action="transfer_credits",
            from_user_id=from_user_id,
            to_user_id=to_user_id,
            amount=float(amount),
        ))
        # session.begin() context manager commits here
    return {"ok": True, "from_credits": float(from_user.credits), "to_credits": float(to_user.credits)}

# Savepoint for partial failures in bulk operations
async def bulk_create_tags(session: AsyncSession, tag_names: list[str]) -> dict:
    results = []
    async with session.begin():
        for name in tag_names:
            try:
                async with session.begin_nested():  # SAVEPOINT
                    tag = Tag(name=name.strip().lower())
                    session.add(tag)
                    await session.flush()  # Detects unique constraint violation here
                    results.append({"name": name, "status": "created", "id": tag.id})
            except Exception as e:
                # SAVEPOINT rolled back — outer transaction still open
                results.append({"name": name, "status": "skipped", "reason": str(e)})
    return {"results": results}

# Read-only tool: use autobegin (no explicit begin needed for reads)
async def get_account_summary(session: AsyncSession, user_id: str) -> dict:
    result = await session.execute(
        select(User).where(User.id == user_id)
    )
    user = result.scalar_one_or_none()
    if user is None:
        return {"error": "user not found"}
    return {"id": user.id, "credits": float(user.credits), "email": user.email}

Avoid calling session.commit() multiple times in a single tool handler. SQLAlchemy's identity map does not automatically refresh after each commit, and partially-committed state may show stale values for subsequent queries in the same session. Wrap everything in one async with session.begin() block and commit once.

Health probe — SQLAlchemy engine reachability from an MCP server

pool_pre_ping=True performs a connectivity test at connection checkout time. For an explicit health check, execute a trivial query via a raw connection and measure the round-trip time.

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncEngine
import time

async def check_sqlalchemy_health(engine: AsyncEngine) -> dict:
    start = time.monotonic()
    try:
        async with engine.connect() as conn:
            await conn.execute(text("SELECT 1"))
        latency_ms = round((time.monotonic() - start) * 1000, 1)
        return {"healthy": True, "latency_ms": latency_ms}
    except Exception as exc:
        return {"healthy": False, "error": str(exc)}

# Pool statistics — surface in health endpoint or metrics
def pool_stats(engine: AsyncEngine) -> dict:
    pool = engine.sync_engine.pool
    return {
        "pool_size":      pool.size(),
        "checked_in":     pool.checkedin(),
        "checked_out":    pool.checkedout(),
        "overflow":       pool.overflow(),
    }

# Startup gate
async def startup_check(engine: AsyncEngine):
    result = await check_sqlalchemy_health(engine)
    if not result["healthy"]:
        raise RuntimeError(f"Database unreachable at startup: {result['error']}")
    print(f"SQLAlchemy engine ready ({result['latency_ms']}ms)")
    stats = pool_stats(engine)
    print(f"Pool: {stats['pool_size']} max, {stats['checked_out']} checked out")

AliveMCP monitors your MCP server endpoints on a 60-second probe cycle and alerts when the health check fails or when response latency exceeds your configured SLA — catching pool exhaustion, database failovers, and stale-connection bursts before they cascade into tool call errors in your agents.

TL;DR

Async engine and session factory — one session per tool invocation

Relationship loading strategies — selectinload vs joinedload for MCP tool results

Transaction management in async tool handlers

Health probe — SQLAlchemy engine reachability from an MCP server

Related guides