de en es ja ko ru zh

MCP Autonomous Data Agent

Name: MCP Autonomous Data Agent
Author: fortuneMog

by fortuneMog

Overview Schema Related Servers Score Discussions

Python

Local

Anthropic Claude API 与 MCP 自主数据代理

Python Version Protocol Test Suite Security

一个生产级的企业金融分析系统，集成了 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               │
                 └───────────────────────────────────────────────┘

5层纵深防御安全模型

系统在 LLM 与数据库引擎之间，通过 5 个独立层级强制执行严格的安全边界：

层级	组件	安全机制	缓解的威胁向量
第 1 层：执行前 AST 门禁	`agent/ast_validator.py`	纯 Python 词法分析器与递归下降解析器，验证单语句 DQL（`SELECT`、`WITH ... SELECT`）。	堆叠查询注入（`;`）、DDL（`DROP`、`ALTER`、`CREATE`）、DML（`INSERT`、`UPDATE`、`DELETE`）、PRAGMA 侦察、注释攻击。
第 2 层：执行前成本门禁	`agent/explain_analyzer.py`	评估 SQLite `EXPLAIN QUERY PLAN`，计算综合成本评分（$0-100$）。	笛卡尔积（$O(N \times M)$ 连接）、无界扫描、临时 B 树导致的内存耗尽。
第 3 层：操作系统与引擎只读模式	`agent/db_engine.py`	使用 URI `file:<path>?mode=ro` 建立 SQLite 连接。	未经授权的磁盘写入尝试、schema 篡改。
第 4 层：运行时授权回调	`agent/db_engine.py`	`sqlite3.set_authorizer` 将操作限制为 `SQLITE_SELECT`、`SQLITE_READ`、`SQLITE_FUNCTION`、`SQLITE_RECURSIVE` 以及安全的 schema PRAGMA。	绕过尝试，如 `ATTACH DATABASE`、`load_extension`、`PRAGMA writable_schema`、表修改。
第 5 层：资源与内存护栏	`agent/db_engine.py`	操作码进度处理器（`conn.set_progress_handler`）监控查询执行时间 + `fetchmany(max_rows + 1)` 行截断。	失控的递归 CTE、CPU 拒绝服务攻击、无界结果集导致的进程内存耗尽崩溃。

AST SQL 安全验证器（`agent/ast_validator.py`）

AST 安全门禁实现了双模式引擎：

零依赖纯 Python 词法加分析器与递归下降解析器：使用 Python 标准库构建，支持完整的坐标跟踪（行/列）。
可选的 sqlglot 引擎：若已安装 sqlglot，方将自动启用支持方言的解析器。

支持的分析型 SQL 语法

单语句 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 ...)、字符串拼接（||）、算术运算。

禁止模式（100% 阻止率）

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*。
注入向量：多语句分号（;）、未终止的块注释（/* ...）、未终止的字符串字面量。

EXPLAIN 执行计划分析器（`agent/explain_analyzer.py`)

解析 SQLite 的 EXPLAIN QUERY PLAN 树，支持 SQLite 3.24+ 4 列格式 (id, parent, notused, detail) 及旧版格式。

计分公式与罚分

$$\text{CostScore} = \min\left(100, ; \合计 \text{罚分} \right)$$

操作详情	分类	严重程度	Penalty
`SCAN TABLE <table>`	未索引的全表扫描	高	+25.0 每次
`SEARCH TABLE <table> USING AUTOMATIC INDEX`	临时索引构建	高	+20.0
`USE TEMP B-TREE FOR ORDER BY`	未索引排序	中	+15.0
`USE TEMP B-TREE FOR GROUP BY/DISTINCT`	临时聚合 B 树	中	+10.0
`MATERIALIZE <id>`	物化子查询	中	+10.0 每个
多表无索引扫描	笛卡尔积连接	严重	+30.0

评级分类

$0.0 - 25.0$（最佳）：完全索引的数值点/范围其查找。即时执行。
$26.0 - 50.0$（可接受）：轻度临时排序或单个小表扫描。
$51.0 - 74.0$（警告）：非最优执行计划；多次扫描。
$75.0 - 100.0$（严重）：笛卡尔积或大规模无索引连接。将被 MCP 执行门阻断。

金融数据仓库 Schema（`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 的过程。

MCP 工具与 JSON-RPC 2.0 协议接口

服务器（agent/server.py）暴露 4 个核心工具：

1. `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。

2. `explain_query`

检查执行计划的各个节点，计算成本评分（$0-100$），检测全表扫描情况，并在不执行任何修改操作的条件下提供索引建议。

输入：query（字符串，必填）。
输出：cost_score、complexity_rating、scanned_tables、indexed_tables、warnings、recommendations。

3. `get_database_schema`

获取数据库的目录元数据、列类型、主键、外键以及索引等信息。

输入：table_name（字符串，可选）。
输出：完整或筛选后的表 Schema 定义。

4. `validate_sql_safety`

在不访问数据库的情况下执行静态 AST 安全分析。

输入：query（字符串，必填）。
输出：is_safe（布尔值）、statement_type、referenced_tables、detected_risks。

自主 Agent 与自愈循环（`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% 离线测试。

快速入门与验证指南

1. 安装与环境配置

# Clone and navigate to repository
cd MCP_Autonomous_Agent

# Install dependencies
pip install -r requirements.txt

2. 生成种子数据仓库

使用确定性的合成金融数据（固定种子 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.

3. 运行完整测试套件

执行全部 66 项单元测试和集成测试，覆盖 AST 验证、EXPLAIN 分析、数据库引擎线程安全、MCP 工具以及 Agent 自愈循环：

python -m unittest discover -s tests -v

4. 运行自主 Agent 演示

针对金融数据仓库执行一次多轮分析型查询会话：

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}")

5. 在 Stdio 上启动 MCP 服务器

连接 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 documentation

许可证

MIT 许可证。专为企业金融数据分析和 AI Agent 作品集演示而创建。

This server cannot be installed

license - not tested

Not graded

quality - not tested

maintenance

How are these scores calculated?

Maintenance

–Maintainers

–Response time

–Release cycle

–Releases (12mo)

Commit activity

Resources

GitHub Repository

Need Help?

Related Servers

Unclaimed servers have limited discoverability.

Looking for Admin?

If you are the server author, to access and configure the admin panel.

Related MCP Connectors

PlainQuery | MCP DB Agent
Query PostgreSQL databases in plain English — LLM-generated, safety-validated SQL.
Prism — Crossdeck
Ask your app anything — revenue, errors, read-cost, growth — and get rendered charts back.
BankBridge
Read-only bank access for your AI agent. Connects Claude, ChatGPT, Cursor, Gemini, Codex.

View all MCP Connectors

Latest Blog Posts

Who's Calling? MCP Hosts Are an Identity Blind Spot (And the Spec Knows It)
By Om-Shree-0709 on July 25, 2026.
mcp
Agent Identity
OAuth 2.1
Your AI Chatbot Just Exposed Your CEO's Salary to an Intern
By Om-Shree-0709 on July 2, 2026.
Agent Identity
MCP Security
OAuth Delegation
Why MCP Servers Need Execution Sandboxing (And Why Your Current Stack Isn't Enough)
By Om-Shree-0709 on June 30, 2026.
Agentic Ai
Prompt Injection
WebAssembly

MCP directory API

We provide all the information about MCP servers via our MCP API.

curl -X GET 'https://glama.ai/api/mcp/v1/servers/fortuneMog/MCP_Autonomous_Agent'

If you have feedback or need assistance with the MCP directory API, please join our Discord server