LangGraph + MCP in Practice: Building a ChatBI Agent That Queries Data in Natural Language
The author used LangGraph to orchestrate the workflow and MCP to wrap database operations, building a ChatBI agent that lets users query the database in plain language and get results back automatically. The flow goes: intent recognition, fetching table schemas, generating SQL, validating SQL, executing the query, and interpreting the results. The MCP Server exposes only three tools—list_tables, get_table_schema, and execute_sql—and enforces read-only validation.
A complete implementation path for a ChatBI agent with LangGraph orchestration and MCP-wrapped database access, including double SQL validation and retry/exit conditions, ready to copy and adapt.
LangGraph + MCP in Practice: Building a ChatBI Agent That Answers Data Questions in Natural Language
In earlier posts we built agents for customer service, multi-agent collaboration, and code review. This time we're moving up to an enterprise scenario: using LangGraph to orchestrate the workflow and MCP to connect to the database, building a ChatBI agent where users just ask in plain language and the agent queries the database and returns the results. 1. Overall Architecture Text
Edit
User query → LangGraph intent recognition → Retrieve table schema → Generate SQL → Validate SQL → Execute query → Interpret results → Output response Key responsibilities: LangGraph: orchestrates the workflow, defining "who does what and when," managing state transitions and exception fallbacks. MCP: wraps database operations into standardized tools that Agents invoke through a unified protocol, so you don't need to write a separate adapter for each database. II. Step 1: Install dependencies bash
Edit
pip install langgraph langchain-openai langchain-mcp-adapters fastmcp pymysql sqlalchemy
三、第二步：构建 MCP Database Server
MCP Server 是 Agent 和数据库之间的"薄代理层"，本身不写业务逻辑，只负责协议转换和安全校验。
python
Edit
# mcp_db_server.py
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()
@mcp.tool()
def list_tables() -> str:
"""列出数据库中所有可用的表名"""
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:
"""获取指定表的字段结构（列名、类型、注释）"""
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)
@mcp.tool()
def execute_sql(sql: str) -> str:
"""执行只读 SQL 查询并返回结果。仅允许 SELECT 语句。"""
sql_upper = sql.strip().upper()
if not sql_upper.startswith("SELECT"):
return json.dumps({"error": "只允许 SELECT 查询，拒绝执行"})
if any(kw in sql_upper for kw in ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER"]):
return json.dumps({"error": "检测到危险操作，拒绝执行"})
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()
启动服务：
bash
编辑
python mcp_db_server.py
四、第三步：定义 State
python
编辑
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
sql_valid: bool # SQL 是否通过校验
query_result: Optional[str] # 查询结果
final_answer: Optional[str] # 最终回答
error: Optional[str] # 错误信息
retry_count: int # 重试次数
五、第四步：连接 MCP 工具
使用 langchain-mcp-adapters 把 MCP Server 暴露的工具转换为 LangChain 标准 Tool 对象。
python
编辑
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
六、第五步：编写核心 Node
. Schema 发现节点：先搞清楚有哪些表、哪些字段
python
编辑
async def schema_discovery_node(state: ChatBIState) -> dict:
"""获取可用表和关键字段信息，为 SQL 生成提供上下文"""
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)
# 根据用户问题，智能选择可能相关的表（简单版：取前3个表的结构）
# 生产环境可用向量检索做语义匹配
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}\n{schema_str}\n")
return {
"tables": tables_str,
"schema": "\n".join(schema_info)
}
. SQL 生成节点：让 LLM 结合表结构生成 SQL
python
编辑
def sql_generation_node(state: ChatBIState) -> dict:
"""基于用户问题和表结构生成 SQL"""
query = state["user_query"]
schema = state["schema"]
prompt = f"""你是一个 SQL 专家。请根据以下数据库结构和用户问题，生成一条可执行的 MySQL 查询语句。
数据库表结构：
{schema}
用户问题：{query}
要求：
. 只使用上述表结构中的表和字段
. 只生成 SELECT 语句
. 结果默认 LIMIT 10，除非用户明确要求更多
. 只输出 SQL，不要任何解释
生成的 SQL："""
response = llm.invoke(prompt)
sql = response.content.strip().strip("```sql").strip("```").strip()
return {"generated_sql": sql}
. SQL 校验节点：双重保险
python
编辑
def sql_validation_node(state: ChatBIState) -> dict:
"""校验 SQL 的安全性（只读）和语法合理性"""
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 包含危险操作: {kw}"
}
if not sql_upper.startswith("SELECT"):
return {"sql_valid": False, "error": "只允许 SELECT 查询"}
# 让 LLM 做一次语义校验
schema = state["schema"]
check_prompt = f"""请校验以下 SQL 是否使用了正确的表和字段。
表结构：{schema}
SQL：{sql}
如果正确，只输出：VALID
如果有问题，输出：INVALID + 原因"""
response = llm.invoke(check_prompt)
if "INVALID" in response.content:
return {
"sql_valid": False,
"error": f"语义校验失败: {response.content}"
}
return {"sql_valid": True}
. SQL 执行节点：通过 MCP 工具查数据库
python
编辑
async def sql_execution_node(state: ChatBIState) -> dict:
"""通过 MCP 工具执行 SQL 查询"""
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)
}
. 结果解读节点：把数据翻译成自然语言
python
编辑
def answer_generation_node(state: ChatBIState) -> dict:
"""将查询结果转化为自然语言回答"""
query = state["user_query"]
result = state["query_result"]
prompt = f"""你是一个数据分析师。用户问：{query}
查询结果如下：
{result}
请用简洁的中文回答用户，突出关键数据。如果结果为空，如实告知。"""
response = llm.invoke(prompt)
return {"final_answer": response.content}
. SQL 修正节点：校验或执行失败时重新生成
python
编辑
def sql_correction_node(state: ChatBIState) -> dict:
"""根据错误信息重新生成 SQL"""
query = state["user_query"]
schema = state["schema"]
error = state.get("error", "")
old_sql = state.get("generated_sql", "")
prompt = f"""之前的 SQL 执行失败，请修正。
用户问题：{query}
表结构：{schema}
原 SQL：{old_sql}
错误信息：{error}
请生成修正后的 SELECT 语句，只输出 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
}
七、第六步：组装图并编译
python
编辑
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()
八、第七步：运行测试
python
编辑
import asyncio
async def main():
result = await app.ainvoke({
"messages": [],
"user_query": "统计每个产品类别的销售额，按销售额降序排列",
"retry_count": 0
})
print("用户问题:", result["user_query"])
print("生成 SQL:", result.get("generated_sql"))
print("查询结果:", result.get("query_result"))
print("最终回答:", result.get("final_answer"))
if result.get("error"):
print("错误:", result["error"])
asyncio.run(main())
九、进阶：多 MCP Server 协同
生产环境中，Agent 往往需要同时访问多个系统。LangGraph MCP Adapters 支持同时连接多个 MCP Server：
python
编辑
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()
This lets the Agent read files and search code repositories while querying the database, choosing the right toolchain on its own based on the user's question. Ten: Production Recommendations SQL security is the baseline: the MCP Server layer must enforce read-only validation, and the database account should use least privilege (a read-only user)—never give the Agent write access. Schema discovery needs to be smart: with many tables you can't load everything, so use vector retrieval (Qdrant/ES) for semantic matching and only recall the schemas of relevant tables. Add exit conditions to the loop: the cycle of SQL generation → validation → execution → correction must cap the number of retries to avoid infinite loops. Integrate LangSmith for tracing: record the full process of every SQL generation, validation, and execution so you can troubleshoot "why did it generate the wrong SQL". Deploy the MCP Server over HTTP: in production, use the streamable-http transport and deploy the MCP Server as a standalone microservice, which makes unified authentication and rate limiting easier. Summary The core idea of this architecture is: LangGraph manages the flow, MCP manages the tools. LangGraph uses a state machine to string "discover table schema → generate SQL → validate → execute → interpret" into a controllable pipeline; MCP standardizes database operations so the Agent calls tools through a unified protocol—switching databases or table schemas requires no changes to the Agent's code. Compared to letting the LLM write raw SQL directly, this adds two safeguards—schema discovery and double validation—making both accuracy and security more controllable.
Source: Juejin · Weekly AI trends · juejin.cn