MCP 서버 구축하기

LLM의 요청에 따라 주소록에서 특정 사람 혹은 특정 그룹에 속한 사람들의 이메일 주소를 제공하는 MCP 서버를 만들어보겠습니다.

라이브러리 설치

1
2
3
4
5
6
7
8
# uv
uv add mcp pandas

# pip
pip install mcp pandas

# conda
conda install mcp pandas

코드

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
# main.py

from mcp.server.fastmcp import FastMCP
import pandas as pd

# 주소록 파일 읽기
file_path = "data/address_book.csv"

# MCP 서버 정의
mcp = FastMCP("AddressBook",
              host="127.0.0.1",  # 호스트 주소
              # host="0.0.0.0",  # 호스트 주소 (전역)
              port=8000,  # 포트 번호
              )

# 파일 읽어 df로 반환
def file_to_df(file_path):
    return pd.read_csv(file_path, encoding="utf8")

# 공백 제거 유틸
def remove_whitespace(text:str) -> str:
    text = text.replace(' ', '')
    return text

# 이름으로 이메일 찾기
@mcp.tool()
def search_email_by_name(name: str) -> str:
    df = file_to_df(file_path)
    result = df.loc[df["name"] == name, "email"]
    if result.empty:
        result = df.loc[df["name"].str.replace(' ', '') == remove_whitespace(name), "email"]
    emails = result.tolist()
    return f"조회 결과: {emails[0]}" if emails else f"name {name} 데이터 없음"

# 그룹명으로 이메일 리스트 찾기
@mcp.tool()
def search_email_by_group(group_name: str) -> str:
    df = file_to_df(file_path)
    result = df.loc[df["group"] == group_name, "email"]
    if result.empty:
        result = df.loc[df["group"].str.replace(' ', '') == remove_whitespace(group_name), "email"]
    emails = result.tolist()
    return f"조회 결과: {emails}" if emails else f"group {group_name} 데이터 없음"

def main():
    # Server Sent Events 형식으로 데이터 전송하는 MCP 서버 구동
    mcp.run(transport="sse")

if __name__ == "__main__":
    main()

주소록 파일

1
2
3
4
5
name,email,group
홍길동,hong@hong.hog,korea
세종대왕,king_sejong@josun.gov,korea
김철수,cs@example.com,dev
이영희,yh@example.com,design

위 파일의 이름을 address_book.csv 로 명명하고, main.py 가 있는 디렉터리 내 data 폴더에 넣습니다.

1
2
3
4
5
# 디렉터리 구조
.
├─ main.py
└─ data/
   └─ address_book.csv

실행

1
2
3
4
5
6
7
8
# uv
uv run python main.py

# python (pip)
python main.py

# conda
python main.py

테스트

1. 세션 ID 값 받기

HTTP 로 테스트할 수 있으며, 이를 위해선 먼저 세션ID 값을 받아야 합니다. 인터넷 브라우저에서 주소창에 <MCP가 실행되는 IP>:<MCP Port>/sse 를 입력하면 세션 ID를 확인할 수 있습니다.

2. 연결 초기화

이제 Postman 과 같은 HTTP 테스트 도구 등에서 붙여넣어 테스트를 합니다. 가장 먼저 연결을 초기화(initialize)해야 합니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# Metod : POST
http://<MCP가 실행되는 IP>:<MCP Port>/messages?session_id=<세션ID>

# BODY
{
  "jsonrpc": "2.0",
  "id": 0,
  "method": "initialize",
  "params": {
    "protocolVersion": "2024-11-05",
    "capabilities": {},
    "clientInfo": {
      "name": "test-client",
      "version": "1.0.0"
    }
  }
}

정상적이라면 위와 같이 ACCEPT 문구를 받아볼 수 있습니다.

3. Tool 호출

이제 실제 Tool 호출을 해봅니다.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# Metod : POST
http://<MCP가 실행되는 IP>:<MCP Port>/messages?session_id=<세션ID>

# BODY
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "search_email_by_name",
    "arguments": {
      "name": "홍길동"
    }
  }
}

다시 아까 웹브라우저로 가서 응답을 보면, 정상적으로 조회가 된 것을 볼 수 있습니다.

Reference

[MCP Study] 2. 다수의 MCP 서버와 클라이언트 통신하기

Comments