Build a ChatBI Agent with LangGraph & MCP
A team that already built customer-service, multi-agent, and code-review agents faces a natural next step: enterprise data access. The goal is a ChatBI agent. A user asks a question in plain language; the agent reads the database schema, writes SQL, validates it, runs the query, and returns an answer. This article walks through that build with LangGraph for workflow orchestration and MCP for database access. The implementation was verified on October 12, 2026.
1. The Architecture: a State Machine Around a Thin Proxy
The full pipeline runs like this:
user question → intent detection → schema discovery → SQL generation → SQL validation → query execution → result interpretation → answer
LangGraph owns the workflow. It defines who does what, in what order, and how the flow branches when a step fails. MCP owns the tools. It wraps database operations as standard tools, so the agent calls them through one protocol instead of writing a per-database adapter.
The separation is the core of the design. LangGraph is the state machine that keeps the pipeline controllable. MCP is the thin proxy that keeps the data layer replaceable. Changing the database engine, or the table structure, does not force a rewrite of the agent code.
2. Step One: Install Dependencies
The required packages cover orchestration, the model client, the MCP adapter layer, and the MySQL driver:
pip install langgraph langchain-openai langchain-mcp-adapters fastmcp pymysql sqlalchemy
langchain-mcp-adapters is the bridge that turns MCP server tools into LangChain tool objects. fastmcp provides the server runtime for the database.
3. Step Two: Build the MCP Database Server
The MCP server sits between the agent and the database. It carries no business logic. It only performs protocol conversion and safety validation. Treating it as a thin proxy keeps the security rules in one place.
The implementation, mcp_db_server.py, starts with the imports and the server instance:
import os
import json
from fastmcp import FastMCP
import pymysql
from contextlib import contextmanager
mcp = FastMCP("DatabaseServer")
DB_CONFIG = {
"host": os.getenv("DB_HOST", "localhost"),
"user": os.getenv("DB_USER", "root"),
"password": os.getenv("DB_PASSWORD", ""),
"database": os.getenv("DB_NAME", "sales"),
"charset": "utf8mb4",
}
@contextmanager
def get_db():
conn = pymysql.connect(**DB_CONFIG)
try:
yield conn
finally:
conn.close()
The server exposes three tools. The first lists tables, the second returns a table's field structure, and the third executes a read-only query.
@mcp.tool()
def list_tables() -> str:
"""List all table names in the database"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute("SHOW TABLES")
tables = [row[0] for row in cursor.fetchall()]
return json.dumps(tables)
@mcp.tool()
def get_table_schema(table_name: str) -> str:
"""Return the field structure of a table (column, type, comment)"""
with get_db() as conn:
cursor = conn.cursor()
cursor.execute(f"DESCRIBE `{table_name}`")
columns = cursor.fetchall()
schema = []
for col in columns:
schema.append({
"field": col[0],
"type": col[1],
"nullable": col[2],
"key": col[3],
"comment": col[5] if len(col) > 5 else "",
})
return json.dumps(schema, ensure_ascii=False)
The execute tool is where safety lives. It accepts only SELECT statements and rejects destructive operations.
@mcp.tool()
def execute_sql(sql: str) -> str:
"""Execute a read-only SQL query. SELECT only."""
sql_upper = sql.strip().upper()
if not sql_upper.startswith("SELECT"):
return json.dumps({"error": "SELECT only; execution refused"})
if any(kw in sql_upper for kw in ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]):
return json.dumps({"error": "dangerous operation detected; refused"})
try:
with get_db() as conn:
cursor = conn.cursor(pymysql.cursors.DictCursor)
cursor.execute(sql)
rows = cursor.fetchall()
return json.dumps(rows, default=str, ensure_ascii=False)
except Exception as e:
return json.dumps({"error": str(e)})
if __name__ == "__main__":
mcp.run()
Start the service:
python mcp_db_server.py
The keyword blocklist is a first line of defense. It stops accidental writes even when the model misbehaves. A second, semantic check comes later in the workflow.
4. Step Three: Define the State
The state object carries everything between nodes. It is a typed dict, which keeps the graph predictable.
from typing import TypedDict, Optional, List
from langchain_core.messages import HumanMessage, AIMessage
class ChatBIState(TypedDict):
messages: list[HumanMessage | AIMessage]
user_query: str
tables: Optional[str]
schema: Optional[str]
generated_sql: Optional[str]
sql_valid: bool
query_result: Optional[str]
final_answer: Optional[str]
error: Optional[str]
retry_count: int
The fields map to pipeline stages: schema and generated_sql feed SQL generation; sql_valid gates execution; retry_count bounds the correction loop.
5. Step Four: Connect the MCP Tools
langchain-mcp-adapters exposes a client that loads tools from one or more servers. The model is initialized separately.
from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)
async def get_mcp_tools():
client = MultiServerMCPClient({
"database": {
"command": "python",
"args": ["mcp_db_server.py"],
"transport": "stdio"
}
})
tools = await client.get_tools()
return tools
The database server runs over stdio in this setup. Each mcp.tool becomes a standard LangChain tool that a node can invoke by name.
6. Step Five: Write the Core Nodes
6.1 Schema discovery node
The agent needs to know what it can query before it writes SQL. This node pulls the table list and the relevant schemas.
async def schema_discovery_node(state: ChatBIState) -> dict:
tools = await get_mcp_tools()
list_tables_tool = next(t for t in tools if t.name == "list_tables")
tables_str = await list_tables_tool.ainvoke({})
tables = json.loads(tables_str)
schema_info = []
get_schema_tool = next(t for t in tools if t.name == "get_table_schema")
for table in tables[:3]:
schema_str = await get_schema_tool.ainvoke({"table_name": table})
schema_info.append(f"table: {table}\n{schema_str}\n")
return {
"tables": tables_str,
"schema": "\n".join(schema_info),
}
The simple version inspects the first three tables. The comment in the source notes a production improvement: use vector retrieval for semantic matching when a database has many tables. Only the relevant schemas should be loaded.
6.2 SQL generation node
The model writes SQL from the schema and the user question. The prompt constrains output: use only tables and fields from the schema, emit SELECT only, default to LIMIT 10, and return SQL without explanation.
def sql_generation_node(state: ChatBIState) -> dict:
query = state["user_query"]
schema = state["schema"]
prompt = f"""
You are an SQL expert. Write an executable MySQL query for the user's question.
Database schema:
{schema}
User question: {query}
Requirements:
- Only use tables and fields from the schema above
- Only produce a SELECT statement
- Default to LIMIT 10 unless the user asks for more
- Output only SQL, no explanation
SQL:"""
response = llm.invoke(prompt)
sql = response.content.strip().strip("```sql").strip("```").strip()
return {"generated_sql": sql}
Stripping markdown fences is a small but necessary step. Models frequently wrap SQL in code blocks; the executor needs the raw statement.
6.3 SQL validation node: double insurance
Validation runs twice. The first pass is syntactic and mechanical. The second pass is semantic and model-based.
def sql_validation_node(state: ChatBIState) -> dict:
sql = state["generated_sql"]
sql_upper = sql.upper()
dangerous_keywords = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "CREATE", "TRUNCATE"]
for kw in dangerous_keywords:
if kw in sql_upper:
return {"sql_valid": False, "error": f"SQL contains a dangerous operation: {kw}"}
if not sql_upper.startswith("SELECT"):
return {"sql_valid": False, "error": "SELECT only"}
schema = state["schema"]
check_prompt = f"""
Check whether the SQL below uses correct tables and fields.
Schema: {schema}
SQL: {sql}
If correct, output only: VALID
If wrong, output: INVALID + reason"""
response = llm.invoke(check_prompt)
if "INVALID" in response.content:
return {"sql_valid": False, "error": f"semantic check failed: {response.content}"}
return {"sql_valid": True}
The mechanical pass catches destructive keywords even if the model wraps them oddly. The semantic pass catches wrong table or field names, which a blocklist cannot. The two together justify the phrase "double insurance."
6.4 SQL execution node
Execution goes through the MCP tool, not a direct connection. This keeps one enforcement point for read-only behavior.
async def sql_execution_node(state: ChatBIState) -> dict:
tools = await get_mcp_tools()
execute_tool = next(t for t in tools if t.name == "execute_sql")
result_str = await execute_tool.ainvoke({"sql": state["generated_sql"]})
result = json.loads(result_str)
if isinstance(result, dict) and "error" in result:
return {"query_result": None, "error": result["error"]}
return {"query_result": json.dumps(result, ensure_ascii=False, indent=2)}
6.5 Answer generation node
The final node translates raw rows into a natural-language answer.
def answer_generation_node(state: ChatBIState) -> dict:
query = state["user_query"]
result = state["query_result"]
prompt = f"""
You are a data analyst. The user asked: {query}
Query result:
{result}
Answer in concise language and highlight the key numbers. If the result is empty, say so plainly."""
response = llm.invoke(prompt)
return {"final_answer": response.content}
The instruction "say so plainly when empty" matters. Models will otherwise invent plausible numbers for an empty result.
6.6 SQL correction node
When validation or execution fails, a correction node regenerates the SQL with the error message as new context.
def sql_correction_node(state: ChatBIState) -> dict:
query = state["user_query"]
schema = state["schema"]
error = state.get("error", "")
old_sql = state.get("generated_sql", "")
prompt = f"""
The previous SQL failed. Fix it.
User question: {query}
Schema: {schema}
Original SQL: {old_sql}
Error: {error}
Generate a corrected SELECT statement. Output only SQL."""
response = llm.invoke(prompt)
sql = response.content.strip().strip("```sql").strip("```").strip()
return {"generated_sql": sql, "retry_count": state.get("retry_count", 0) + 1}
7. Step Six: Assemble the Graph and Compile
Routing functions decide where control goes after validation and after execution.
from langgraph.graph import StateGraph, START, END
from typing import Literal
def route_after_validation(state: ChatBIState) -> Literal["execute_sql", "correct_sql"]:
if state["sql_valid"]:
return "execute_sql"
return "correct_sql"
def route_after_execution(state: ChatBIState) -> Literal["answer", "correct_sql", "__end__"]:
if state.get("error"):
if state.get("retry_count", 0) < 3:
return "correct_sql"
return "__end__"
return "answer"
workflow = StateGraph(ChatBIState)
workflow.add_node("discover_schema", schema_discovery_node)
workflow.add_node("generate_sql", sql_generation_node)
workflow.add_node("validate_sql", sql_validation_node)
workflow.add_node("execute_sql", sql_execution_node)
workflow.add_node("correct_sql", sql_correction_node)
workflow.add_node("generate_answer", answer_generation_node)
workflow.add_edge(START, "discover_schema")
workflow.add_edge("discover_schema", "generate_sql")
workflow.add_edge("generate_sql", "validate_sql")
workflow.add_conditional_edges("validate_sql", route_after_validation)
workflow.add_conditional_edges("execute_sql", route_after_execution)
workflow.add_edge("correct_sql", "validate_sql")
workflow.add_edge("generate_answer", END)
app = workflow.compile()
The retry bound of three is the exit condition. Without it, a persistent error would loop forever between correction and validation.
8. Step Seven: Run the Test
A single query exercises the full graph.
import asyncio
async def main():
result = await app.ainvoke({
"messages": [],
"user_query": "sum sales per product category, ordered by sales descending",
"retry_count": 0,
})
print("user question:", result["user_query"])
print("generated SQL:", result.get("generated_sql"))
print("query result:", result.get("query_result"))
print("final answer:", result.get("final_answer"))
if result.get("error"):
print("error:", result["error"])
asyncio.run(main())
The test question is representative: it needs a GROUP BY, an aggregate, and an ordering clause. It also forces the schema and the natural-language answer to align.
9. Advanced: Multiple MCP Servers Working Together
Production agents rarely touch one system. LangGraph MCP Adapters connects several servers at once.
async def get_multi_tools():
client = MultiServerMCPClient({
"database": {
"command": "python",
"args": ["mcp_db_server.py"],
"transport": "stdio"
},
"github": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-github"],
"transport": "stdio"
},
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/data"],
"transport": "stdio"
}
})
return await client.get_tools()
Now the agent reads files and searches a code repository while it queries the database. It selects the tool chain by the user question. With several servers behind one agent, access control and traffic shaping become infrastructure concerns rather than code concerns.
10. Production Recommendations
Five practices separate a demo from a deployable system.
First, SQL safety is the baseline. The MCP server must enforce read-only validation, and the database account should use minimum privilege. A dedicated read-only user for the agent is the right default. No write permission should be granted.
Second, schema discovery should scale. When a database holds many tables, loading all schemas is wasteful and noisy. Use vector retrieval, with Qdrant or Elasticsearch, for semantic matching, and recall only the tables relevant to the question.
Third, give the correction loop an exit. The generate-validate-execute-correct cycle must cap its retry count. An unbounded loop turns a query error into a runaway cost.
Fourth, trace the pipeline. LangSmith records the full path of every SQL generation, validation, and execution step. That history answers the recurring question: why was the wrong SQL generated?
Fifth, deploy the MCP server over HTTP. In production, streamable-http transport lets each server run as a standalone microservice. That is also where unified authentication and rate limiting belong. An API gateway such as 4sapi is the standard layer for those policies: keys, limits, and per-route metering sit in one place instead of being repeated inside each server.
These two observability layers answer different questions. LangSmith records what happened in the workflow. Gateway accounting records what each route cost. Both are needed: one for debugging, one for budgeting.
11. Summary: LangGraph Manages Flow, MCP Manages Tools
The architecture's core idea is a clean separation. LangGraph, as a state machine, strings schema discovery, SQL generation, validation, execution, and interpretation into a controllable pipeline. MCP standardizes database operations, so the agent calls tools through a unified protocol. Replacing the database or changing table structure does not touch the agent code.
Compared with letting a model write SQL directly, this design adds schema discovery and double validation. Accuracy and safety both become more controllable, because every risky step is gated and every failure has a bounded retry path.
International access: [https://4sapi.com](https://4sapi.com)
Domestic access: [https://4sapi.org](https://4sapi.org)
