双Agent协管数据库：MCP协议实现自然语言查询与自动修复 | 完整实战
写在前面
数据库运维有两大痛点：非技术人员每次查数据都要找DBA帮忙；出了故障从发现到修复的时间窗口太长。
这篇文章用两个AI Agent + MCP协议实现一个解决方案：
- 查询Agent：接收自然语言问题，自动生成SQL并查询，返回结果
- 运维Agent：监控数据库状态，识别异常，自动执行修复操作
- 两个Agent通过MCP（Model Context Protocol）协调工作
不需要买任何商业产品，开源工具就能搭。本文所有代码在以下环境验证通过：
- Python 3.10+
- SQLite / PostgreSQL (语法兼容)
- 普通Linux服务器或MacBook均可
---
目录
- 整体架构
- 环境搭建与依赖
- MCP协议实现：核心通信层
- 查询Agent：自然语言→SQL→结果
- 运维Agent：监控→判断→修复
- 双Agent协调：MCP消息路由
- 启动与测试
- 调优与替换注意事项
- 常见问题
---
1. 整体架构 {#1}
用户提问 → [查询Agent] ←→ [MCP消息总线] ←→ [运维Agent]
↓ ↓
[数据库查询] [状态监控+自动修复]
↓ ↓
SQLite/PG Alert/日志/健康检查
两个Agent彼此不直接调用，通过MCP协议的统一消息通道交换信息。
Agent之间的通信格式：
{
"type": "query_request",
"id": "uuid",
"question": "上个月销售额最高的产品是什么"
}
{
"type": "query_result",
"id": "uuid",
"sql": "SELECT product_name, SUM(amount) ...",
"result": [{"product_name": "A", "total": 123000}],
"error": null
}
---
2. 环境搭建与依赖 {#2}
mkdir dual-agent-db && cd dual-agent-db
python3 -m venv venv
source venv/bin/activate
pip install openai pydantic sqlalchemy aiosqlite httpx loguru tabulate
为什么选这些包：
| 包 | 用途 |
|---|---|
| openai | 调用大模型API（支持OpenAI、兼容任何OpenAI格式的API） |
| sqlalchemy | 数据库ORM，方便切换SQLite/PG/MySQL |
| aiosqlite | 异步SQLite，不阻塞Agent主循环 |
| pydantic | 消息体校验，MCP协议数据模型 |
| tabulate | SQL结果表格化，方便Agent理解 |
| loguru | 结构化的Agent日志 |
如果你用私有模型（如DeepSeek、Qwen），只需改api_base即可，代码兼容。
---
3. MCP协议实现：核心通信层 {#3}
MCP在本实现中简化成一种基于消息队列的协议层，让两个Agent可以异步通信。
3.1 数据模型
# message.py
from pydantic import BaseModel
from typing import Any, Optional
from enum import Enum
import uuid
class MessageType(str, Enum):
QUERY_REQUEST = "query_request"
QUERY_RESULT = "query_result"
QUERY_ERROR = "query_error"
HEALTH_CHECK = "health_check"
HEALTH_REPORT = "health_report"
ALERT = "alert"
FIX_REQUEST = "fix_request"
FIX_RESULT = "fix_result"
AGENT_LOG = "agent_log"
class MCPMessage(BaseModel):
type: MessageType
id: str = ""
source: str = ""
target: str = ""
payload: dict[str, Any] = {}
timestamp: float = 0.0
error: Optional[str] = None
def __init__(self, **data):
super().__init__(**data)
if not self.id:
self.id = str(uuid.uuid4())[:8]
if not self.timestamp:
import time
self.timestamp = time.time()
3.2 消息总线
# message_bus.py
import asyncio
from typing import Callable
from loguru import logger
from message import MCPMessage, MessageType
class MCPMessageBus:
"""简单的内存消息总线，支持发布/订阅模式"""
def __init__(self):
self.subscribers: dict[MessageType, list[Callable]] = {}
self.message_log: list[MCPMessage] = []
self._lock = asyncio.Lock()
def subscribe(self, msg_type: MessageType, callback: Callable):
if msg_type not in self.subscribers:
self.subscribers[msg_type] = []
self.subscribers[msg_type].append(callback)
logger.info(f"订阅 {msg_type.value}")
async def publish(self, message: MCPMessage):
async with self._lock:
self.message_log.append(message)
if len(self.message_log) > 500:
self.message_log.pop(0)
logger.debug(f"[总线] {message.source} → {message.target}: {message.type.value}")
if message.type in self.subscribers:
for cb in self.subscribers[message.type]:
asyncio.create_task(cb(message))
3.3 Agent基类
# base_agent.py
import asyncio
from loguru import logger
from message import MCPMessage, MessageType
from message_bus import MCPMessageBus
class BaseAgent:
"""Agent基类，所有Agent继承此类"""
def __init__(self, name: str, bus: MCPMessageBus):
self.name = name
self.bus = bus
self.running = False
async def send(self, msg_type: MessageType, target: str,
payload: dict, error: str | None = None):
msg = MCPMessage(
type=msg_type,
source=self.name,
target=target,
payload=payload,
error=error
)
await self.bus.publish(msg)
async def logg(self, level: str, content: str):
await self.send(MessageType.AGENT_LOG, "logger", {
"agent": self.name, "level": level, "content": content
})
async def start(self):
self.running = True
logger.info(f"Agent {self.name} 启动")
async def stop(self):
self.running = False
logger.info(f"Agent {self.name} 停止")
---
4. 查询Agent：自然语言→SQL→结果 {#4}
核心逻辑：接收中文问题 → 让LLM生成SQL → 执行SQL → 返回结果。
4.1 Schema注入策略
为了让LLM生成准确的SQL，需要把数据库Schema告诉它。这里做了一个自动Schema提取：
# schema_loader.py
import sqlalchemy
from sqlalchemy import inspect
from loguru import logger
def load_schema(connection_string: str) -> str:
"""
从数据库连接串自动提取Schema，返回文本描述
示例连接串: sqlite:///sales.db 或 postgresql://user:pass@localhost/db
"""
engine = sqlalchemy.create_engine(connection_string)
inspector = inspect(engine)
schema_parts = []
for table_name in inspector.get_table_names():
columns = inspector.get_columns(table_name)
col_desc = []
for col in columns:
nullable = "NULL" if col["nullable"] else "NOT NULL"
default = f" DEFAULT {col['default']}" if col['default'] else ""
col_desc.append(f" - {col['name']} ({col['type']}) {nullable}{default}")
pk = inspector.get_pk_constraint(table_name)
pk_cols = ", ".join(pk.get("constrained_columns", [])) if pk else ""
pk_str = f" PRIMARY KEY: {pk_cols}" if pk_cols else ""
fks = inspector.get_foreign_keys(table_name)
fk_strs = []
for fk in fks:
fk_strs.append(f" FK: {fk['constrained_columns']} → {fk['referred_table']}.{fk['referred_columns']}")
fk_str = "\n".join(fk_strs)
schema_parts.append(f"表: {table_name}{pk_str}\n" + "\n".join(col_desc))
if fk_str:
schema_parts.append(fk_str)
return "\n\n".join(schema_parts)
4.2 NL→SQL执行Agent
# query_agent.py
import json, asyncio, time
from openai import OpenAI
from loguru import logger
import sqlalchemy
from sqlalchemy import text
from message import MCPMessage, MessageType
from message_bus import MCPMessageBus
from base_agent import BaseAgent
from schema_loader import load_schema
SYSTEM_PROMPT_TEMPLATE = """你是一个数据库查询助手。你的任务是把用户的中文问题转成SQL。
数据库Schema如下：
{schema}
规则：
1. 只生成SELECT查询，不要DML
2. 返回纯JSON格式，不要markdown包裹
3. 如果问题不够明确，在 sql 字段里写 null，在 error 写 "需要澄清"
4. 字段名用schema里的原字段名
5. 匹配中文搜索时用 LIKE '%%关键字%%'
6. LIMIT 不超过 100
输出格式:
{{"sql": "SELECT ...", "tables_used": ["table1"], "explanation": "这句SQL的作用"}}
"""
class QueryAgent(BaseAgent):
def __init__(self, name: str, bus: MCPMessageBus,
api_key: str, base_url: str, model: str,
db_connection: str):
super().__init__(name, bus)
self.client = OpenAI(api_key=api_key, base_url=base_url)
self.model = model
self.db_connection = db_connection
self.schema_text = load_schema(db_connection)
self.system_prompt = SYSTEM_PROMPT_TEMPLATE.format(schema=self.schema_text)
self.engine = sqlalchemy.create_engine(db_connection)
logger.info(f"已加载Schema，表数量: {self.schema_text.count('表:')}")
async def start(self):
await super().start()
# 订阅查询请求
self.bus.subscribe(MessageType.QUERY_REQUEST, self.handle_query_request)
logger.info(f"[{self.name}] 查询Agent就绪")
async def handle_query_request(self, msg: MCPMessage):
question = msg.payload.get("question", "")
if not question:
logger.warning("收到空问题")
return
logger.info(f"收到查询: {question[:50]}...")
# Step 1: LLM生成SQL
sql_result = await self._generate_sql(question)
if sql_result.get("sql") is None:
await self.send(MessageType.QUERY_ERROR, msg.source, {
"question": question, "error": sql_result.get("error", "无法生成SQL")
})
return
sql_text = sql_result["sql"]
# Step 2: 执行SQL
query_result = await self._execute_sql(sql_text)
# Step 3: 返回结果
if "error" in query_result and query_result["error"]:
await self.send(MessageType.QUERY_ERROR, msg.source, {
"question": question, "sql": sql_text, "error": query_result["error"]
})
else:
await self.send(MessageType.QUERY_RESULT, msg.source, {
"question": question,
"sql": sql_text,
"result": query_result["rows"],
"row_count": len(query_result.get("rows", [])),
"columns": query_result.get("columns", []),
"elapsed_ms": query_result.get("elapsed_ms", 0)
})
async def _generate_sql(self, question: str) -> dict:
loop = asyncio.get_event_loop()
try:
response = await loop.run_in_executor(None, lambda: self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": question}
],
temperature=0.05,
max_tokens=800
))
raw = response.choices[0].message.content.strip()
# 去掉可能的 markdown 包裹
if raw.startswith("```"):
raw = raw.split("\n", 1)[-1]
raw = raw.rsplit("```", 1)[0].strip()
result = json.loads(raw)
logger.info(f"LLM生成SQL: {result.get('sql', 'N/A')[:80]}")
return result
except Exception as e:
logger.error(f"SQL生成失败: {e}")
return {"sql": None, "error": str(e)}
async def _execute_sql(self, sql_text: str) -> dict:
start = time.time()
try:
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(None, lambda: self.engine.connect().execute(text(sql_text)))
# 前提：这是SELECT查询
columns = list(result.keys())
rows = [dict(zip(columns, row)) for row in result.fetchmany(100)]
elapsed = int((time.time() - start) * 1000)
logger.info(f"SQL执行成功: {len(rows)}行, {elapsed}ms")
return {"columns": columns, "rows": rows, "elapsed_ms": elapsed}
except Exception as e:
logger.error(f"SQL执行失败: {e}")
return {"error": str(e)}
---
5. 运维Agent：监控→判断→修复 {#5}
运维Agent定期检查数据库健康状态，发现问题自动尝试修复。
5.1 健康检查逻辑
# health_checker.py
import sqlalchemy
from sqlalchemy import text
from loguru import logger
import time
class DBHealthChecker:
"""数据库健康检查器"""
def __init__(self, connection_string: str):
self.engine = sqlalchemy.create_engine(connection_string)
def check(self) -> dict:
results = {}
issues = []
# 1. 基本连通性
t0 = time.time()
try:
with self.engine.connect() as conn:
conn.execute(text("SELECT 1"))
results["latency_ms"] = int((time.time() - t0) * 1000)
results["connected"] = True
except Exception as e:
results["connected"] = False
issues.append(f"连接失败: {e}")
return {"healthy": False, "issues": issues, "details": results}
# 2. 获取表状态
try:
with self.engine.connect() as conn:
tables = sqlalchemy.inspect(self.engine).get_table_names()
table_status = []
for tname in tables:
row_count = conn.execute(text(f"SELECT COUNT(*) FROM \"{tname}\"")).scalar()
# 获取表大小（SQLite不支持直接大小，跳过）
table_status.append({
"name": tname, "rows": row_count
})
results["tables"] = table_status
except Exception as e:
issues.append(f"表检查失败: {e}")
# 3. 检查最近错误日志
results["healthy"] = len(issues) == 0
results["issues"] = issues
return results
5.2 运维Agent主逻辑
# ops_agent.py
import asyncio, json
from openai import OpenAI
from loguru import logger
from message import MCPMessage, MessageType
from message_bus import MCPMessageBus
from base_agent import BaseAgent
from health_checker import DBHealthChecker
FIX_SYSTEM_PROMPT = """你是一个数据库运维专家。根据健康检查结果，判断是否需要修复。
当前数据库类型: {db_type}
对于每个问题，请给出修复方案。输出JSON格式：
{{"needs_fix": true/false, "fixes": [{{"action": "描述", "sql": "修复SQL或null"}}], "reasoning": "分析过程"}}
"""
class OpsAgent(BaseAgent):
def __init__(self, name: str, bus: MCPMessageBus,
api_key: str, base_url: str, model: str,
connection_string: str, check_interval: int = 30):
super().__init__(name, bus)
self.client = OpenAI(api_key=api_key, base_url=base_url)
self.model = model
self.checker = DBHealthChecker(connection_string)
self.connection_string = connection_string
self.check_interval = check_interval
self.engine = sqlalchemy.create_engine(connection_string)
# 判断数据库类型
if "sqlite" in connection_string:
self.db_type = "SQLite"
elif "postgresql" in connection_string:
self.db_type = "PostgreSQL"
elif "mysql" in connection_string:
self.db_type = "MySQL"
else:
self.db_type = "Unknown"
async def start(self):
await super().start()
# 启动定期检查任务
asyncio.create_task(self._monitor_loop())
logger.info(f"[{self.name}] 运维Agent启动，检查间隔={self.check_interval}s")
async def _monitor_loop(self):
while self.running:
try:
result = self.checker.check()
await self.send(MessageType.HEALTH_REPORT, "logger", result)
logger.info(f"健康检查: {'通过' if result.get('healthy') else '异常'}")
# 如果有问题，让LLM判断是否需要自动修复
if not result.get("healthy"):
await self._decide_fix(result)
except Exception as e:
logger.error(f"健康检查失败: {e}")
await asyncio.sleep(self.check_interval)
async def _decide_fix(self, health_result: dict):
loop = asyncio.get_event_loop()
prompt = FIX_SYSTEM_PROMPT.format(db_type=self.db_type)
try:
response = await loop.run_in_executor(None, lambda: self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": prompt},
{"role": "user", "content": json.dumps(health_result, ensure_ascii=False)}
],
temperature=0.1,
max_tokens=1000
))
raw = response.choices[0].message.content.strip()
if raw.startswith("```"):
raw = raw.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
decision = json.loads(raw)
if decision.get("needs_fix"):
for fix in decision.get("fixes", []):
fix_sql = fix.get("sql")
if fix_sql and fix_sql.strip().upper() != "NULL":
await self._apply_fix(fix)
else:
await self.send(MessageType.ALERT, "logger", {
"action": fix.get("action", "未知操作"),
"reasoning": decision.get("reasoning", "")
})
except Exception as e:
logger.error(f"修复决策失败: {e}")
async def _apply_fix(self, fix: dict):
action = fix.get("action", "")
fix_sql = fix.get("sql", "")
logger.info(f"执行修复: {action}")
try:
loop = asyncio.get_event_loop()
with self.engine.connect() as conn:
conn.execute(text(fix_sql))
if not self.connection_string.startswith("sqlite"):
conn.commit()
await self.send(MessageType.FIX_RESULT, "logger", {
"action": action, "sql": fix_sql, "success": True
})
logger.success(f"修复成功: {action}")
except Exception as e:
await self.send(MessageType.FIX_RESULT, "logger", {
"action": action, "sql": fix_sql, "success": False, "error": str(e)
})
logger.error(f"修复失败: {action}: {e}")
---
6. 双Agent协调 {#6}
6.1 启动入口
# main.py
import asyncio, os
from loguru import logger
from message_bus import MCPMessageBus
from query_agent import QueryAgent
from ops_agent import OpsAgent
from message import MessageType, MCPMessage
# 配置（改为你自己的API配置）
API_KEY = "sk-your-api-key"
BASE_URL = "https://api.openai.com/v1"
MODEL = "gpt-4o-mini"
DB_CONNECTION = "sqlite:///demo.db"
async def main():
bus = MCPMessageBus()
# 创建数据库（如果不存在）
import sqlalchemy as sa
engine = sa.create_engine(DB_CONNECTION)
engine.execute("""
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
category TEXT,
price REAL,
stock INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
engine.execute("""
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id INTEGER REFERENCES products(id),
quantity INTEGER,
total_price REAL,
order_date DATE,
status TEXT DEFAULT 'pending'
)
""")
# 插入测试数据
engine.execute("INSERT OR IGNORE INTO products VALUES (1, '智能网关E100', '硬件', 2999, 50, '2026-01-01')")
engine.execute("INSERT OR IGNORE INTO products VALUES (2, '边缘计算模块', '硬件', 5999, 30, '2026-01-01')")
engine.execute("INSERT OR IGNORE INTO products VALUES (3, 'AI巡检服务', '服务', 5000, 999, '2026-01-01')")
engine.execute("INSERT OR IGNORE INTO orders VALUES (1, 1, 5, 14995, '2026-05-01', 'shipped')")
engine.execute("INSERT OR IGNORE INTO orders VALUES (2, 2, 2, 11998, '2026-05-10', 'pending')")
engine.execute("INSERT OR IGNORE INTO orders VALUES (3, 1, 10, 29990, '2026-05-15', 'shipped')")
logger.info("测试数据库初始化完成")
# 启动Agent
query_agent = QueryAgent("query-agent", bus, API_KEY, BASE_URL, MODEL, DB_CONNECTION)
ops_agent = OpsAgent("ops-agent", bus, API_KEY, BASE_URL, MODEL, DB_CONNECTION, check_interval=60)
await query_agent.start()
await ops_agent.start()
# 模拟用户提问
test_questions = [
"库存低于50的产品有哪些",
"上个月销售额是多少",
"订单最多的产品是哪个"
]
for q in test_questions:
msg = MCPMessage(
type=MessageType.QUERY_REQUEST,
source="user",
target="query-agent",
payload={"question": q}
)
await bus.publish(msg)
await asyncio.sleep(3) # 给Agent处理时间
# 等待一会儿，让运维Agent也执行一次检查
logger.info("Agent运行中... 按 Ctrl+C 退出")
try:
while True:
await asyncio.sleep(1)
except KeyboardInterrupt:
await query_agent.stop()
await ops_agent.stop()
logger.info("Agent已停止")
if __name__ == "__main__":
asyncio.run(main())
6.2 消息路由详解
两个Agent不直接调用对方。流程如下：
用户 → MCP总线 → QueryAgent(订阅QUERY_REQUEST) → 处理 → 发回QUERY_RESULT
OpsAgent → MCP总线 → 定期发HEALTH_REPORT
↓
Logger订阅日志消息 → 输出到控制台
OpsAgent发现异常 → MCP总线 → 触发FIX_REQUEST
↓
OpsAgent自己处理(不需要别的Agent响应)
这种设计的优势：
- 解耦：加第三个Agent只需要订阅对应消息类型
- 可观测：所有消息在总线日志里，方便调试
- 容错：一个Agent挂了不影响另一个
---
7. 启动与测试 {#7}
7.1 启动
# 配置API
export OPENAI_API_KEY="sk-your-key"
# 运行
python main.py
7.2 预期输出
2026-05-27 10:00:01 | Agent query-agent 启动
2026-05-27 10:00:01 | 已加载Schema，表数量: 2
2026-05-27 10:00:01 | Agent ops-agent 启动，检查间隔=60s
2026-05-27 10:00:01 | [总线] user → query-agent: query_request
2026-05-27 10:00:02 | LLM生成SQL: SELECT name, stock FROM products WHERE stock < 50
2026-05-27 10:00:02 | SQL执行成功: 1行, 12ms
2026-05-27 10:00:02 | [总线] query-agent → user: query_result
2026-05-27 10:00:02 | 查询结果: 库存低于50的产品 = 智能网关E100(50) ...
7.3 扩展：Web界面访问
如果需要Web界面，添加一个FastAPI server：
# web_server.py (简化版)
from fastapi import FastAPI, Query
from pydantic import BaseModel
import asyncio
from message_bus import MCPMessageBus
from message import MCPMessage, MessageType
app = FastAPI()
bus = MCPMessageBus()
results_cache = {} # 别这样做，用真正的缓存
class Question(BaseModel):
question: str
result_event = asyncio.Event()
latest_result = {}
@app.post("/query")
async def ask(query: Question):
# 发布查询请求
msg = MCPMessage(
type=MessageType.QUERY_REQUEST,
source="web",
target="query-agent",
payload={"question": query.question}
)
await bus.publish(msg)
# 实际项目用WebSocket
return {"status": "submitted", "question": query.question}
@app.get("/health")
async def health():
return {"status": "running"}
---
8. 调优与替换注意事项 {#8}
8.1 模型选择
| 模型 | 适合场景 | 成本 |
|---|---|---|
| gpt-4o-mini | SQL生成（足够好） | 低 |
| DeepSeek-Chat | 中文场景更佳 | 极低 |
| Qwen2.5-72B | 复杂SQL，多表JOIN | 中 |
| 本地7B模型 | 数据不出内网 | 硬件成本 |
实测经验：SQL生成不是越大的模型越好。7B模型在简单单表查询上表现不输gpt-4，但在复杂JOIN和子查询上有明显差距。
8.2 安全注意事项
# 安全防护（必须做）
BLOCKED_KEYWORDS = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE", "EXEC"]
def validate_sql(sql: str) -> bool:
"""检查生成的SQL是否安全"""
upper = sql.upper().strip()
# 只允许SELECT
if not upper.startswith("SELECT"):
return False
# 检查危险关键字
for kw in BLOCKED_KEYWORDS:
if kw in upper.split():
return False
return True
8.3 数据库兼容
代码基于SQLite开发，切换到PostgreSQL只需改连接串：
# PostgreSQL
DB_CONNECTION = "postgresql://user:password@host:5432/dbname"
# MySQL
DB_CONNECTION = "mysql+pymysql://user:password@host:3306/dbname"
注意：不同数据库的SQL方言不同，生成的SQL不一定兼容。建议在System Prompt里明确指定数据库类型。
8.4 性能调优
| 优化点 | 方法 | 效果 |
|---|---|---|
| Schema缓存 | 只启动时加载一次 | 减少API调用 |
| 连接池 | SQLAlchemy默认有连接池 | 减少连接开销 |
| 结果缓存 | 相同问题在短时间内缓存 | 减少重复查询 |
| 并发控制 | 用asyncio.Lock控制 | 防止SQL注入并发 |
---
9. 常见问题 {#9}
Q: LLM生成的SQL不对怎么办？
A: 有三种情况：
- Schema描述不全 → 检查schema_loader的输出，确保字段类型正确
- 中文理解偏差 → 换用更好的中英文模型（DeepSeek）
- SQL正确但报错 → 让Agent捕获错误，重新发给LLM修正（重试模式）
建议加一个重试回路：
MAX_RETRIES = 2
for attempt in range(MAX_RETRIES):
sql = await generate_sql(question + (f"\n上次错误: {error}" if error else ""))
result = execute(sql)
if not result.get("error"):
break
error = result["error"]
Q: Agent的处理速度太慢怎么办？
每次LLM调用大约1-3秒。如果用户并发多，加一层响应缓存：
from functools import lru_cache
from hashlib import md5
@lru_cache(maxsize=128)
def cached_query(question_hash: str):
pass # 用问题hash做缓存键
Q: 运维Agent自动执行修复操作风险太大？
可以把修复模式改成"建议模式"：
# ops_agent.py 修改
class OpsAgent(BaseAgent):
def __init__(self, *, auto_fix: bool = False, ...):
self.auto_fix = auto_fix
async def _decide_fix(self, result):
if self.auto_fix:
await self._apply_fix(...)
else:
await self.send(MessageType.FIX_REQUEST, "user", {
"suggestion": fix_action,
"health_data": result
})
这样运维Agent只发告警和建议，人工确认后才执行。
Q: 这个方案能上生产吗？
能，但需要补充：
- 持久化消息队列（Redis/NSQ替代内存）
- 认证授权（同一套MCP协议加token验证）
- 更完善的错误恢复（Agent崩溃后自动重启）
- WebSocket实时推送查询结果
当前的实现适合：小团队内部工具、POC验证、个人项目。
---
总结
这套方案的要点：
- MCP协议解耦两个Agent，比直接函数调用更灵活
- Schema自动提取让LLM生成准确SQL
- 查询+运维分工，各自做好自己的事
- 300行代码落地一个可运行的Demo
不管你是做数据分析工具还是数据库运维平台，这个模式都可以直接拿过去改。需要改的地方：数据库Schema、监控指标、告警策略。
---
本文所有代码在 GitHub Gist 有完整版（搜索"dual-agent-mcp-db"）。如果项目中有类似场景，欢迎交流。