프로덕션 등급의 엔터프라이즈 금융 분석 시스템으로, Anthropic Model Context Protocol(MCP) 을 자율 추론 에이전트와 통합합니다. 이 시스템은 표준 JSON-RPC 2.0 stdio 전송을 통해 다중 테이블 관계형 금융 데이터 웨어하우스를 대규모 언어 모델(LLM)에 안전하게 노출합니다.
이 시스템은 지능형 5계층 심층방어 아키텍처, 순수 Python 기반 SQL AST Lexer 및 재귀 파서, EXPLAIN 실행 계획 성능 분석기, 연산코드 실행 타임아웃이 있는 스레드 안전 커넥션 풀, 그리고 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
이 시스템은 5개의 독립 계층에 걸쳐 LLM과 데이터베이스 엔진 간에 엄격한 보안 경계를 적용합니다.
계층 | 구성 요소 | 보안 메커니즘 | 완화되는 위협 벡터 |
계층 1: 실행 전 AST 게이트 |
| 단일 문 DQL( |
|
계층 2: 실행 전 비용 게이트 |
| SQLite의 | 데카르트 곱($O(N \times M)$ 조인), 무제한 스캔, 임시 B-트리 생성으로 인한 메모리 고갈. |
계층 3: OS 및 엔진 읽기 전용 모드 |
| URI | 허가되지 않은 디스크 쓰기 시도, 스키마 변조를 방지합니다. |
계층 4: 런타임 권한자 콜백 |
|
|
|
계층 5: 리소스 및 메모리 보호 장치 |
| Opcode 진행 상태 핸들러( | 폭주하는 재귀 CTE, CPU 서비스 거부(DoS), 무제한 결과 집합으로 인한 메모리 부족 충돌을 방지합니다. |
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 3.24+의 4열 형식 (id, parent, notused, detail) 및 이전 형식에서 SQLite의 EXPLAIN QUERY PLAN 트리를 파싱합니다.
$$\text{CostScore} = \min\left(100, ; \sum \text{Penalties}\right)$$
작업 상세 | 분류 | 심각도 | 패널티 |
| 인덱스 없는 전체 테이블 스캔 | 높음 | +25.0 each |
| 일시적 인덱스 생성 | 높음 | +20.0 |
| 인덱스 없는 정렬 | 중간 | +15.0 |
| 임시 집계 B-트리 | 중간 | +10.0 |
| 즉시 본 (구용 서브쿼리) | 중간 | +10.0 each |
다중 테이블 비인덱스 스캔 | 데카르트 곱 조인 | 매우 치명 | +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: 로그정규소득 분포, DTI(부채/소득) 비율, SHA-256 개인정보(PII) 해시를 가진 300개 차입자 프로필.
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)실행 전 AST 검증, opcode 타임아웃, 행 제한을 적용하여 안전한 읽기 전용 SQL 쿼리를 실행합니다.
입력: query(str, 필수), max_rows(int, 기본: 100), timeout_seconds(float, 기본: 5.0).
출력: columns, rows, row_count, is_truncated, execution_time_ms를 포함하는 JSON 페이로드.
explain_query데이터 수정을 수행하지 않고 테이블별 실행 계획 노드를 검사하고, 비용 점수($0-100$)를 계산하며, 스캔을 감지하고 인덱스 권고를 제공합니다.
입력: query(str, 필수).
출력: cost_score, complexity_rating, scanned_tables, indexed_tables, warnings, recommendations.
get_database_schema데이터베이스 카탈로그 메타데이터, 컬럼 타입, 기본 키, 외래 키, 인덱스를 조사합니다.
입력: table_name(str, 선택).
출력: 테이블 스키마 정의 (전체 또는 필터링).
validate_sql_safety데이터베이스 접근 없이 정적 AST 보안 분석을 수행합니다.
입력: query(str, 필수).
출력: is_safe(bool), 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 문법 / 스키마 오류 (예: 잘못된 컬럼명): 스키마 카탈로그와 함께 SYNTAX_ERROR_TEMPLATE를 주입합니다. 에이전트는 컬럼명을 재조정합니다.
AST 보안 거부 (예: DQL이 아닌 쿼리): AST_VIOLATION_TEMPLATE를 주입합니다. 에이전트는 만족하는 단일 문 SELECT로 다시 수정합니다.
높은 쿼리 비용 / 데카르트 조인: PLAN_WARNING_TEMPLATE를 주입합니다. 에이전트는 참조 인덱스 조인 조건을 추가합니다.
결정적 오프라인 실행: MockClaudeClient는 Anthropic API 키 없이 100% 오프라인 테스트를 가능하게 합니다.
# Clone and navigate to repository
cd MCP_Autonomous_Agent
# Install dependencies
pip install -r requirements.txtsalt/salt: 고정 시드 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.AST 검증, EXPLAIN 분석, DB 엔진 스레드 안전성, MCP 도구, 에이전트 자가 복구 루프에 대한 66가지 단위 테스트 및 통합 테스트를 컴파일 실행합니다.
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 Desktop 앱 또는 MCP Inspector에 연결하려면:
python agent/server.pyClaude 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 License. 해당 프로젝트는 엔터프라이즈 금융 데이터 분석 및 AI 에이전트 포트폴리오 시연용으로 제작되었습니다.
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.
