Building a SQL-Powered AI Agent with LangGraph and FastMCP 🤖📑

Rishiksai SanthoshRishiksai Santhosh
5 min read

Table of contents

🚀 What Is the Model Context Protocol (MCP)?

MCP is an open protocol that standardizes how applications provide context to LLMs. Think of MCP like a USB-C port for AI applications. Just as USB-C provides a standardized way to connect your devices to various peripherals and accessories, MCP provides a standardized way to connect AI models to different data sources and tools.

Why MCP ?

MCP helps you build agents and complex workflows on top of LLMs. LLMs frequently need to integrate with data and tools, and MCP provides:

  • A growing list of pre-built integrations that your LLM can directly plug into

  • The flexibility to switch between LLM providers and vendors

  • Best practices for securing your data within your infrastructure

The blog is divided into two parts :

  • Part 1: discusses the methodology for building our custom MCP for database interaction

  • Part 2: Integerating custom MCP Servers with LangGraph Agents

✅ What You'll Learn

  • How to build an MCP server using FastMCP

  • How to manage state (selected database) across tools

  • How to define database-related tools (create DB, insert rows, run queries)

  • How LangGraph agents can interact with this MCP for SQL workflows

📦 Dependencies

To get started, install the following Python packages:

uv pip install mysql-connector-python "mcp[cli]"
uv pip install -U langgraph langchain-mcp-adapters

mkdir sql_agent
cd sql_agent
touch sql_mcp.py
touch app.py

🔧 Part 1 : Step-by-Step: Writing the MCP Server (sql_mcp.py)

1. Configuration for MySQL

We begin by defining a DBConfig class to store credentials and create connections.

from dataclasses import dataclass
from typing import Optional
import mysql.connector

@dataclass
class DBConfig:
    mysqlHost: str
    mysqlPort: int
    mysqlUser: str
    mysqlPassword: str
    mysqlDatabase: Optional[str] = None

    def connect(self):
        return mysql.connector.connect(
            host=self.mysqlHost,
            port=self.mysqlPort,
            user=self.mysqlUser,
            password=self.mysqlPassword,
            database=self.mysqlDatabase
        )

We store a global config object that keeps track of the current connection state, including which database to use.

config = DBConfig(
    mysqlHost="localhost",
    mysqlPort=3306,
    mysqlUser="root",
    mysqlPassword="your_password"
)

Replace the User and Password with your credentials, ensure the credentials are valid

2. Creating the FastMCP Server

Now, we create the MCP server:

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("SQL MCP Server", dependencies=["mysql-connector-python"])

You can give it any name. The dependencies field helps LangGraph know what packages are needed if the MCP is containerized or deployed.

3. Tool: Create and Drop Tables

@mcp.tool("create_table")
def create_table(table_name: str, columns: str) -> str:
    """
    Create a table with column definitions.
    Example:
        create_table("users", "id INT PRIMARY KEY, name VARCHAR(100)")
    """
    conn = config.connect()
    cursor = conn.cursor()
    cursor.execute(f"CREATE TABLE `{table_name}` ({columns})")
    cursor.close()
    conn.close()
    return f"✅ Table `{table_name}` created."

@mcp.tool("drop_table")
def drop_table(table_name: str) -> str:
    """
    Drop a table from the current database.
    Example:
        drop_table("users")
    """
    conn = config.connect()
    cursor = conn.cursor()
    cursor.execute(f"DROP TABLE `{table_name}`")
    cursor.close()
    conn.close()
    return f"🗑️ Table `{table_name}` dropped."

4. Tool: Insert Rows

@mcp.tool("insert_row")
def insert_row(table_name: str, columns: str, values: str) -> str:
    """
    Insert a single row into a table.
    Example:
        insert_row("users", "name, age", "'Alice', 25")
    """
    conn = config.connect()
    cursor = conn.cursor()
    try:
        cursor.execute(f"INSERT INTO `{table_name}` ({columns}) VALUES ({values})")
        conn.commit()
        result = f"✅ Inserted row into `{table_name}`."
    except Exception as e:
        result = f"❌ Error: {str(e)}"
    finally:
        cursor.close()
        conn.close()
    return result

5. Tool: Insert Multiple Rows

