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

5계층 심층방어 보안 모델

이 시스템은 5개의 독립 계층에 걸쳐 LLM과 데이터베이스 엔진 간에 엄격한 보안 경계를 적용합니다.

계층	구성 요소	보안 메커니즘	완화되는 위협 벡터
계층 1: 실행 전 AST 게이트	`agent/ast_validator.py`	단일 문 DQL(`SELECT`, `WITH ... SELECT`)을 검증하는 순수 Python 렉서 및 재귀 하강 파서.	`;` 스택 쿼리 주입, DDL(`DROP`, `ALTER`, `CREATE`), DML(`INSERT`, `UPDATE`, `DELETE`), PRAGMA 정찰, 무한 주석 (또는 block comment) 공격을 방어합니다.
계층 2: 실행 전 비용 게이트	`agent/explain_analyzer.py`	SQLite의 `EXPLAIN QUERY PLAN`를 평가하여 복합 비용 점수($0-100$)를 계산합니다.	데카르트 곱($O(N \times M)$ 조인), 무제한 스캔, 임시 B-트리 생성으로 인한 메모리 고갈.
계층 3: OS 및 엔진 읽기 전용 모드	`agent/db_engine.py`	URI `file:<path>?mode=ro`로 SQLite 연결을 생성합니다.	허가되지 않은 디스크 쓰기 시도, 스키마 변조를 방지합니다.
계층 4: 런타임 권한자 콜백	`agent/db_engine.py`	`sqlite3.set_authorizer`를 이용해 `SQLITE_SELECT`, `SQLITE_READ`, `SQLITE_FUNCTION`, `SQLITE_RECURSIVE` 및 안전한 스키마 PRAGMA로 작업을 제한합니다.	`ATTACH DATABASE`, `load_extension`, `PRAGMA writable_schema`, 테이블 변경을 시도하는 우회 공격을 차단합니다.
계층 5: 리소스 및 메모리 보호 장치	`agent/db_engine.py`	Opcode 진행 상태 핸들러(`conn.set_progress_handler`)가 실행 시간을 모니터링하고, `fetchmany(max_rows + 1)`로 행 수를 잘라냅니다.	폭주하는 재귀 CTE, CPU 서비스 거부(DoS), 무제한 결과 집합으로 인한 메모리 부족 충돌을 방지합니다.

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 3.24+의 4열 형식 (id, parent, notused, detail) 및 이전 형식에서 SQLite의 EXPLAIN QUERY PLAN 트리를 파싱합니다.

점수 산정 공식 및 패널티

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

작업 상세	분류	심각도	패널티
`SCAN TABLE <table>`	인덱스 없는 전체 테이블 스캔	높음	+25.0 each
`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 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로 전환되는 것을 추적하는 불변 감사 기록.

MCP 도구 및 JSON-RPC 2.0 프로토콜 인터페이스

서버(agent/server.py)는 4개의 핵심 도구를 제공합니다.

1. `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 페이로드.

2. `explain_query`

데이터 수정을 수행하지 않고 테이블별 실행 계획 노드를 검사하고, 비용 점수($0-100$)를 계산하며, 스캔을 감지하고 인덱스 권고를 제공합니다.

입력: query(str, 필수).
출력: cost_score, complexity_rating, scanned_tables, indexed_tables, warnings, recommendations.

3. `get_database_schema`

데이터베이스 카탈로그 메타데이터, 컬럼 타입, 기본 키, 외래 키, 인덱스를 조사합니다.

입력: table_name(str, 선택).
출력: 테이블 스키마 정의 (전체 또는 필터링).

4. `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% 오프라인 테스트를 가능하게 합니다.

빠른 시작 및 확인 가이드

1. 설치 및 환경 설정

# Clone and navigate to repository
cd MCP_Autonomous_Agent

# Install dependencies
pip install -r requirements.txt

2. 시드 데이터 웨어하우스 생성

salt/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.

3. 포괄적 테스트 슈트 실행

AST 검증, EXPLAIN 분석, DB 엔진 스레드 안전성, MCP 도구, 에이전트 자가 복구 루프에 대한 66가지 단위 테스트 및 통합 테스트를 컴파일 실행합니다.

python -m unittest discover -s tests -v

4. 자율 에이전트 데모 실행

금융 웨어하우스에서 다수의 연속되는 분석 쿼리 세션을 실행합니다.

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 Desktop 앱 또는 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 License. 해당 프로젝트는 엔터프라이즈 금융 데이터 분석 및 AI 에이전트 포트폴리오 시연용으로 제작되었습니다.

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