MCP 服务器

MCP Autonomous Data Agent

Enables secure, read-only analytical querying of financial data through natural language, with built-in SQL injection defense and automatic query repair.

README

Anthropic Claude API & MCP Autonomous Data Agent

A production-grade, enterprise financial analytics system integrating the Anthropic Model Context Protocol (MCP) with an Autonomous Reasoning Agent. The system securely exposes a multi-table relational financial data warehouse to Large Language Models (LLMs) via standard JSON-RPC 2.0 stdio transport.

It features an intelligent 5-Layer Defense-in-Depth Architecture, a pure-Python SQL AST Lexer & Recursive Descent Parser, an EXPLAIN Plan Performance Analyzer, a thread-safe connection pool with opcode execution timeouts, and an Autonomous Agent Self-Healing Loop capable of auto-recovering from SQL syntax errors, AST security violations, and Cartesian join warnings.

Architecture Overview

┌─────────────────────────────────────────────────────────────────────────────┐
│                          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-Layer Defense-in-Depth Security Model

The system enforces strict security boundaries between the LLM and the database engine across 5 independent layers:

Layer	Component	Security Mechanism	Threat Vector Mitigated
Layer 1: Pre-Execution AST Gate	`agent/ast_validator.py`	Pure-Python Lexer & Recursive Descent Parser verifying single-statement DQL (`SELECT`, `WITH ... SELECT`).	Stacked query injection (`;`), DDL (`DROP`, `ALTER`, `CREATE`), DML (`INSERT`, `UPDATE`, `DELETE`), PRAGMA reconnaissance, comment exploits.
Layer 2: Pre-Execution Cost Gate	`agent/explain_analyzer.py`	Evaluates SQLite `EXPLAIN QUERY PLAN`, computing composite cost scores ($0-100$).	Cartesian products ($O(N \times M)$ joins), unbounded scans, memory exhaustion from temporary B-trees.
Layer 3: OS & Engine Read-Only Mode	`agent/db_engine.py`	SQLite connection established with URI `file:<path>?mode=ro`.	Unauthorized disk write attempts, schema tampering.
Layer 4: Runtime Authorizer Callback	`agent/db_engine.py`	`sqlite3.set_authorizer` restricting operations to `SQLITE_SELECT`, `SQLITE_READ`, `SQLITE_FUNCTION`, `SQLITE_RECURSIVE`, and safe schema PRAGMAs.	Bypasses attempting `ATTACH DATABASE`, `load_extension`, `PRAGMA writable_schema`, table mutation.
Layer 5: Resource & Memory Guardrails	`agent/db_engine.py`	Opcode progress handler (`conn.set_progress_handler`) monitoring query execution time + `fetchmany(max_rows + 1)` row truncation.	Runaway recursive CTEs, CPU denial-of-service, out-of-memory crashes from unbounded result sets.

AST SQL Security Validator (`agent/ast_validator.py`)

The AST Security Gate implements a dual-mode engine:

Zero-Dependency Pure Python Lexer & Recursive Descent Parser: Built using Python standard libraries with full coordinate tracking (line/column).
Optional sqlglot Engine: Dialect-aware parser activated automatically if sqlglot is installed.

Supported Analytical SQL Grammar

Single-Statement DQL: SELECT and WITH [RECURSIVE] ... SELECT.
Common Table Expressions (CTEs): Single and multiple chained CTEs. The parser recursively traverses CTE definitions ensuring no embedded DML.
Window Functions: OVER (PARTITION BY ... ORDER BY ... [ROWS/RANGE ...]), ROW_NUMBER(), RANK(), SUM() OVER ().
Multi-Table Joins: INNER JOIN, LEFT OUTER JOIN, CROSS JOIN, NATURAL JOIN with ON and USING (...).
Subqueries: Subqueries in FROM clauses, scalar subqueries in SELECT, IN (SELECT ...), EXISTS (SELECT ...).
Compound Set Operations: UNION [ALL], INTERSECT, EXCEPT.
Scalar Expressions: CASE WHEN ... THEN ... ELSE ... END, CAST(... AS ...), string concatenation (||), arithmetic.

Prohibited Patterns (100% Block Rate)

