Ein produktionsreifes, Enterprise-Finanzanalysesystem, das das Anthropic Model Context Protocol (MCP) mit einem autonomen Reasoning-Agenten integriert. Das System stellt Large Language Models (LLMs) über einen standardkonformen JSON-RPC-2.0-Transport mit stdio ein sicheres, relationales Finanz-Data-Warehouse mit mehreren Tabellen bereit.
Es verfügt über eine intelligente 5-Ebenen-Defense-in-Depth-Architektur, einen SQL-AST-Lexer und Recursive-Descent-Parser in reinem Python, einen EXPLAIN-Plan-Performance-Analysator, einen threadsicheren Verbindungspool mit Opcode-Ausführungs-Timeouts und eine Autonomous-Agent-Self-Healing-Schleife, die sich automatisch von SQL-Syntaxfehlern, AST-Sicherheitsverletzungen und Warnungen zu kartesischen Joins erholen kann.
┌─────────────────────────────────────────────────────────────────────────────┐
│ 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
Das System erzwingt über 5 unabhängige Ebenen hinweg strikte Sicherheitsgrenzen zwischen dem LLM und der Datenbank-Engine:
Ebene | Komponente | Sicherheitsmechanismus | AbgedecktBedrohungsvektor |
Ebene 1: AST-Gate vor der Ausführung |
| Reiner Python-Lexer und Recursive-Descendant Parser verifizieren DQL-Einzelanweisungen ( | Gestapelte Query-Injection ( |
Ebene 2: Kosten-Gate vor der Ausführung |
| Wertet die SQLite | Kartesische Produkte ($O(N \times M)$-Joins), unbegrenzte Scans, Speichererschöpfung durch temporäre B-Bäume. |
Ebene 3: Schreibgeschützter OS- & Engine-Modus |
| SQLite-Verbindung wird über die URI | Nicht autorisierte Schreibversuche auf dem Datenträger, Schema-Manipulation. |
Ebene 4: Autorisierungs-Callback zur Laufzeit |
|
| Umgehungsversuche über |
Ebene 5: Ressourcen- und Speicher-Guardrails |
| Opcode-Fortschrittshandler ( | Runaway-rekursive CTEs, CPU-Denial-of-Service, Out-of-Memory-Abstürze durch unbegrenzte Ergebnismengen. |
agent/ast_validator.py)Das AST-Sicherheitsgate implementiert eine Engine mit zwei Modi:
Lexer- und Recursive-Descendant-Parser in reinem Python ohne Abhängigkeiten: ImplementierenSie mit Python-Standardbibliotheken und vollständigem Ursprung-Tracking (Zeile/Spalte).
Optionale sqlglot-Engine: Dialektbewusster Parser, der automatisch aktiviert wird, wenn sqlglot installiert ist.
Ein-Anweisung DQL: SELECT und WITH [RECURSIVE] ... SELECT.
Common Table Expressions (CTEs): einzelne und mehrfach verkettete CTEs. Der Parser traversiert CTE-Definitionen rekursiv, um sicherzustellen, dass kein DML eingebettet ist.
Fensterfunktionen: OVER (PARTITION BY ... ORDER BY ... [ROWS/RANGE ...]), ROW_NUMBER(), RANK(), SUM() OVER ().
Mehr-Tabellen-Joins: INNER JOIN, LEFT OUTER JOIN, CROSS JOIN, NATURAL JOIN mit ON und USING (...).
Subquerys: Subquerys in FROM-Klauseln, skalare Subquerys in SELECT, IN (SELECT ...), EXISTS (SELECT ...).
Zusammengesetzte Mengenoperationen: UNION [ALL], INTERSECT, EXCEPT.
Skalare Ausdrücke: CASE WHEN ... THEN ... ELSE ... END, CAST(... AS ...), String-Konkatenierung (||), Arithmetik.
DDL-Anweisungen: DROP, CREATE, ALTER, TRUNCATE.
DML-Anweisungen: INSERT, UPDATE, DELETE, REPLACE, UPSERT, MERGE.
Administrative Befehle: PRAGMA, ATTACH, DETACH, VACUUM, REINDEX, ANALYZE, BEGIN, COMMIT.
Gefährliche Funktionen: load_extension, readfile, writefile, edit, fts3_tokenizer, eval, randomblob.
Systemtabellen: sqlite_master, sqlite_schema, sqlite_temp_master, sqlite_temp_schema, sqlite_sequence, sqlite_stat*.
Injektionsvektoren: Multi-Statement-Semikolons (;), nicht abgeschlossene Blockkommentare (/* ...), nicht abgeschlossene Strings.
agent/explain_analyzer.py)Parsert den EXPLAIN QUERY PLAN-Baum von SQLite, sowohl im SQLite-3.24+-4-Spaltenformat (id, parent, notused, detail) als auch in Legacy-Formaten.
$$\text{CostScore} = \min\left(100, \sum \text{Penalties}\right)$$
Operations-Detail | Klassifizierung | Schweregrad | Strafe |
| Nicht indizierter Voll-Scan | Hoch | +25.0 je |
| Aufbau einem ephemeren Index | Hoch | +20.0 |
| Nicht indizierte Sortierung | Mittel | +15.0 |
| Temporäre Aggregations-Bäume | Mittel | +10.0 |
| Materialisierte Unterabfrage | Mittel | +10.0 je |
| Kartesisches Produkt-Join | Kritisch | +30.0 |
$0.0-25.0$ (OPTIMAL): Vollständig über alle Indizes gelöst, sofortige Ausführung.
$26.0-50.0$ (AKZEPTABEL): Geringere temporäre Sortierung oder ein Zeilen-Scan auf kleiner Tabelle.
$51.0-74.0$ (WARNUNG): Suboptimale Nach-/Abfrage; mehrere Tabellen-Scans.
$75.0-100.0$ (KRITISCH): Kartesisches Produkt oder schwerer unindexierter Join. Vom MCP-Ausführungsgate blockiert.
data/schema.sql)Das Warehouse modelliert eine Domäne für Fahrzeugfinanzierung mit 6 relationalen Tabellen:
┌──────────────┐ 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 regionaler Filialen und Retail-Filialen mit rekursiver Eltern-Kind-Hierarchie (parent_branch_id).
customers: 300 Kundenprofile mit lognormaler Einkommensverteilung, Verhältnis Schuld-zu-Einkommen und SHA-256-PII-Hashes.
credit_ratings: Über 600 longitudinale Bureau-Score-Snapshots in 5 Risikostufen (PRIME_PLUS bis DEEP_SUBPRIME).
loans: 500 Fahrzeugfinanzierungs- und KMU-Verträge mit Risikogew-Zinsen und monatlichen Tilgungsraten.
repayments: Über 17.000 Transaktionsbuchhalter-Einträge mit Aufschlüsselung nach Prinzipal/Zins/Gebühren und Zahlung-VZTracking.
audit_log: Unveränderliche Prüfprotokolle zur Verfolgung der Loan-Zustandsübergänge bis zu DELINQUENT_90, DEFAULTED und WriteOff.
Der Server (agent/server.py) bietet 4 Kern-Tools an:
query_database (Alias: query_financial_lakehouse)Führt sichere, read-only-SQL-Abfragen aus – mit automatischer AST-Validierung vor der Ausführung, Opcode-Timeout und Zeilendeckel.
Eingabe: query (str, erforderlich), max_rows (int, Standard: 100), timeout_seconds (float, Standard: 5.0).
Ausgabe: JSON-Objekt mit columns, rows, row_count, is_truncated, execution_time_ms.
explain_queryInspiziert die Ausführungsplan-Node, berechnet einen Cost Score ($0–100$), erkennt Scans und liefert Index-Empfehlungen, ohne Mutationen auszuführen.
Eingabe: query (str, erforderlich).
Ausgabe: cost_score, complexity_rating, scanned_tables, indexed_tables, warnings, recommendations.
get_database_schemaReflectiert Katalog-Metadaten, Spaltentypen, Primär- und Fremdschlüssel sowie Indizes.
Eingabe: table_name (str, optional).
Ausgabe: Vollständige oder gefilterte Tabellenschema-Definitionen.
validate_sql_safetyFührt statische AST-Sicherheitsanalyse ohne Datenbankzugriff aus.
Eingabe: query (str, erforderlich).
Ausgabe: is_safe (bool), statement_type, referenced_tables, detected_risks.
agent/client_runner.py)Der AutonomousDataAgent implementiert eine iterative Tool-Kommandoschleife mit geschlossenem Regelkreis zur Fehlerbehebung:
┌─────────────────────────────────────┐
│ 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-Syntax-/Schema-Fehler (z. B. falscher Spaltenname): Setzt SYNTAX_ERROR_TEMPLATE mit Katalogtabelle ein; der Agent repariert die Spaltennamen.
AST-Sicherheitsablehnung (z. B. nicht-DQL-Abfrage): Setzt AST_VIOLATION_TEMPLATE ein; der Agent formuliert eine konforme einzelne SELECT-Anweisung.
Hoher Kostenpunkt / kartesischer Join: Setzt PLAN_WARNING_TEMPLATE; der Agent fügt indizierte Join-Prädikate hinzu.
Deterministischer Offline-Betrieb: MockClaudeClient ermöglicht zu 100 % Testausführung ohne Anthropic-API-Key.
# Clone and navigate to repository
cd MCP_Autonomous_Agent
# Install dependencies
pip install -r requirements.txtBefülle data/warehouse.db mit deterministischen, synthetischen Finanzdaten (Seed 42):
python data/seed_warehouse.pyAusgabe:
[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.Alle 66 Unit- und Integrationstests über AST-Validierung, EXPLAIN-Analysen, DB-Engine-Thread-Schere, MCP-Tools und Agent-Self-Healing-Loops:
python -m unittest discover -s tests -vFührt eine mehrstufige, analytische Query-Sitzung gegenüber dem Daten-Warehouse aus:
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}")Für die Verbindung mit der Anthropic Claude Desktop App oder dem MCP Inspector:
python agent/server.pyKonfiguriere in 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 documentationThis 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.
