SSE(Server‑Sent Events) 기반 MCP 서버 SSE 호출 가이드

이번 글에서는 SSE 전송 방식으로 MCP 툴 서버를 개설하고, 클라이언트가 실제 HTTP API로 호출하는 예제를 소개합니다.

---

1. 왜 SSE인가?

전송 방식	특징	장점
STDIO	파이프(동일 머신 내)	설정이 간단하지만 로컬 한정
HTTP	일반 REST	방화벽·LB 친화적이나 요청‑응답 1회성
SSE	단방향 스트리밍	서버가 즉시 Push → 실시간 로그·중간 결과 전송에 유용

실시간 문서 스트리밍, 진행률 표시 등 점진적 응답이 필요한 시나리오에 SSE가 적합합니다.

---

2. SSE 툴 서버 생성

아래 예시는 FastAPI를 사용해 /sse 엔드포인트를 노출합니다.

# weather_server.py
from typing import List
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """Get weather for location."""
    return "It's always sunny in New York"

if __name__ == "__main__":
    mcp.run(transport="sse")

• FastMCP 객체를 만들고 → @mcp.tool()로 함수를 노출 → mcp.run(transport="sse")로 실행
• 클라이언트(MultiServerMCPClient 등)는 transport: "sse" 설정으로 해당 엔드포인트에 연결
• 요청이 들어오면 get_weather()가 실행되고, 반환 결과(또는 스트리밍 중간 값)가 SSE 이벤트로 전송됩니다.

3. Client 생성

# client.py
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")

async def main():
    async with MultiServerMCPClient(
        {
            "weather": {
                "url": "http://localhost:8082/sse",
                "transport": "sse",
            }
        }
    ) as client:
        agent = create_react_agent(model, client.get_tools())

        response = await agent.ainvoke(
            {
                "messages": (
                    "뉴욕 날씨는?"
                )
            }
        )
        print(response)

if __name__ == "__main__":
    asyncio.run(main())

4. MCP SSE Flow