프로덕션급 MCP 서버 직접 구현: 인증·스트리밍·상태 관리

저자: 吴佳浩(Alben)

공식 계정: 全栈架构师笔记

시리즈 칼럼: 《엔터프라이즈급 Agent 실전 가이드———MCP와 Agent Tools 엔지니어링 실전 적용》 · 제 02편

들어가며

공식 Demo의 Stdio 모드는 로컬 토이 프로젝트에서만 잠깐 돌려볼 수 있을 뿐이고, 실제로 기업 인트라넷에 들어가려면 MCP Server는 반드시 인증, 멀티테넌트 격리, 연결 풀과 헬스 체크를 갖춘 독립 마이크로서비스여야 한다.

많은 사람들은 MCP Server를 작성하는 것이 그저 FastMCP 데코레이터를 감싸는 것이라고 생각한다. 하지만 고동시성 프로덕션 환경에서는 연결 누수, 잡히지 않은 예외로 인한 자식 프로세스 급사, 그리고 멀티테넌트 Token 월권이 가장 치명적인 보이지 않는 킬러다.

Demo에서 프로덕션으로 나아가는 핵심은 무상태 HTTP 요청을 제어 가능한 Long-Session 상태 머신으로 감싸는 데 있다.

제 01편에서 우리는 MCP 프로토콜의 설계 철학을 분해해 보았다. 많은 엔지니어는 공식 문서를 본 뒤 보통 공식 Python SDK로 Stdio 기반 스크립트를 작성한다:

전형적인 로컬 Demo 작성법 (기업 프로덕션에 바로 사용 불가) from mcp.server.fastmcp import FastMCP mcp = FastMCP( "DemoServer" ) @mcp.tool() def query_db ( sql: str ) -> str : # 데이터베이스에 그대로 연결, 인증 없음, 테넌트 격리 없음, 타임아웃 제어 없음 return db.execute(sql) mcp.run() # 기본적으로 stdio로 실행

이러한 작성법은 로컬 테스트에서는 매우 매끄럽지만, 일단 이 MCP Server를 기업용 Kubernetes 클러스터에 배포하여 전사 수십 개의 Agent가 동시에 호출하도록 하면, 즉시 세 가지 프로덕션급 재앙을 맞닥뜨리게 된다:

재앙 현상 | 구체적 양상 | 아키텍처 근본 원인 1. 크로스 테넌트 월권 침투 (Tenant Leakage) | 연구개발부의 Agent가 재무부의 급여표를 조회하여 심각한 컴플라이언스 심사 사고를 유발 | Request 단위 기반의 Tenant 컨텍스트 주입과 동적 연결 풀 격리 부재 2. 장기 연결 눈사태 (Connection Exhaust) | 여러 Agent가 동시에 SSE 장기 연결을 수립하여 서버 파일 핸들 고갈로 인한 크래시 발생 | 연결 하트비트 킵얼라이브, 세션 재사용과 좀비 세션(Ghost Session) 퇴출 부재 3. 예외로 인한 전역 급사 (Process Crash) | 특정 Tool 실행에서 잡히지 않은 예외가 발생하여 전체 MCP 프로세스가 곧바로 종료되어 서비스 중단 | 전역 오류 배리어와 JSON-RPC 표준 오류 코드 캡슐화 부재

엔터프라이즈급 MCP Server를 구축하려면 반드시 Streamable HTTP / SSE 채널을 지원하고, 완전한 인증 인터셉트, 테넌트 동적 격리와 라이프사이클 관리를 갖추어야 한다.

一、엔터프라이즈급 MCP Server의 마이크로서비스 아키텍처 토폴로지

프로덕션 환경에서 MCP Server는 결코 단일 머신 스크립트가 아니라 표준 클라우드 네이티브 마이크로서비스다:

- 🔸 이중 채널 아키텍처(SSE + HTTP POST): 클라이언트는 /sse 엔드포인트를 통해 Server-Sent Events 장기 연결을 수립하여 하향 이벤트를 수신하고, /messages?session_id=xxx 를 통해 상향 JSON-RPC 요청을 전송한다;

- 🔸 세션 라이프사이클 격리: 각 Agent 연결마다 유일한 session_id 를 할당하여 여러 Agent 간의 상태가 서로 간섭하지 않도록 보장한다;

- 🔸 멀티테넌트 컨텍스트 투과: 게이트웨이 인증 후의 tenant_id 를 비동기 코루틴 컨텍스트(ContextVar)에 주입하고, 하위 데이터베이스 연결 풀이 테넌트별로 동적 라우팅한다.

이 장의 핵심 관점을 한마디로 정리하면:

프로덕션급 MCP Server의 본질은 전통적인 Web API를 JSON-RPC 2.0 규격과 장기 연결 세션 프로토콜에 부합하는 마이크로서비스로 감싸는 것이다.

二、프로덕션급 코드 구현: FastAPI 기반 엔터프라이즈급 MCP Server

