pulseMCP, smithery 등에 업데이트되는 정보들이 빠르게 증가하는 것을 보니, MCP(Model Context Protocol)을 이용한 서버와 클라이언트 개발이 매우 활발하게 이루어지는 것 같습니다. . 관련해서 이번 포스트에서는 Python의 mcp 라이브러리를 사용하여 간단한 MCP 서버를 구축하는 방법을 stdio와 SSE 두 가지 방식으로 나누어 설명하겠습니다.
SSE와 stdio에 대한 설명
SSE (Server-Sent Events)
SSE는 서버-센트 이벤트(Server-Sent Events)의 약자로, 웹 브라우저(클라이언트)가 서버로부터 자동으로 업데이트를 푸시(push)받을 수 있도록 하는 웹 기술입니다.
서버에서 클라이언트로의 단방향 통신을 위해 설계되었으며, HTTP 프로토콜 위에서 동작합니다. 클라이언트가 한번 연결을 요청하면, 서버는 그 연결을 계속 열어두고 있다가 새로운 데이터나 이벤트가 발생할 때마다 클라이언트로 데이터를 보낼 수 있습니다.
- 주요 특징:
- 단방향 통신 (서버 → 클라이언트): 서버에서 클라이언트로만 데이터를 전송합니다. 클라이언트가 서버에 데이터를 보낼 필요 없이, 업데이트만 받으면 되는 경우에 이상적입니다.
- 자동 재연결: 네트워크 문제 등으로 연결이 끊어지면 브라우저가 자동으로 재연결을 시도하는 표준 기능이 내장되어 있습니다.
- 간단한 구현: WebSocket에 비해 구현이 훨씬 간단합니다. 클라이언트 측에서는 자바스크립트의
EventSourceAPI를 사용해 쉽게 구현할 수 있습니다. - 용도: 실시간 뉴스 피드, 스포츠 경기 스코어 업데이트, 주식 시세 알림, 채팅 알림 등 서버로부터 지속적인 업데이트가 필요한 기능에 주로 사용됩니다.
stdio (Standard Input/Output)
stdio는 표준 입출력(Standard Input/Output)의 약자로, C 언어와 같은 시스템 프로그래밍 언어에서 사용되는 핵심 라이브러리입니다. 프로그램이 실행되는 컴퓨터 환경 내에서 기본적인 데이터 입출력을 처리하는 함수들의 모음입니다.
- 주요 특징:
- 로컬 입출력: 네트워크 통신이 아닌, 프로그램이 실행되는 로컬 시스템의 자원(키보드, 화면, 파일 등)과의 데이터 교환을 다룹니다.
- 스트림(Stream): 데이터의 흐름을 ‘스트림’이라는 개념으로 추상화하여 다룹니다. 대표적인 표준 스트림으로는 다음 세 가지가 있습니다.
stdin(Standard Input): 표준 입력 (기본값: 키보드)stdout(Standard Output): 표준 출력 (기본값: 화면)stderr(Standard Error): 표준 오류 출력 (기본값: 화면)
- 파일 처리:
fopen,fread,fwrite,fclose등의 함수를 통해 파일 시스템의 파일을 읽고 쓰는 기능을 제공합니다. - 용도: 명령줄 인터페이스(CLI) 애플리케이션, 시스템 유틸리티, 파일 처리 프로그램 등 거의 모든 종류의 로컬 애플리케이션 개발에 기본적으로 사용됩니다.
비교 요약
| 구분 | SSE (Server-Sent Events) | stdio (Standard Input/Output) |
| 분야 | 웹 기술 (Web Technology) | 프로그래밍 언어 라이브러리 (C Library) |
| 목적 | 서버 → 클라이언트 간의 실시간 데이터 푸시 | 프로그램과 로컬 자원(키보드, 화면, 파일) 간의 데이터 입출력 |
| 통신 방식 | 네트워크 통신 (HTTP 기반, 단방향) | 로컬 I/O (시스템 내부 처리) |
| 사용 환경 | 웹 브라우저와 웹 서버 간 | 운영체제 내에서 실행되는 응용 프로그램 |
| 주요 구성 요소 | EventSource API, HTTP 프로토콜 | stdin, stdout, stderr, FILE 포인터 |
| 대표적 사용 사례 | 실시간 알림, 뉴스 피드, 라이브 스코어 | 명령줄 도구, 텍스트 편집기, 파일 변환기 |
| 장점 | 웹 기반 환경에서 사용 가능 네트워크를 통한 원격 접근 가능 실시간 스트리밍 지원 | 구현이 간단함 로컬 환경에서 빠른 통신 설정이 최소화됨 |
| 단점 | 구현이 복잡함 HTTP 서버 설정 필요 네트워크 오버헤드 존재 | 네트워크 통신 불가 웹 환경에서 사용 제한적 |
MCP Server 만들기
MCP 서버는 AI 모델이 외부 리소스나 도구에 접근할 수 있도록 해주는 서버입니다. 파일 시스템 접근, 데이터베이스 연결, API 호출 등 다양한 기능을 AI 모델에게 제공할 수 있습니다.
사전 준비
먼저 필요한 패키지를 설치합니다:
pip install mcp
방식 1: stdio 방식 MCP Server
stdio 방식은 표준 입출력(Standard I/O)을 통해 통신하는 가장 간단한 방식입니다. 주로 Local 시스템 내에 연동에 사용됩니다.
MCP Client가 설치된 Local System 내에 MCP Server가 존재할 때, 활용할 수 있습니다.
기본 구조
# stdio_server.py
import asyncio
import sys
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp import types
# 서버 인스턴스 생성
server = Server("my-mcp-server")
@server.list_tools()
async def list_tools() -> list[types.Tool]:
"""사용 가능한 도구 목록을 반환합니다."""
return [
types.Tool(
name="calculator",
description="간단한 수학 계산을 수행합니다",
inputSchema={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "계산할 수학 표현식 (예: 2+2)"
}
},
"required": ["expression"]
}
),
types.Tool(
name="echo",
description="입력된 메시지를 그대로 반환합니다",
inputSchema={
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "반환할 메시지"
}
},
"required": ["message"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
"""도구 호출을 처리합니다."""
if name == "calculator":
try:
expression = arguments["expression"]
# 보안을 위해 간단한 검증
if any(char in expression for char in ['import', 'exec', 'eval', '__']):
raise ValueError("허용되지 않는 표현식입니다")
result = eval(expression)
return [types.TextContent(
type="text",
text=f"계산 결과: {expression} = {result}"
)]
except Exception as e:
return [types.TextContent(
type="text",
text=f"계산 오류: {str(e)}"
)]
elif name == "echo":
message = arguments["message"]
return [types.TextContent(
type="text",
text=f"Echo: {message}"
)]
else:
raise ValueError(f"알 수 없는 도구: {name}")
async def main():
"""메인 실행 함수"""
async with stdio_server() as (read_stream, write_stream):
await server.run(
read_stream,
write_stream,
server.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
stdio 서버 실행
python stdio_server.py
방식 2: SSE(Server-Sent Events) 방식 MCP Server
SSE 방식은 HTTP를 통해 실시간 스트리밍 통신을 하는 방식입니다. 웹 기반 환경에서 유용합니다.
SSE 서버 구현
# sse_server.py
import asyncio
import json
from mcp.server import Server
from mcp.server.sse import SseServerTransport
from mcp import types
from starlette.applications import Starlette
from starlette.responses import Response
from starlette.routing import Route
import uvicorn
# 서버 인스턴스 생성
server = Server("my-sse-mcp-server")
@server.list_tools()
async def list_tools() -> list[types.Tool]:
"""사용 가능한 도구 목록을 반환합니다."""
return [
types.Tool(
name="weather",
description="날씨 정보를 가져옵니다 (모의 데이터)",
inputSchema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "도시 이름"
}
},
"required": ["city"]
}
),
types.Tool(
name="timestamp",
description="현재 시간을 반환합니다",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
"""도구 호출을 처리합니다."""
if name == "weather":
city = arguments["city"]
# 실제로는 날씨 API를 호출하겠지만, 여기서는 모의 데이터 사용
weather_data = {
"서울": "맑음, 22°C",
"부산": "흐림, 25°C",
"대구": "비, 18°C"
}
weather = weather_data.get(city, f"{city}의 날씨 정보를 찾을 수 없습니다")
return [types.TextContent(
type="text",
text=f"{city} 날씨: {weather}"
)]
elif name == "timestamp":
from datetime import datetime
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
return [types.TextContent(
type="text",
text=f"현재 시간: {now}"
)]
else:
raise ValueError(f"알 수 없는 도구: {name}")
# SSE 엔드포인트 설정
async def handle_sse(request):
"""SSE 연결을 처리합니다."""
transport = SseServerTransport("/message")
async def message_handler():
async for message in request.stream():
data = await message
# 메시지 처리 로직
yield f"data: {json.dumps({'response': 'processed'})}\n\n"
return Response(
message_handler(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"Access-Control-Allow-Origin": "*",
}
)
# Starlette 앱 설정
app = Starlette(routes=[
Route("/sse", handle_sse, methods=["POST"]),
])
# 서버 실행을 위한 메인 함수
async def run_sse_server():
"""SSE 서버를 실행합니다."""
config = uvicorn.Config(
app,
host="0.0.0.0",
port=8000,
log_level="info"
)
server_instance = uvicorn.Server(config)
await server_instance.serve()
if __name__ == "__main__":
print("SSE MCP 서버를 시작합니다...")
print("서버 주소: http://localhost:8000")
asyncio.run(run_sse_server())
SSE 방식을 위한 추가 의존성 설치:
pip install starlette uvicorn
SSE 서버 실행
python sse_server.py
고급 기능 추가
리소스 제공 기능
@server.list_resources()
async def list_resources() -> list[types.Resource]:
"""사용 가능한 리소스 목록을 반환합니다."""
return [
types.Resource(
uri="file://config.json",
name="설정 파일",
description="애플리케이션 설정 파일",
mimeType="application/json"
)
]
@server.read_resource()
async def read_resource(uri: str) -> str:
"""리소스를 읽어 반환합니다."""
if uri == "file://config.json":
return json.dumps({
"version": "1.0",
"debug": True,
"max_connections": 100
}, indent=2)
else:
raise ValueError(f"알 수 없는 리소스: {uri}")
프롬프트 기능
@server.list_prompts()
async def list_prompts() -> list[types.Prompt]:
"""사용 가능한 프롬프트 목록을 반환합니다."""
return [
types.Prompt(
name="code_review",
description="코드 리뷰를 위한 프롬프트",
arguments=[
types.PromptArgument(
name="code",
description="리뷰할 코드",
required=True
)
]
)
]
@server.get_prompt()
async def get_prompt(name: str, arguments: dict) -> types.GetPromptResult:
"""프롬프트를 반환합니다."""
if name == "code_review":
code = arguments["code"]
return types.GetPromptResult(
description="코드 리뷰 프롬프트",
messages=[
types.PromptMessage(
role="user",
content=types.TextContent(
type="text",
text=f"다음 코드를 리뷰해주세요:\n\n```\n{code}\n```"
)
)
]
)
else:
raise ValueError(f"알 수 없는 프롬프트: {name}")
배포 및 운영
Docker를 이용한 배포
# Dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . EXPOSE 8000 CMD ["python", "sse_server.py"]
요구사항 파일
# requirements.txt
mcp>=1.0.0
starlette>=0.27.0
uvicorn>=0.22.0
마무리
MCP 서버를 구축하는 두 가지 방식을 살펴보았습니다. stdio 방식은 간단한 로컬 도구에 적합하고, SSE 방식은 웹 기반 서비스나 원격 접근이 필요한 경우에 유용합니다. 프로젝트의 요구사항에 맞는 방식을 선택하여 AI 모델과 외부 도구 간의 효율적인 통신을 구현하기 위해 MCP Client 구현이 필요합니다. 다음 포스트에서는 Client 구현 케이스를 준비하도록 하겠습니다.
각 방식의 코드는 개념이해를 돕기 위한 기본적인 구조만 제공하였으며, 실제 사용 시에는 보안, 에러 처리, 로깅 등을 추가로 고려해야 합니다.
같이 보면 좋을 컨텐츠
stdio 타입만 사용할수 있는 MCP Client를 위해 stdio 와 sse를 연결해주는 Proxy서버 :
MCP Proxy: 서버 트랜스포트 간 전환을 위한 필수 도구
답글 남기기