一个生产级的企业金融分析系统，集成了 Anthropic 模型上下文协议（MCP） 与 自主推理代理。该系统通过标准 JSON-RPC 2.0 stdio 传输，将多表关系型金融数据仓库安全地暴露给大型语言模型（LLM）。
该系统具备智能 5 层纵深防御架构、纯 Python SQL AST 词法分析器与递归下降解析器、EXPLAIN 执行计划性能分析器、带有操作码执行超时的线程安全连接池，以及 自主 Agent 自愈循环，能够自动从 SQL 语法错误、AST 安全违规和笛卡尔积连接警告中恢复。
┌─────────────────────────────────────────────────────────────────────────────┐
│ Stakeholder / User Prompt │
│ ("Identify branches with elevated 60+ delinquency") │
└──────────────────────────────────────┬──────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ Autonomous Agent Runner (agent/client_runner.py) │
│ - Multi-Turn Tool-Calling Loop (Anthropic Claude API / MockClaudeClient) │
│ - Schema-First Reflection & Planning │
│ - Closed-Loop Self-Correction & Query Repair Engine (Max Turns: 5) │
└──────────────────────────────────────┬──────────────────────────────────────┘
│ JSON-RPC 2.0 (stdio)
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ MCP Server Engine (agent/server.py) │
│ ┌───────────────────────────────────────────────────────────────────────┐ │
│ │ Methods: initialize, ping, tools/list, tools/call, resources, prompts │ │
│ └───────────────────────────────────┬───────────────────────────────────┘ │
│ │ │
│ ┌───────────────────────────────┼───────────────────────────────┐ │
│ ▼ ▼ ▼ │
│ query_database explain_query get_database_ │
│ (query_financial_lakehouse) schema │
└──────┬───────────────────────────────┬───────────────────────────────┬──────┘
│ │ │
▼ ▼ ▼
┌─────────────────────────┐ ┌─────────────────────────┐ ┌─────────────────────┐
│ Layer 1: AST Gate │ │ Layer 2: Plan Analyzer │ │ Layer 3: Connection │
│ (agent/ast_validator.py)│ │(agent/explain_analyzer) │ │ Pool & Sandboxing │
│ - Pure Python Lexer │ │ - Cost Scoring (0-100) │ │ (agent/db_engine.py)│
│ - Recursive AST Parser │ │ - Full Scan Detection │ │ - URI mode=ro │
│ - 100% Non-DQL Block │ │ - Cartesian Join Flag │ │ - sqlite authorizer │
│ - Injection Defense │ │ - Index Tuning Advice │ │ - Opcode Timeouts │
└────────────┬────────────┘ └────────────┬────────────┘ └──────────┬──────────┘
│ │ │
└───────────────────────────┼─────────────────────────┘
▼
┌───────────────────────────────────────────────┐
│ Financial Data Warehouse (data/warehouse.db) │
│ - 6 Relational Tables & Composite Indexes │
│ - branches, customers, credit_ratings, │
│ loans, repayments, audit_log │
└───────────────────────────────────────────────┘Related MCP server: SQL Server MCP
系统在 LLM 与数据库引擎之间，通过 5 个独立层级强制执行严格的安全边界：
层级 | 组件 | 安全机制 | 缓解的威胁向量 |
第 1 层：执行前 AST 门禁 |
| 纯 Python 词法分析器与递归下降解析器，验证单语句 DQL（ | 堆叠查询注入（ |
第 2 层：执行前成本门禁 |
| 评估 SQLite | 笛卡尔积（$O(N \times M)$ 连接）、无界扫描、临时 B 树导致的内存耗尽。 |
第 3 层：操作系统与引擎只读模式 |
| 使用 URI | 未经授权的磁盘写入尝试、schema 篡改。 |
第 4 层：运行时授权回调 |
|
| 绕过尝试，如 |
第 5 层：资源与内存护栏 |
| 操作码进度处理器（ | 失控的递归 CTE、CPU 拒绝服务攻击、无界结果集导致的进程内存耗尽崩溃。 |
agent/ast_validator.py）AST 安全门禁实现了双模式引擎：
零依赖纯 Python 词法加分析器与递归下降解析器：使用 Python 标准库构建，支持完整的坐标跟踪（行/列）。
可选的 sqlglot 引擎：若已安装 sqlglot，方将自动启用支持方言的解析器。
单语句 DQL：SELECT 和 WITH [RECURSIVE] ... SELECT。
公共表表达式（CTE）：单个和多个链式 CTE。解析器递归遍历 CTE 定义，确保其中不包含 DML。
窗口函数：OVER (PARTITION BY ... ORDER BY ... [ROWS/RANGE ...])、ROW_NUMBER()、RANK()、SUM() OVER ()。
多表连接：INNER JOIN、LEFT OUTER JOIN、CROSS JOIN、NATURAL JOIN，支持 ON 和 USING (...)。
子查询：FROM 子句中的子查询、SELECT 中的标量子查询、IN (SELECT ...)、EXISTS (SELECT ...)。
复合集合操作：UNION [ALL]、INTERSECT、EXCEPT。
标量表达式：CASE WHEN ... THEN ... ELSE ... END、CAST(... AS ...)、字符串拼接（||）、算术运算。
DDL：DROP、CREATE、ALTER、TRUNCATE。
DML：INSERT、UPDATE、DELETE、REPLACE、UPSERT、MERGE。
管理命令：PRAGMA、ATTACH、DETACH、VACUUM、REINDEX、ANALYZE、BEGIN、COMMIT。
危险函数：load_extension、readfile、writefile、edit、fts3_tokenizer、eval、randomblob。
系统表：sqlite_master、sqlite_schema、sqlite_temp_master、sqlite_temp_schema、sqlite_sequence、sqlite_stat*。
注入向量：多语句分号（;）、未终止的块注释（/* ...）、未终止的字符串字面量。
agent/explain_analyzer.py)解析 SQLite 的 EXPLAIN QUERY PLAN 树，支持 SQLite 3.24+ 4 列格式 (id, parent, notused, detail) 及旧版格式。
$$\text{CostScore} = \min\left(100, ; \合计 \text{罚分} \right)$$
操作详情 | 分类 | 严重程度 | Penalty |
| 未索引的全表扫描 | 高 | +25.0 每次 |
| 临时索引构建 | 高 | +20.0 |
| 未索引排序 | 中 | +15.0 |
| 临时聚合 B 树 | 中 | +10.0 |
| 物化子查询 | 中 | +10.0 每个 |
多表无索引扫描 | 笛卡尔积连接 | 严重 | +30.0 |
$0.0 - 25.0$（最佳）：完全索引的数值点/范围其查找。即时执行。
$26.0 - 50.0$（可接受）：轻度临时排序或单个小表扫描。
$51.0 - 74.0$（警告）：非最优执行计划；多次扫描。
$75.0 - 100.0$（严重）：笛卡尔积或大规模无索引连接。将被 MCP 执行门阻断。
data/schema.sql)该数据仓库建模了一个车辆资产融资领域业务，包含 6 张关系表：
┌──────────────┐ 1:N ┌──────────────┐ 1:N ┌──────────────┐
│ branches ├────────────────►│ customers ├────────────────►│credit_ratings│
└──────┬───────┘ └──────┬───────┘ └──────────────┘
│ 1:N │ 1:N
│ ┌──────────────┐ │
└────────►│ loans │◄──────┘
└──────┬───────┘
│ 1:N
┌──────▼───────┐
│ repayments │
└──────────────┘
┌──────────────┐
│ audit_log │ (Immutable lifecycle state transition log)
└──────────────┘branches：12 个区域枢纽和零售分支机构，具有递归父子层级（parent_branch_id）。
customers：300 份借款人档案，包含对数正态收入分布、负债收入比以及将权利息身份的 SHA-256 PII 哈希值。
credit_ratings：横跨 5 个风险档位（PRIME_PLUS 至 DEEP_SUBPRIME）的 600 多个纵向信用局评分快照。
loans：500 份汽车金融与 SME 贷款合同，包含风险调整利率及月度摊销还款额。
repayments：17,000 多条交易流水账目，包含本金/利息/费用明细与逾期跟踪。
audit_log：不可篡改的审计记录，跟踪贷款状态转变为 DELINQUENT_90、DEFAULTED 和 WRITE_OFF 的过程。
服务器（agent/server.py）暴露 4 个核心工具：
query_database（别名：query_financial_lakehouse)执行安全的只读 SQL 查询，自动进行查询前 AST 验证、操作码进度超时控制以及限制行数。
输入：query（字符串，必填）、max_rows（整数，默认：100）、timeout_seconds（浮点数，默认：5.0）。
输出：JSON 负载，包含 columns、rows、row_count、is_truncated、execution_time_ms。
explain_query检查执行计划的各个节点，计算成本评分（$0-100$），检测全表扫描情况，并在不执行任何修改操作的条件下提供索引建议。
输入：query（字符串，必填）。
输出：cost_score、complexity_rating、scanned_tables、indexed_tables、warnings、recommendations。
get_database_schema获取数据库的目录元数据、列类型、主键、外键以及索引等信息。
输入：table_name（字符串，可选）。
输出：完整或筛选后的表 Schema 定义。
validate_sql_safety在不访问数据库的情况下执行静态 AST 安全分析。
输入：query（字符串，必填）。
输出：is_safe（布尔值）、statement_type、referenced_tables、detected_risks。
agent/client_runner.py）AutonomousDataAgent 实现了一个迭代式工具调用循环，具备闭环错误修复能力：
┌─────────────────────────────────────┐
│ User: "Top 5 default risk branches" │
└──────────────────┬──────────────────┘
│
▼
┌─────────────────────────────────────┐
│ Turn 1: Introspect Database Schema │
└──────────────────┬──────────────────┘
│
▼
┌─────────────────────────────────────┐
│ Turn 2: Synthesize & Explain Plan │
└──────────┬──────────────────────┬───┘
│ │
Plan Warning / ▼ ▼ Pass
Cartesian Join ┌──────────────────┐ ┌──────────────────┐
│ 🔄 Repair Query │ │ Turn 3: Execute │
│ (Add JOIN ... ON)│ │ query_database │
└────────┬─────────┘ └────────┬─────────┘
│ │
▼ ▼
┌──────────────────┐ ┌──────────────────┐
│ Turn 4: Re-check │ │ Synthesize Final │
│ & Run Query │ │ Executive Report │
└──────────────────┘ └──────────────────┘SQLite 语法 / Schema 错误（例如列名拼写错误）：注入包含 Schema 目录的 SYNTAX_ERROR_TEMPLATE；Agent 修复列名。
AST 安全拒绝（例如非 DQL 查询）：注入 AST_VIOLATION_TEMPLATE；Agent 重新构建一条合规的单条 SELECT 查询。
高查询成本 / 笛卡尔连接：注入 PLAN_WARNING_TEMPLATE；Agent 补充索引连接谓词。
确定性离线执行：MockClaudeClient 可在无需 Anthropic API 密钥的情况下实现 100% 离线测试。
# Clone and navigate to repository
cd MCP_Autonomous_Agent
# Install dependencies
pip install -r requirements.txt使用确定性的合成金融数据（固定种子 42）填充 data/warehouse.db：
python data/seed_warehouse.py输出：
[SeedWarehouse] branches : 12 rows
[SeedWarehouse] customers : 300 rows
[SeedWarehouse] credit_ratings : 627 rows
[SeedWarehouse] loans : 500 rows
[SeedWarehouse] repayments : 17120 rows
[SeedWarehouse] audit_log : 44 rows
[SeedWarehouse] Database seeding successfully completed.执行全部 66 项单元测试和集成测试，覆盖 AST 验证、EXPLAIN 分析、数据库引擎线程安全、MCP 工具以及 Agent 自愈循环：
python -m unittest discover -s tests -v针对金融数据仓库执行一次多轮分析型查询会话：
from agent.client_runner import AutonomousDataAgent
agent = AutonomousDataAgent()
response = agent.run("Identify the top default risk branches with delinquency counts and total exposure")
print(f"Success: {response.success}")
print(f"Turns Taken: {response.turns_taken}")
print(f"SQL Executed: {response.sql_executed}")
print(f"\n{response.final_answer}")连接 Anthropic Claude 桌面应用或 MCP Inspector：
python agent/server.py在 Claude Desktop 的 claude_desktop_config.json 中进行配置：
{
"mcpServers": {
"financial-data-agent": {
"command": "python",
"args": ["-m", "agent.server"],
"cwd": "/path/to/MCP_Autonomous_Agent"
}
}
}MCP_Autonomous_Agent/
├── data/
│ ├── __init__.py
│ ├── schema.sql # 6-table relational financial warehouse DDL
│ ├── seed_warehouse.py # Deterministic synthetic data generator (seed 42)
│ └── warehouse.db # Generated SQLite database file
├── agent/
│ ├── __init__.py
│ ├── ast_validator.py # Pure-Python SQL Lexer & Recursive Descent AST Parser
│ ├── explain_analyzer.py # SQLite EXPLAIN QUERY PLAN analyzer & cost scorer
│ ├── db_engine.py # Thread-safe read-only connection pool & opcode timeout
│ ├── prompts.py # System prompts, tool schemas & remediation templates
│ ├── client_runner.py # Autonomous agent loop with closed-loop self-correction
│ └── server.py # MCP JSON-RPC 2.0 stdio server implementation
├── tests/
│ ├── __init__.py
│ ├── test_ast_validator.py # Unit tests for AST security and analytical DQL (29 tests)
│ ├── test_explain_analyzer.py# Unit tests for plan parsing, scans, cartesian (7 tests)
│ ├── test_db_engine.py # Unit tests for read-only pool, timeouts, threads (8 tests)
│ ├── test_mcp_tools.py # Unit tests for MCP protocol, tool calls, errors (15 tests)
│ └── test_client_runner.py # Unit tests for agent loop and self-healing (5 tests)
├── requirements.txt # Dependency specification (mcp, anthropic, sqlglot, pytest)
└── README.md # Complete architectural & technical documentationMIT 许可证。专为企业金融数据分析和 AI Agent 作品集演示而创建。
This server cannot be deployed
Query your org's data in natural language — read-only MCP access to SQL, NoSQL, files & warehouses.
Ask questions in plain language, get answers from your business database. No SQL required.
Ask business questions in plain English. Get instant answers from your database, no SQL needed.
Query OneLens cloud-cost data in natural language: breakdowns, trends, cost centers. Read-only.
