MCP Autonomous Data Agent
by fortuneMog
README.md
# Anthropic Claude API & MCP Autonomous Data Agent
[](https://www.python.org/)
[](https://modelcontextprotocol.io/)
[]()
[]()
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:
1. **Zero-Dependency Pure Python Lexer & Recursive Descent Parser**: Built using Python standard libraries with full coordinate tracking (line/column).
2. **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)
└──────────────┘
```
1. **`branches`**: 12 regional hubs and retail branches with recursive parent-child hierarchy (`parent_branch_id`).
2. **`customers`**: 300 borrower profiles with lognormal income distributions, debt-to-income ratios, and SHA-256 PII hashes.
3. **`credit_ratings`**: 600+ longitudinal bureau score snapshots across 5 risk tiers (`PRIME_PLUS` to `DEEP_SUBPRIME`).
4. **`loans`**: 500 vehicle finance and SME contracts with risk-adjusted interest rates and monthly amortization installments.
5. **`repayments`**: 17,000+ transaction ledger entries with principal/interest/fee breakdowns and delinquency tracking.
6. **`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
```bash
# 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`):
```bash
python data/seed_warehouse.py
```
Output:
```text
[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:
```bash
python -m unittest discover -s tests -v
```
### 4. Run the Autonomous Agent Demo
Run a multi-turn analytical query session against the financial warehouse:
```python
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:
```bash
python agent/server.py
```
Configure in Claude Desktop `claude_desktop_config.json`:
```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.
This server cannot be deployed
Maintenance
ActivityMaintained
ResponsivenessNo issues