SQL Analyst LangGraph
blueprint · 4 agents · 3 MCP servers
842★ stars
119forks
4.7rating
What it does
A four-agent workflow. A planner breaks the question down, an analyst writes and runs SQL against a Postgres server, a critic checks the SQL for correctness and cost, and a writer explains the result for a non-technical reader. The analyst and critic loop until the critic signs off. Swap the database server for your own and it keeps working.
The cast · 4 agents
Criticcriticreasoning
Check the analyst's SQL for correctness and cost. Run explain plans. Return an object with an 'approved' boolean and specific notes. Approve only when the query is correct and reasonably cheap.
db.explain
Writerwriterfast
Explain the final result for someone who does not read SQL. Keep it to a short paragraph. Save the explanation to a report file.
fs.write_file
SQL Analystanalystcoding
Write and run read-only SQL that answers the plan. Use the schema tools to check table and index names first. If a data point needs outside context, like a definition or an industry benchmark, look it up with web search before writing the query. If you receive a review, address every point and try again.
db.list_indicesdb.query_dslsearch.web
Plannerplannerreasoning
Break the user's question into the concrete data points needed to answer it. Do not write SQL. Produce a short plan.
no tools · reasoning only
Flow
→ Planner → plan
↻ loop until review.approved == true · max 3
→ SQL Analyst → draft
→ Critic → review
→ Writer → answer
Interface
Inputs
questionstring
The business question in plain English.
Outputs
answerstring
A plain-language answer for a non-technical reader, including the query it ran.
Depends on · 3 MCP servers
io.github.supabase/postgres-mcpnot in the registry yetdb
io.github.brave/brave-search-mcpnot in the registry yetsearch
io.github.modelcontextprotocol/filesystemnot in the registry yetfs⤓ Export runnable code
# Generated by @socketcat/compiler for target: langgraph
# blueprint: com.socketcat/sql-analyst v1.2.0 schema: socketcat.dev/blueprint/v0
# This code is yours. Edit it freely. The socketcat_runtime helper is optional and can be vendored.
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
import socketcat_runtime as _rt
class State(TypedDict, total=False):
question: object
answer: object
plan: object
__loop_1: object
draft: object
review: object
def node_AgentInvoke_0(state):
return _rt.run_agent("planner", state, {"question":"question"}, "plan")
def node_LoopHeader_1(state):
return {}
def node_LoopTick_2(state):
return {"__loop_1": state.get("__loop_1", 0) + 1}
def node_Nop_3(state):
return {}
def node_AgentInvoke_4(state):
return _rt.run_agent("analyst", state, {"plan":"plan","review":"review"}, "draft")
def node_AgentInvoke_5(state):
return _rt.run_agent("critic", state, {"draft":"draft"}, "review")
def node_AgentInvoke_6(state):
return _rt.run_agent("writer", state, {"draft":"draft"}, "answer")
def route_LoopHeader_1(state):
count = state.get("__loop_1", 0)
return "exit" if (_rt.cond("review.approved == true", state) or count >= 3) else "loop"
def build():
b = StateGraph(State)
b.add_node("AgentInvoke_0", node_AgentInvoke_0)
b.add_node("LoopHeader_1", node_LoopHeader_1)
b.add_node("LoopTick_2", node_LoopTick_2)
b.add_node("Nop_3", node_Nop_3)
b.add_node("AgentInvoke_4", node_AgentInvoke_4)
b.add_node("AgentInvoke_5", node_AgentInvoke_5)
b.add_node("AgentInvoke_6", node_AgentInvoke_6)
b.add_edge(START, "AgentInvoke_0")
b.add_edge("AgentInvoke_0", "AgentInvoke_4")
b.add_conditional_edges("LoopHeader_1", route_LoopHeader_1, {"loop": "AgentInvoke_4", "exit": "Nop_3"})
b.add_edge("LoopTick_2", "LoopHeader_1")
b.add_edge("Nop_3", "AgentInvoke_6")
b.add_edge("AgentInvoke_4", "AgentInvoke_5")
b.add_edge("AgentInvoke_5", "LoopTick_2")
b.add_edge("AgentInvoke_6", END)
return b.compile()
INPUTS = ["question"]
OUTPUTS = ["answer"]
if __name__ == "__main__":
_rt.main(build, INPUTS, OUTPUTS)
▸blueprint.json (the portable format)
{
"id": "com.socketcat/sql-analyst",
"flow": {
"type": "sequence",
"blocks": [
{
"in": {
"question": "question"
},
"out": "plan",
"use": "planner",
"type": "agent"
},
{
"max": 3,
"type": "loop",
"until": "review.approved == true",
"blocks": [
{
"in": {
"plan": "plan",
"review": "review"
},
"out": "draft",
"use": "analyst",
"type": "agent"
},
{
"in": {
"draft": "draft"
},
"out": "review",
"use": "critic",
"type": "agent"
}
]
},
{
"in": {
"draft": "draft"
},
"out": "answer",
"use": "writer",
"type": "agent"
}
]
},
"tags": [
"sql",
"analytics",
"databases",
"evaluator-optimizer"
],
"title": "SQL Analyst",
"agents": {
"critic": {
"model": {
"hint": "reasoning"
},
"title": "Critic",
"tools": [
"db.explain"
],
"output": {
"type": "object",
"required": [
"approved"
],
"properties": {
"notes": {
"type": "string"
},
"approved": {
"type": "boolean"
}
}
},
"instructions": "Check the analyst's SQL for correctness and cost. Run explain plans. Return an object with an 'approved' boolean and specific notes. Approve only when the query is correct and reasonably cheap."
},
"writer": {
"model": {
"hint": "fast"
},
"title": "Writer",
"tools": [
"fs.write_file"
],
"instructions": "Explain the final result for someone who does not read SQL. Keep it to a short paragraph. Save the explanation to a report file."
},
"analyst": {
"model": {
"hint": "coding"
},
"title": "SQL Analyst",
"tools": [
"db.list_indices",
"db.query_dsl",
"search.web"
],
"instructions": "Write and run read-only SQL that answers the plan. Use the schema tools to check table and index names first. If a data point needs outside context, like a definition or an industry benchmark, look it up with web search before writing the query. If you receive a review, address every point and try again."
},
"planner": {
"model": {
"hint": "reasoning"
},
"title": "Planner",
"instructions": "Break the user's question into the concrete data points needed to answer it. Do not write SQL. Produce a short plan."
}
},
"estate": {
"plan": {
"type": "string",
"description": "The planner's list of data points needed to answer the question."
},
"draft": {
"type": "object",
"description": "The analyst's current SQL and result."
},
"review": {
"type": "object",
"description": "The critic's verdict, including an approved flag."
}
},
"$schema": "socketcat.dev/blueprint/v0",
"authors": [
{
"url": "https://socketcat.com",
"name": "SocketCat"
}
],
"license": "MIT",
"servers": [
{
"ref": "io.github.supabase/postgres-mcp",
"alias": "db",
"version": "^1.1.0"
},
{
"ref": "io.github.brave/brave-search-mcp",
"alias": "search"
},
{
"ref": "io.github.modelcontextprotocol/filesystem",
"alias": "fs"
}
],
"summary": "Turns a plain-English business question into a validated SQL answer with a written explanation.",
"targets": [
"langgraph",
"crewai",
"*"
],
"version": "1.2.0",
"interface": {
"inputs": {
"question": {
"type": "string",
"description": "The business question in plain English."
}
},
"outputs": {
"answer": {
"type": "string",
"description": "A plain-language answer for a non-technical reader, including the query it ran."
}
}
},
"extensions": {
"com.crewai": {
"process": "sequential"
},
"dev.langgraph": {
"checkpointer": "memory"
}
},
"description": "A four-agent workflow. A planner breaks the question down, an analyst writes and runs SQL against a Postgres server, a critic checks the SQL for correctness and cost, and a writer explains the result for a non-technical reader. The analyst and critic loop until the critic signs off. Swap the database server for your own and it keeps working."
}More in DevOps & data
Browse all →Pipeline Firefighter
com.socketcat/pipeline-firefighter
LangGraph
A failed data job triggers three probes in parallel, then a fixer and verifier loop on a patch until it passes a dry run — and open a PR.
★ 14254 servers
⑂ 208