DDL: DROP, CREATE, ALTER, TRUNCATE.
DML: INSERT, UPDATE, DELETE, REPLACE, UPSERT, MERGE.
Administrative Commands: PRAGMA, ATTACH, DETACH, VACUUM, REINDEX, ANALYZE, BEGIN, COMMIT.
Dangerous Functions: load_extension, readfile, writefile, edit, fts3_tokenizer, eval, randomblob.
System Tables: sqlite_master, sqlite_schema, sqlite_temp_master, sqlite_temp_schema, sqlite_sequence, sqlite_stat*.
Injection Vectors: Multi-statement semicolons (;), unterminated block comments (/* ...), unterminated string literals.

EXPLAIN Query Plan Analyzer (`agent/explain_analyzer.py`)

Parses SQLite's EXPLAIN QUERY PLAN tree across SQLite 3.24+ 4-column format (id, parent, notused, detail) and legacy formats.

Scoring Formula & Penalties

$$\text{CostScore} = \min\left(100, ; \sum \text{Penalties}\right)$$

Operation Detail	Classification	Severity	Penalty
`SCAN TABLE <table>`	Unindexed Full Table Scan	High	+25.0 each
`SEARCH TABLE <table> USING AUTOMATIC INDEX`	Ephemeral Index Build	High	+20.0
`USE TEMP B-TREE FOR ORDER BY`	Unindexed Sort	Medium	+15.0
`USE TEMP B-TREE FOR GROUP BY/DISTINCT`	Temp Aggregation B-Tree	Medium	+10.0
`MATERIALIZE <id>`	Materialized Subquery	Medium	+10.0 each
Multi-Table Unindexed Scan	Cartesian Product Join	Critical	+30.0

Rating Categories

$0.0 - 25.0$ (OPTIMAL): Fully indexed point/range lookups. Instant execution.
$26.0 - 50.0$ (ACCEPTABLE): Minor temp sorting or single small table scan.
$51.0 - 74.0$ (WARNING): Sub-optimal plan; multiple scans.
$75.0 - 100.0$ (CRITICAL): Cartesian product or heavy unindexed join. Blocked by MCP execution gate.

Financial Data Warehouse Schema (`data/schema.sql`)

The warehouse models a vehicle asset financing domain with 6 relational tables:

┌──────────────┐       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 regional hubs and retail branches with recursive parent-child hierarchy (parent_branch_id).
customers: 300 borrower profiles with lognormal income distributions, debt-to-income ratios, and SHA-256 PII hashes.
credit_ratings: 600+ longitudinal bureau score snapshots across 5 risk tiers (PRIME_PLUS to DEEP_SUBPRIME).
loans: 500 vehicle finance and SME contracts with risk-adjusted interest rates and monthly amortization installments.
repayments: 17,000+ transaction ledger entries with principal/interest/fee breakdowns and delinquency tracking.
audit_log: Immutable audit records tracking loan state transitions to DELINQUENT_90, DEFAULTED, and WRITE_OFF.

MCP Tools & JSON-RPC 2.0 Protocol Interface

The server (agent/server.py) exposes 4 core tools:

1. `query_database` (Alias: `query_financial_lakehouse`)

Executes safe read-only SQL queries with automatic pre-execution AST validation, opcode progress timeouts, and row capping.

Input: query (str, required), max_rows (int, default: 100), timeout_seconds (float, default: 5.0).
Output: JSON payload with columns, rows, row_count, is_truncated, execution_time_ms.

2. `explain_query`

Inspects execution plan nodes, calculates cost score ($0-100$), detects scans, and provides indexing recommendations without executing mutations.

Input: query (str, required).
Output: cost_score, complexity_rating, scanned_tables, indexed_tables, warnings, recommendations.

3. `get_database_schema`

Reflects database catalog metadata, column types, primary keys, foreign keys, and indexes.

Input: table_name (str, optional).
Output: Full or filtered table schema definitions.

4. `validate_sql_safety`

Performs static AST security analysis without database access.

Input: query (str, required).
Output: is_safe (bool), statement_type, referenced_tables, detected_risks.

Autonomous Agent & Self-Healing Loop (`agent/client_runner.py`)

The AutonomousDataAgent implements an iterative tool-calling loop with closed-loop error remediation:

                  ┌─────────────────────────────────────┐
                  │ 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 │
                             └──────────────────┘   └──────────────────┘

Self-Correction Scenarios Handled

SQLite Syntax / Schema Error (e.g. misspelled column): Injects SYNTAX_ERROR_TEMPLATE with schema catalog; agent repairs column names.
AST Security Rejection (e.g. non-DQL query): Injects AST_VIOLATION_TEMPLATE; agent reformulates compliant single-statement SELECT.
High Query Cost / Cartesian Join: Injects PLAN_WARNING_TEMPLATE; agent adds indexed join predicates.
Deterministic Offline Execution: MockClaudeClient allows 100% offline testing without an Anthropic API key.

Quickstart & Verification Guide

1. Installation & Environment Setup

# Clone and navigate to repository
cd MCP_Autonomous_Agent

# Install dependencies
pip install -r requirements.txt

2. Generate Seed Data Warehouse

Populate data/warehouse.db with deterministic synthetic financial data (fixed seed 42):

python data/seed_warehouse.py

Output:

[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. Run the Comprehensive Test Suite

Execute all 66 unit and integration tests across AST validation, EXPLAIN analysis, DB engine thread safety, MCP tools, and agent self-healing loops:

python -m unittest discover -s tests -v

4. Run the Autonomous Agent Demo

Run a multi-turn analytical query session against the financial warehouse:

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. Launch MCP Server on Stdio

To connect with the Anthropic Claude Desktop app or MCP Inspector:

python agent/server.py

Configure in Claude Desktop claude_desktop_config.json:

{
  "mcpServers": {
    "financial-data-agent": {
      "command": "python",
      "args": ["-m", "agent.server"],
      "cwd": "/path/to/MCP_Autonomous_Agent"
    }
  }
}

Project Structure

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

License

MIT License. Created for enterprise financial data analytics and AI agent portfolio demonstration.