다음 예제는 Python 3.11+ 와 FastAPI를 기반으로 엔터프라이즈급 MCP Server를 구축하고, 네이티브 StreamingResponse 로 Server-Sent Events(SSE) 데이터 전송을 구현한다. 예제는 엔터프라이즈급 MCP 서버 측의 핵심 역량을 완전히 시연하며, 멀티테넌트 인증, 도구 등록, JSON-RPC 프로토콜 처리, SSE 양방향 통신, 세션 관리 그리고 도구 예외 격리 등의 핵심 모듈을 포함한다.

설명하자면, 이 예제는 주로 엔터프라이즈급 MCP Server의 전체 아키텍처 설계와 핵심 구현 사상을 보여주기 위한 것이다. 실제 프로덕션 환경에서는 나아가 OAuth/JWT 인증, 세션 회수(Session GC), 속도 제한·회로 차단, 로그 감사, 관측성(OpenTelemetry), 고가용성 배포 등의 역량을 결합하여 완전한 엔터프라이즈급 MCP 서비스 체계를 구축해야 한다. 예제는 FastAPI 네이티브 StreamingResponse 로 SSE를 수작업 구현했기 때문에 별도의 서드파티 SSE 라이브러리 의존 없이 프로토콜 통신을 완성할 수 있다. FastAPI는 더 높은 수준의 SSE 지원도 제공하므로 프로젝트 요구에 따라 선택하여 사용할 수 있다.

""" enterprise_mcp_server.py 프로덕션급 엔터프라이즈 MCP Server 구현 포함 사항: - Streamable HTTP / SSE 이중 채널 - JWT 인증 - 멀티테넌트 컨텍스트 주입 - Tool 등록 - Tool 예외 격리 """ import asyncio import json import uuid from contextvars import ContextVar from typing import Any , Dict , List from fastapi import Depends, FastAPI, Header, HTTPException, Request, status from fastapi.responses import JSONResponse, StreamingResponse from pydantic import BaseModel

app = FastAPI(title= "Enterprise MCP Server" , version= "1.0.0" )

current_tenant_id: ContextVar[ str ] = ContextVar( "current_tenant_id" , default= "default" )

class SessionContext : def __init__ ( self, session_id: str , tenant_id: str ): self.session_id = session_id self.tenant_id = tenant_id self.queue: asyncio.Queue = asyncio.Queue() self.last_active = asyncio.get_event_loop().time()

active_sessions: Dict [ str , SessionContext] = {}

class ToolDefinition ( BaseModel ): name: str description: str inputSchema: Dict [ str , Any ]

REGISTERED_TOOLS: Dict [ str , Any ] = {} TOOL_SCHEMAS: List [ToolDefinition] = []

def mcp_tool ( name: str , description: str , schema: Dict [ str , Any ] ): def decorator ( fn ): REGISTERED_TOOLS[name] = fn TOOL_SCHEMAS.append( ToolDefinition( name=name, description=description, inputSchema=schema, ) ) return fn return decorator

@mcp_tool( name= "query_tenant_metrics" , description= "Query production metrics for the authenticated tenant safely." , schema={ "type" : "object" , "properties" : { "metric_name" : { "type" : "string" , "description" : "e.g. qps, error_rate, latency" , }, "time_range" : { "type" : "string" , "description" : "e.g. 1h, 24h, 7d" , }, }, "required" : [ "metric_name" ], }, ) async def query_tenant_metrics ( metric_name: str , time_range: str = "1h" ) -> str : tenant = current_tenant_id.get() return json.dumps( { "tenant_id" : tenant, "metric" : metric_name, "time_range" : time_range, "data" : { "avg_value" : 142.5 , "p99_latency_ms" : 23.4 , "status" : "healthy" , }, } )

async def auth_middleware ( x_tenant_id: str = Header( ..., alias= "X-Tenant-ID" ), authorization: str = Header( ..., alias= "Authorization" ), ) -> str : if not authorization.startswith( "Bearer " ) or not x_tenant_id: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail= "Unauthorized: Missing valid Bearer token or Tenant Header" , ) current_tenant_id. set (x_tenant_id) return x_tenant_id

@app.get( "/sse" ) async def sse_endpoint ( request: Request, tenant_id: str = Depends( auth_middleware ) ): session_id = str (uuid.uuid4()) session_ctx = SessionContext(session_id=session_id, tenant_id=tenant_id) active_sessions[session_id] = session_ctx

async def event_generator (): try : yield ( "event: endpoint\n" f"data: /messages?session_id= {session_id} \n\n" ) while True : if await request.is_disconnected(): break try : msg = await asyncio.wait_for( session_ctx.queue.get(), timeout= 15.0 ) yield ( "event: message\n" f"data: {json.dumps(msg)} \n\n" ) except asyncio.TimeoutError: yield ": ping\n\n" finally : active_sessions.pop(session_id, None )

return StreamingResponse( event_generator(), media_type= "text/event-stream" , headers={ "Cache-Control" : "no-cache" , "Connection" : "keep-alive" , "X-Accel-Buffering" : "no" , }, )

@app.post( "/messages" ) async def message_endpoint ( request: Request, session_id: str , tenant_id: str = Depends( auth_middleware ), ): session_ctx = active_sessions.get(session_id) if not session_ctx: raise HTTPException(status_code= 404 , detail= "Session not found or expired" )

