MCP Server: stdio와 SSE 방식 차이점

https://blog.choonzang.com/it/python/3341/

SSE (Server-Sent Events)

• SSE는 서버-센트 이벤트(Server-Sent Events)의 약자로, 웹 브라우저(클라이언트)가 서버로부터 자동으로 업데이트를 푸시(push)받을 수 있도록 하는 웹 기술입니다.

• 서버에서 클라이언트로의 단방향 통신을 위해 설계되었으며, HTTP 프로토콜 위에서 동작합니다. 클라이언트가 한번 연결을 요청하면, 서버는 그 연결을 계속 열어두고 있다가 새로운 데이터나 이벤트가 발생할 때마다 클라이언트로 데이터를 보낼 수 있습니다.

• 주요 특징

  • 단방향 통신 (서버 → 클라이언트): 서버에서 클라이언트로만 데이터를 전송합니다. 클라이언트가 서버에 데이터를 보낼 필요 없이, 업데이트만 받으면 되는 경우에 이상적입니다.

  • 자동 재연결: 네트워크 문제 등으로 연결이 끊어지면 브라우저가 자동으로 재연결을 시도하는 표준 기능이 내장되어 있습니다.

  • 간단한 구현: WebSocket에 비해 구현이 훨씬 간단합니다. 클라이언트 측에서는 자바스크립트의 EventSource API를 사용해 쉽게 구현할 수 있습니다.

  • 용도: 실시간 뉴스 피드, 스포츠 경기 스코어 업데이트, 주식 시세 알림, 채팅 알림 등 서버로부터 지속적인 업데이트가 필요한 기능에 주로 사용됩니다.

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 vs. stdio 비교 분석

현재 진행 중인 MCP(Model Context Protocol) 프로젝트 관점에서 이 두 방식을 선택하는 기준은 다음과 같습니다.

1. stdio 방식 선택 시

   • Claude Desktop과 같은 로컬 앱에 직접 MCP 서버를 연결할 때 사용합니다.

   • 가장 빠르고 설정이 간편하며, 보안상 외부 네트워크 노출이 필요 없을 때 최적입니다.

2. SSE 방식 선택 시

   • 멀리 떨어진 서버에 에이전트 도구를 배포하거나, 여러 클라이언트가 동시에 MCP 서버에 접속해야 할 때 사용합니다.

   • 웹 기반 UI(예: 현재 프로젝트의 React 프론트엔드)에서 실시간으로 도구 실행 로그를 받아올 때 유리합니다.

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())

• FastMCP

FastMCP의 장점

• 코드량 감소: list_tools와 call_tool 내부의 if-else 분기 로직이 사라졌습니다. 데코레이터가 이를 자동화합니다.
• 타입 힌트 활용: FastMCP는 Python의 타입 힌트(str, int 등)를 읽어서 자동으로 MCP용 inputSchema를 생성합니다. JSON Schema를 직접 작성할 필요가 없습니다.
• Docstring 활용: 함수 아래의 독스트링("""...""")이 자동으로 도구의 설명(Description)으로 변환되어 에이전트에게 전달됩니다.
• 일관된 인터페이스: 비즈니스 로직(함수)은 그대로 두고, 마지막 mcp.run()의 transport 값만 바꾸는 것으로 로컬 전용(stdio)과 웹 서비스용(SSE)을 자유롭게 오갈 수 있습니다.

# fastmcp_stdio.py
from mcp.server.fastmcp import FastMCP

# 서버 인스턴스 생성
mcp = FastMCP("My FastMCP Stdio Server")

@mcp.tool()
def calculator(expression: str) -> str:
    """간단한 수학 계산을 수행합니다. (예: 2+2)"""
    try:
        # 보안을 위한 기초 검증
        if any(char in expression for char in ['import', 'exec', 'eval', '__']):
            return "오류: 허용되지 않는 표현식입니다."
        
        result = eval(expression)
        return f"계산 결과: {expression} = {result}"
    except Exception as e:
        return f"계산 오류: {str(e)}"

@mcp.tool()
def echo(message: str) -> str:
    """입력된 메시지를 그대로 반환합니다."""
    return f"Echo: {message}"

if __name__ == "__main__":
    # transport를 stdio로 설정 (기본값)
    mcp.run(transport="stdio")

• stdio 서버 실행

python stdio_server.py

방식 2: SSE(Server-Sent Events) 방식 MCP Server

• SSE 방식은 HTTP를 통해 실시간 스트리밍 통신을 하는 방식입니다. 웹 기반 환경에서 유용합니다.

• SSE 방식을 위한 추가 의존성 설치

  pip install starlette uvicorn

# 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())

• FastMCP

# fastmcp_sse.py
from mcp.server.fastmcp import FastMCP
from datetime import datetime

# 서버 인스턴스 생성
mcp = FastMCP("My FastMCP SSE Server")

@mcp.tool()
def weather(city: str) -> str:
    """도시 이름을 입력하면 현재 날씨를 알려줍니다."""
    weather_data = {
        "서울": "맑음, 22°C",
        "부산": "흐림, 25°C",
        "대구": "비, 18°C"
    }
    result = weather_data.get(city, f"{city}의 날씨 정보를 찾을 수 없습니다.")
    return f"{city} 날씨: {result}"

@mcp.tool()
def timestamp() -> str:
    """현재 시간을 반환합니다."""
    return f"현재 시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

if __name__ == "__main__":
    # transport를 sse로 설정하고 호스트/포트 지정
    # 내부적으로 uvicorn을 사용하여 서버를 구동합니다.
    mcp.run(transport="sse", host="0.0.0.0", port=8000)

• 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_resources(uri: str) -> str:
	"""리소스를 읽어서 반환합니다."""
    if uri == "file://config.json":
    	return json.dumps({
        	"version": "1.0",
            "debug": True,
            "max_connections": 100
        }, indents=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="코드 리뷰 프롬프트"
            message=[
            	types.PromptMessage(               		
                  role="user"
                  content=types.TextContent(
                  	type="text",
                    text=f"다음 코드를 리뷰해주세요:\n\n```\n{code}\n```"
                  )
                )
            ]
        )
	else:
    	raise ValueError(f"알 수 없는 프롬프트: {name}")