@mcp.tool("insert_multiplt_rows")
def insert_multiple_rows(table_name: str, columns: str, values: list[str]) -> str:
    """
    Insert multiple rows into a table.
    Example:
        insert_multiple_rows("users", "name, age", ["'Alice', 25", "'Bob', 30"])
    """
    conn = config.connect()
    cursor = conn.cursor()
    try:
        placeholder = ", ".join(["%s"] * len(columns.split(",")))
        sql = f"INSERT INTO `{table_name}` ({columns}) VALUES ({placeholder})"
        data = [tuple(v.strip().strip("'").split(",")) for v in values]
        cursor.executemany(sql, data)
        conn.commit()
        result = f"✅ Inserted multiple rows into `{table_name}`."
    except Exception as e:
        result = f"❌ Error: {str(e)}"
    finally:
        cursor.close()
        conn.close()
    return result

6. Tool: Execute Raw SQL Query

@mcp.tool("execute_query")
def execute_query(query: str):
    """
    Execute any SQL query and return result.
    Example:
        execute_query("SELECT * FROM users")
    """
    conn = config.connect()
    cursor = conn.cursor()
    try:
        cursor.execute(query)
        if cursor.with_rows:
            result = cursor.fetchall()
        else:
            result = [(f"Affected rows: {cursor.rowcount}",)]
        conn.commit()
    except Exception as e:
        result = [(f"❌ Error: {str(e)}",)]
    finally:
        cursor.close()
        conn.close()
    return result

if __name__ == "__main__":
    # Run the MCP server
    mcp.run(transport="stdio")

🧪 Testing the Server

We can test the server locally to ensure proper execution of functionalities. To run the MCP server locally run the below code in terminal window

mcp dev server.py

You’ll see something like:

Starting MCP inspector...
⚙️ Proxy server listening on 127.0.0.1:6277
🔑 Session token: 976de7ef07e7bc6ba4a359ff7e8dc5515444561901f7583fe1668b98f735fef1
Use this token to authenticate requests or set DANGEROUSLY_OMIT_AUTH=true to disable auth

🔗 Open inspector with token pre-filled:
   http://localhost:6274/?MCP_PROXY_AUTH_TOKEN=976de7ef07e7bc6ba4a359ff7e8dc5515444561901f7583fe1668b98f735fef1
   (Auto-open is disabled when authentication is enabled)

🔍 MCP Inspector is up and running at http://127.0.0.1:6274 🚀

Part 2 : Integrate the MCP Server with LangGraph workflow (app.py)

Required imports

from langchain_mcp_adapters.tools import load_mcp_tools
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain.chat_models import init_chat_model
import asyncio

Defining the MCP Client

client = MultiServerMCPClient(
    {
        "sql_mcp_server" : {
            "command" : "python",
            "args" : ["sql_mcp.py"],
            "transport" : "stdio",
        }
    }
)

Define an async function to write the agent logic

Note: The tools must be used inside a an async function else it causes errors. Also ensure the await is properly used wherever necessary

agent = create_react_agent(
        model="google_genai:gemini-2.5-flash",
        tools=tools,
    )

    query = """Create a table 'students' with columns 'id', 'name', and 'age'.
                                            Add 4 students.
                                            Then, list all students in the database."""
    # Run the agent
    response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": query}]},debug=True
    )

    print(response)

Complete main function

async def main():

    tools = await client.get_tools()

    prompt = PromptTemplate.from_template(
        """
        You are a helpful assistant that can interact with a SQL database. You can perform CRUD operations on the database.
        Your task is to analyze the users request and perform the appropriate action using the available tools.
        """
    )

    agent = create_react_agent(
        model="google_genai:gemini-2.5-flash",
        tools=tools,
    )

    query = """Create a table 'students' with columns 'id', 'name', and 'age'.
                                            Add 4 students.
                                            Then, list all students in the database."""
    # Run the agent
    response = await agent.ainvoke(
        {"messages": [{"role": "user", "content": query}]},debug=True
    )

    print(response)

Run the agent function using asyncio

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

Execute the app.py file to run the agent

python app.py

References

  1. LangGraph MCP Documentation

  2. MCP Python SDK Documentation

  3. Model Context Protocol Introduction

0
Subscribe to my newsletter

Read articles from Rishiksai Santhosh directly inside your inbox. Subscribe to the newsletter, and don't miss out.

generative aiModel Context ProtocollanggraphDatabases

Written by

Rishiksai Santhosh
Rishiksai Santhosh