MCP로 AI 데이터 분석가 만들기
MCP를 이용해 AI 데이터 분석가를 만들어 보았습니다.
지난 번 Text-to-SQL 글에서는 사용자가 질문을 던지면, RAG로부터 생성한 프롬프트를 이용해 LLM이 적절한 SQL 쿼리를 생성하도록 하는 내용을 다루었다.
이번 글은 그 연장선으로, LLM이 해당 쿼리를 데이터베이스에서 실행하고, 그 결과를 분석해 사용자에게 응답까지 하는 Agent를 개발해보려고 한다. 이를 위해서는 우선 MCP에 대해 알아야 할 필요가 있다.
MCP란?
MCP(Model Context Protocol)는 LLM이 안전하고 표준화된 방식으로 외부 리소스와 상호작용할 수 있도록 돕는 오픈 소스 프로토콜이다. Anthropic(Claude)이 주도해서 개발하고 있으며, 현재 많은 개발자들이 생태계에 참여하여 일종의 마켓플레이스를 형성하고 있다.
이러한 프로토콜이 필요했던 이유는, 클라우드 환경에서 제공되는 LLM들은 AI라고 하더라도 사용자가 필요로하는 모든 정보를 갖고 있지는 않기 때문이다. 사용자의 데이터베이스에는 어떤 데이터가 있는지, 로컬 환경 또는 원격 환경에 어떤 파일이 있는지는 사용자가 정보를 넘겨주어야 LLM이 알 수 있다.
기존에도 Function Call과 같은 플러그인을 구현하여 LLM이 사용자가 운영하고 있는 시스템에 접근할 수 있었지만, 이제는 MCP라는 더 간단하고 통일된 수단을 이용할 수 있게 되었다.
구성 요소
MCP에는 대표적으로 세 가지 구성 요소가 있다.
MCP Server
LLM이 사용자가 운영하는 시스템에 접근할 수 있는 도구를 제공한다. 그리고 각 도구의 사용법, 파라메터, 응답 형식 등을 Manifest로 제공한다.
Manifest
Manifest는 MCP Server가 제공하는 도구들의 기능, 입력값, 출력값 등을 정의한 일종의 명세서이다. LLM은 이 문서를 참고해서 MCP Server에는 어떤 도구가 있는지, 언제 도구들을 사용해야 할지를 판단한다.
MCP Client
MCP Client는 도구를 실제로 사용하는 주체, 즉 AI Agent이다. Manifest를 Client에 등록해주면, MCP Server가 제공하는 도구 목록을 받아, 필요할 때 도구를 호출해서 작업을 수행한다.
Claude, Cursor 등의 Desktop Application과 연동해서 사용가능하며, 사용자가 직접 Client를 구현할 수도 있다.
AI Agent 만들기
이제 우리는 MCP를 활용하여 LLM이 데이터베이스에 쿼리를 실행시키도록 만들 수 있다. 이를 이용해 데이터 분석가 역할을 하는 AI Agent를 만들어보자.
이번 글에서는 Perfact팀(Workflow Orchestration의 그 Perfect가 맞다)의 FastMCP를 이용하여 MCP Server를, OpenAI가 제공하는 OpenAI-Agents를 이용하여 MCP Client를 구현해볼 예정이다.
DB 세팅
우선 분석 환경은 Trino로 가정하였다. Docker로 Trino 서버를 띄워주도록 한다.
1
2
3
4
5
6
7
8
9
services:
trino:
container_name: trino
hostname: trino
image: trinodb/trino:450
ports:
- "543:543"
volumes:
- ./docker/volume/trino:/var/lib/trino/data
자세한 설정은 이곳에 남겨두었다.
이제 Trino에 데이터를 업로드한 후, 테이블을 생성해보자. 사용한 데이터는 MovieLens ml-1m 데이터이며, 파일 포멧을 Parquet 형태로 변환하여 MinIO에 업로드해두었다.
그리고 Trino에서 해당 경로에 대한 스키마와 External Table을 생성하여, 쿼리를 통해 데이터에 접근 가능하도록 만들었다.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
CREATE SCHEMA movielens WITH ( LOCATION = 's3://hive/movielens' );
CREATE TABLE hive.movielens.users (
id INTEGER,
gender VARCHAR COMMENT 'The value should be either female "F" or male "M"',
age INTEGER,
occupation VARCHAR,
zip_code VARCHAR
)
WITH (
format = 'PARQUET',
external_location = 's3://hive/movielens/users/'
)
;
CREATE TABLE hive.movielens.ratings (
user_id INTEGER,
movie_id INTEGER,
rating INTEGER,
timestamp BIGINT
)
WITH (
format = 'PARQUET',
external_location = 's3://hive/movielens/ratings/'
)
;
CREATE TABLE hive.movielens.movies (
id INTEGER,
title VARCHAR,
genres VARCHAR
)
WITH (
format = 'PARQUET',
external_location = 's3://hive/movielens/movies/'
)
;
MCP Server 구현
다음은 MCP Server 어플리케이션을 구현해보자. 우선 로컬에 필요한 파이썬 디펜던시는 다음과 같다.
1
pip3 install 'mcp[cli]' sqlalchemy trino
이제 LLM이 사용할 도구를 구현해보자. FastMCP를 이용하면 데코레이터 패턴으로 Tool, Resource, Prompt 등을 선언할 수 있다. 이 데코레이터로 감싸진 함수들은 추후 MCP Client에 의해 호출되어 응답값을 넘겨줄 예정이다. (코드를 작성한 파일 이름에 유의하자.)
[mcp_example/query.py]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator
from dataclasses import dataclass
from sqlalchemy import create_engine, text, Engine
from mcp.server.fastmcp import Context, FastMCP
@dataclass
class AppContext:
engine: Engine
@asynccontextmanager
async def lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
engine = create_engine(
url="trino://admin@localhost:543/hive/movielens",
echo=True,
)
try:
yield AppContext(engine=engine)
finally:
pass
mcp = FastMCP(
name="execute-query",
lifespan=lifespan,
dependencies=["sqlalchemy", "trino"]
)
@mcp.tool()
def query_data(ctx: Context, sql: str) -> str:
engine: Engine = ctx.request_context.lifespan_context.engine
statement = text(sql)
with engine.connect() as conn:
rows = conn.execute(statement).fetchall()
return "\n".join(str(row) for row in rows)
if __name__ == "__main__":
print("Starting server...")
mcp.run(transport="stdio")
구현이 완료되면, 다음 스크립트를 실행하여 작성한 코드를 테스트해볼 수 있다. MCP는 자동으로 Web UI를 생성하여 테스트 환경을 제공한다.
1
mcp dev mcp_example/query.py
[http://localhost:5173]
MCP Client 구현
만약 Claude Desktop Application을 MCP Client로 사용하고 싶다면, 다음 스크립트를 실행하여 Config를 쉽게 생성할 수 있다.
1
mcp install mcp_example/query.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
{
"mcpServers": {
"execute-query": {
"command": "uv",
"args": [
"run",
"--with",
"mcp[cli]",
"--with",
"sqlalchemy",
"--with",
"trino",
"mcp",
"run",
"/Users/dglee/Side/mcp_example/query.py"
]
}
}
}
사실 이 Config는 Claude 뿐만 아니라, 모든 MCP Client에 적용 가능하다. 이를 이용해서 OpenAI ChatGPT에서 사용할 MCP Client를 구현해보자.
우선 필요한 파이썬 디펜던시는 다음과 같다.
1
pip3 install openai-agents
OpenAI API Key를 발급받아 환경 변수로 넘겨준 후, 다음과 같이 코드를 구현해준다. 이때 사용한 프롬프트들은 Text-to-SQL에서 사용했던 Contextual Prompting을 적당히 수정하여 재사용하였다.
[mcp_example/client.py]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
import os
import asyncio
import textwrap
from agents import Agent, Runner, trace
from agents.mcp import MCPServer, MCPServerStdio
os.environ["OPENAI_API_KEY"] = { Your OpenAI API Key }
async def run(mcp_server: MCPServer):
instructions = textwrap.dedent(
f"""
당신은 Trino SQL 전문가입니다. 유저의 질문에 대해 SQL 쿼리를 생성하고, 필요하다면 MCP 도구를 이용해 답변해주세요.
MCP 도구를 사용하기 전에, 제공되는 컨텍스트를 참고하여 SQL 쿼리를 생성해야 하며, 반드시 가이드라인을 준수해주세요.
==Given Context
CREATE TABLE movielens.users (
id INTEGER,
gender VARCHAR COMMENT 'The value should be either female "F" or male "M"',
age INTEGER,
occupation VARCHAR,
zip_code VARCHAR
)
CREATE TABLE movielens.ratings (
user_id INTEGER,
movie_id INTEGER,
rating INTEGER,
timestamp BIGINT
)
CREATE TABLE movielens.movies (
id INTEGER,
title VARCHAR,
genres VARCHAR
)
==Response Guidelines
1. 제공된 컨텍스트가 충분하다면 질문에 대한 설명 없이 정확한 SQL 쿼리를 생성해주세요.
2. 제공된 컨텍스트가 충분하지 않을 때에는 쿼리를 생성할 수 없는 이유를 설명해주세요.
3. 컨텍스트에서 테이블 리스트가 주어지면 그 중에서 가장 관련성 높은 테이블들을 사용해주세요.
4. JOIN 구문을 사용할 경우, 반드시 테이블 별칭(alias)을 명시해 주세요.
5. Subquery의 depth가 2보다 큰 경우, CTE를 사용해주세요.
6. 응답 전에 쿼리의 문법이 맞는지, 사용한 컬럼이 테이블에 존재하는지 한번 더 확인해주세요.
"""
)
agent = Agent(
name="Assistant",
instructions=instructions,
mcp_servers=[mcp_server],
)
question = "평점 4.0 이상을 한 번 이라도 준 남성 유저가 총 유저 중 몇 퍼센트인지 알려줘"
print("\n" + "-" * 40)
result = await Runner.run(starting_agent=agent, input=question)
print(result.final_output)
async def main():
async with MCPServerStdio(
cache_tools_list=True,
params={
"command": "uv",
"args": [
"run",
"--with",
"mcp[cli]",
"--with",
"sqlalchemy",
"--with",
"trino",
"mcp",
"run",
"/Users/dglee/Side/mcp_example/query.py"
]
},
) as server:
with trace(workflow_name="MCP AI Data Analyst"):
await run(server)
if __name__ == "__main__":
asyncio.run(main())
이제 작성한 파이썬 스크립트를 실행하면, LLM은 사용자의 질문에 맞게 알아서 SQL 쿼리를 작성하고 실행한다.
1
python3 mcp_example/client.py
[Question]
1
"평점 4.0 이상을 한 번 이라도 준 남성 유저가 총 유저 중 몇 퍼센트인지 알려줘"
SQLAlchemy에서 실행되는 쿼리를 출력하도록 echo=True로 설정했기 때문에, 우리가 넘겨준 DDL을 참고하여 LLM이 어떤 SQL을 작성했는지 확인할 수 있다.
그리고 MCP Client는 생성된 SQL을 데이터베이스에서 실행시키기 위해 MCP Server의 query_data 도구를 호출한다.
[Answer]
1
평점 4.0 이상을 한 번이라도 준 남성 유저는 전체 유저 중 약 71.67%입니다.
또한 Trino UI에 접속해서 실행 쿼리를 살펴보면, MCP Server가 쿼리를 정상적으로 처리했음을 볼 수 있다.
[http://localhost:543/ui]