current_tenant_id. set (tenant_id) payload = await request.json() method = payload.get( "method" ) req_id = payload.get( "id" )

if method == "initialize" : await session_ctx.queue.put( { "jsonrpc" : "2.0" , "id" : req_id, "result" : { "protocolVersion" : "2024-11-05" , "capabilities" : { "tools" : {}}, "serverInfo" : { "name" : "EnterpriseProductionMCPServer" , "version" : "1.0.0" , }, }, } ) return JSONResponse({ "status" : "accepted" })

if method == "tools/list" : await session_ctx.queue.put( { "jsonrpc" : "2.0" , "id" : req_id, "result" : { "tools" : [t.model_dump() for t in TOOL_SCHEMAS], }, } ) return JSONResponse({ "status" : "accepted" })

if method == "tools/call" : params = payload.get( "params" , {}) tool_name = params.get( "name" ) arguments = params.get( "arguments" , {}) fn = REGISTERED_TOOLS.get(tool_name)

if fn is None : response = { "jsonrpc" : "2.0" , "id" : req_id, "error" : { "code" : - 32601 , "message" : f"Tool ' {tool_name} ' not found" , }, } else : try : result = await fn(**arguments) response = { "jsonrpc" : "2.0" , "id" : req_id, "result" : { "content" : [ { "type" : "text" , "text" : str (result) } ] }, } except Exception as e: response = { "jsonrpc" : "2.0" , "id" : req_id, "error" : { "code" : - 32000 , "message" : f"Execution Error: {e} " , }, }

await session_ctx.queue.put(response) return JSONResponse({ "status" : "accepted" })

return JSONResponse({ "status" : "ignored" })

三、프로덕션급 MCP Server의 4대 함정 회피 가이드

MCP Server를 기업 사설 클라우드에 배포할 때, 팀은 반드시 다음 네 가지 전형적인 심수구(深水区) 문제를 철저히 경계해야 한다:

- 🔸 NGINX / 게이트웨이의 Buffer 버퍼링으로 인한 SSE 연결 끊김: 리버스 프록시 서버는 기본적으로 응답 버퍼링(Buffering)을 켜기 때문에 클라이언트가 롱 커넥션 이벤트를 실시간으로 받을 수 없다. 반드시 응답 헤더에 X-Accel-Buffering: no 를 명시적으로 추가해야 한다;

- 🔸 코루틴 동시성 환경에서의 테넌트 데이터 혼선: 전역 변수에 tenant_id 를 임시 저장하는 것을 엄격히 금지하며, 비동기 Python 환경에서는 반드시 contextvars.ContextVar 를 사용해야 하고, 비동기 스케줄링 전환 시 테넌트 경계가 빈틈없이 격리되도록 보장해야 한다;

- 🔸 좀비 연결로 인한 메모리 누수: asyncio.TimeoutError 기반의 하트비트(Ping) 메커니즘을 반드시 설계해야 하며, 클라이언트가 비정상적으로 네트워크가 끊겨 종료 신호를 보내지 않았을 때 active_sessions 딕셔너리를 제때 정리해야 한다;

- 🔸 도구 타임아웃의 하드 서킷 브레이커 메커니즘: 모든 도구 호출에는 반드시 타임아웃 데코레이터(예: 30s 서킷 브레이커)를 씌워야 하며, 백엔드에서 멈춰버린 특정 SQL 쿼리가 전체 워커 스레드 풀을 끌어내리는 것을 방지해야 한다.

이번 편 요약

- 🔸 Stdio는 로컬 극客 디버깅에만 사용하고, 기업용 미드플랫폼은 반드시 Streamable HTTP/SSE 듀얼 채널 아키텍처를 도입해야 한다;

- 🔸 ContextVar와 의존성 주입을 통해 멀티 테넌트 컨텍스트의 절대적인 물리적/논리적 격리를 구현한다;

- 🔸 전역 예외 장벽과 JSON-RPC 표준 오류 래핑을 구축하여 프로세스가 예기치 않게 크래시하는 것을 방지한다;

- 🔸 NGINX 제로 버퍼링과 하트비트 keep-alive를 설정하여 프로덕션 환경의 롱 커넥션 눈사태를 원천 차단한다.

고가용성 MCP Server를 작성하는 방법을 익혔으니, 다음 핵심 문제는 이것이다: Agent가 강력한 도구 호출 능력을 갖추게 되었을 때, 민감한 명령을 실행할 때 프로덕션 환경을 훼손하지 않도록 어떻게 보장할 것인가?

다음 편에서는 심도 있게 다룬다: 《Tool의 안전성과 실행 샌드박스: Docker에서 gVisor까지의 방어 아키텍처》!

여러분, 이번 편은 《기업级 Agent 실전 가이드》· 제2장의 두 번째 글입니다. 앞으로 agent 개발의 전체 과정을 계속 업데이트할 예정이니, Agent 개발에 관심이 있다면 이 컬렉션을 팔로우해 보시기 바랍니다.

吴佳浩Alben

AI扫地僧 @China

113

글

205k

조회수

379

팬