데이터 엔지니어링

잡담.

MCP 서버를 내가 외부에 제공한다고 했을 때, 여러 방법이 있을 것 같다.

가장 먼저 생각나는 건 서버 코드를 공유하고 stdio 실행 방식으로 제공하는 방식, 그리고 또 다른 방법은 웹서버를 띄워 StreamableHTTP 방식으로 제공하는 것이다.

MCP 서버를 직접 운영하면서 오픈한다는 의미는 인증과 인가를 붙이겠다는 의미일 것 같다.

AWS Lambda 환경에서 MCP 서버를 배포할 경우 아래의 장점이 있다.
- 인프라 관리가 불필요하여, 서버 운영 부담이 없다.
- 사용한 만큼 비용이 지불되기 때문에 비용 효율성이 높다.
- 자동으로 스케일링이 되기 때문에 트래픽에 따라 h/w 자원을 동적으로 변경할 수 있다.
- 보안 및 권한 관리가 쉽고, 도메인 주소관리, SSL 관리 등의 네트워크 리소스 운영 작업이 편하다.

단점도 있다.
- 최대 실행시간과 메모리 제한이 있고, 로깅과 제어가 제한적이다.

이 포스팅의 목적은 비교적 최근에 나온 MCP 서버를 AWS Lambda 환경에서 MCP 배포가 가능할지 확인해보기 위함이다.


그럼 시작해보자.

AWS Lambda 에서 MCP 서버 구축 해보기

A) 사전 준비

• fastMCP 로 구현한 mcp server 코드
• 컨테이너 이미지 로 생성할 Dockerfile 코드
• requirements.txt 패키지 설치 파일

B) 시나리오 

1. teams webhook 을 이용해 message 를 발송하는 MCP 서버 구축
2. mcp 서버를 컨테이너 이미지로 생성하고, ECR 레포지토리에 업로드
3. aws lambda 를 생성하고 ecr 에 업로드 된 이미지를 이용하여 배포 
4. 아래 두 가지 방식으로 MCP Server 테스트를 한다.

  a. Lambda 의 FunctionURL 을 이용해 MCP Client 로 연결하여 Teams 채널로 메세지를 전송해본다.
(LLM 모델은 로컬에서 테스트를 위해 Ollama 모델을 활용)
  b. cursor 의 MCP tool 로 등록해서 Teams 채널로 메세지를 전송해본다.

C) Step by Step

MCP 서버 코드 

from fastmcp import FastMCP
import os
import requests
import json
from typing import Optional, Dict, Any
WEBHOOK_URL = "웹훅 주소"
mcp = FastMCP("MCP-TEST-SERVER")

@mcp.tool()
async def get_my_info() -> str:
    """ 안녕하세요. 저는 러닝을 좋아하는 jssvs 입니다."""
    return f""

@mcp.tool()
async def get_weather() -> str:
    """현재 날씨를 조회합니다"""
    return f"오늘 날씨는 맑습니다."


@mcp.tool()
def send_teams_message(
    message: str,
    title: str = None,
) -> bool:
    """
    Microsoft Teams 채널에 webhook을 통해 메시지를 전송합니다.
    
    Args:
        message (str): 전송할 메시지 내용
        title (str, optional): 메시지 제목
    Returns:
        bool: 전송 성공 여부
    Example:
        send_teams_message(message="안녕하세요!", title="알림")
    """
    color = "00FF00"
    # MessageCard 형식으로 페이로드 구성
    payload = {
        "@type": "MessageCard",
        "@context": "https://schema.org/extensions",
        "summary": title or "알림",
        "sections": [{
            "activityTitle": title,
            "text": message
        }]
    }
    
    # 색상이 지정된 경우 추가
    if color:
        payload["themeColor"] = color
    
    # 제목이 없는 경우 섹션에서 제거
    if not title:
        del payload["sections"][0]["activityTitle"]
    
    try:
        response = requests.post(
            WEBHOOK_URL,
            headers={"Content-Type": "application/json"},
            data=json.dumps(payload)
        )
        
        # 성공 여부 확인
        if response.status_code == 200:
            print(f"✓ 메시지 전송 성공")
            return True
        else:
            print(f"✗ 메시지 전송 실패: {response.status_code}")
            print(f"응답: {response.text}")
            return False
            
    except Exception as e:
        print(f"✗ 오류 발생: {str(e)}")
        return False

def main():
    host = os.environ.get("HOST","0.0.0.0")
    port = int(os.environ.get("PORT",8080))
    print(f"Starting FastMCP HTTP Server on {host}:{port}")
    print("Available tools:")
    print("- get_bdp_team_info: Returns BDP team information")
    print("- get_weather: Returns weather information")
    print(f"Server accessible at: http://{host}:{port}")
    mcp.run(transport="http",host=host, port=port, stateless_http=True)

if __name__ == "__main__":
    main()

requirements.txt

fastmcp>=0.1.0
requests

Dockerfile

FROM python:3.12-slim
WORKDIR /app

# ### Environment Variables
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1


# Install the specified packages
COPY requirements.txt .
RUN pip install --no-cache-dir --upgrade pip && \
    pip install --no-cache-dir -r requirements.txt

# MCP 서버 코드 복사
COPY mcp_server.py .

# 환경 변수 설정
ENV PORT=8080
ENV AWS_LWA_INVOKE_MODE=response_stream
ENV AWS_LWA_READINESS_CHECK_PORT=8080
ENV AWS_LWA_READINESS_CHECK_PATH=/health

# === Lambda Web Adapter ===
COPY --from=public.ecr.aws/awsguru/aws-lambda-adapter:0.7.0 /lambda-adapter /opt/extensions/lambda-adapter

# 서버 시작 명령어
CMD ["python", "mcp_server.py"]

D) 참고해야 할 부분 !

1) Lambda 설정 시 참고 
-> 환경변수에 2개의 값 추가 (Dockerfile 에 ENV 에 추가해줬다면 생략해도 된다.)

2) 기본 TImeout 값으로는 MCP tool 호출이 되기 전에 호출이 종료되기 때문에 값을 조정한다.

E) 테스트 해보기

A) Strands Agent 코드를 작성하여 테스트 발송

from strands import Agent
from strands.models.ollama import OllamaModel
from strands.tools.mcp.mcp_client import MCPClient
from mcp.client.streamable_http import streamablehttp_client


client = MCPClient(
    lambda: streamablehttp_client(
        # url = "http://127.0.0.1:8080/mcp"
        url = "https://LambdaFunctinURL 주소.lambda-url.ap-northeast-2.on.aws/mcp"
    )
)
with client:
    tools = client.list_tools_sync()
    
    base_model = OllamaModel(
        host = "http://localhost:11434",
        model_id = "llama3.1:8b",
        temperature = 0.1
        )
    agent = Agent(tools = tools, model = base_model)
    # print(agent.tool.get_weather())
    result = agent("teams 채널에 아래 메세지를 전송해줘 \n message : Lambda 에서 구축한 MCP 서버 테스트 \n title : MCP 서버 테스트")
    print(result)

B) Cursor MCP 툴 연동하여 테스트 발송

결론

• lambda 의 장점은 서버리스, 즉 물리 서버를 운용하지 않으면서 서비스를 배포하는데 있다. 그렇다면 MCP 서버 역시 이런 장점을 누릴 수 있다.
• 내가 테스트한 부분은 운영 버전용으로 활용할 수 는 없다. 운영 버전까지 고려할 때는 보안을 위해 VPC 안에서 배포를 한다던지, 앞단에 Gateway 의 셋팅, 동시성 테스트 등 다양한 준비가 필요하다.
• 그리고 Dockerfile 을 보면 알겠지만, lambda web adapter 를 멀티스테이지 빌드를 사용하여 로드한 부분이 있는데, lambda 의 extension 기능으로 붙인 것이다. 즉 기능 추가, 의존성 등의 확장에 있어 제한이 있을 수 있음을 의미한다.
  
  

참고

[1] Welcome to FastMCP 2.0 - https://gofastmcp.com/getting-started/welcome 
[2] GitHub - awslabs/aws-lambda-web-adapter - https://github.com/awslabs/aws-lambda-web-adapter 
[3] Deploy MCP Server on AWS Lambda | Step-by-Step Guide - https://www.youtube.com/watch?v=JncV8hXuyRg&t=341s 
[4] 컨테이너 이미지로 Python Lambda 함수 배포 - Python용 AWS 기본 이미지 사용 - https://docs.aws.amazon.com/ko_kr/lambda/latest/dg/python-image.html#python-image-instructions

1. lambda 란?

간단하게 서버리스로 코드를 실행할 수 있는 aws 컴퓨팅 서비스라고 할 수 있다.

2. lambda 를 왜 쓰는가?

프로그램 실행을 위한 런타임 환경이나 실행가능한 서버가 필요없다.  - 서버리스

동시성등을 고려한 개발을 프로그래머가 직접 해줄 필요가 없다 - 옵션 에서 동시성이나 병렬 제어가 가능하기 때문에

모니터링이 편하다. 

데이터 엔지니어 입장에서 aws lambda 를 통한 스트리밍 처리도 가능하다고 생각한다. (허용되는 데이터 발생량이나 규모에 따라)

3. lambda 만들기

람다는 웹 콘솔에서 GUI 로도 손쉽게 만들 수 있다.

하지만 이번 포스팅은 로컬 서버에서 aws cli 를 이용하여 lambda 함수를 만들어 본다.

실습 예제는 aws 공식 홈페이지에 나와있는 내용들이다. ( 공식 홈페이지 가이드가 더 내용이 풍부하니까 그 쪽을 먼저 보길 바란다) 

a) 사전 준비 (시간이 부족해서 생략) 

• 로컬 PC 에서 aws cli 가 설치 및 IAM 구성이 되어 있어야 한다.
• lambda 생성과 실행 등 필요한 IAM 권한이 미리 준비되어 있어야 한다 .
• 실행권한 arn 이 필요하다.

b) lambda 생성

# 워킹 디렉토리 생성 및 이동
$ mkdir lambda_python_sample 
$ cd lambda_python_sample

# 람다 코드 작성
$ vi lambda_function.py  


# 사용되는 패키지를 working 디렉토리에 다운로드
$ pip install --target ./package requests 
$ cd package

# 패키지 압축
$ zip -r ../jssvs-development-package.zip .

# lambda 코드 압축
$ zip -g jssvs-deployment-package.zip lambda_function.py  

# lambda 생성 명령어  - 함수이름, 압축 파일 경로, 이벤트 핸들러 명(function 이름) # 역할 arn 순으로.
$ aws lambda create-function --function-name lambda-jssvs-dev \
--zip-file fileb://[home dir]/WorkSpace/aws/lambda_python_sample/jssvs-deployment-package.zip \ 
--handler lambda_function.main --runtime python3.7  \
--role arn:aws:iam::[ars code]:role/gamebi-lambda-role

-- lambda 코드 

import requests
def main(event, context):
    response = requests.get("https://www.test.com/")
    print(response.text)
    return response.text
if __name__ == "__main__":
    main('', '')

c) lambda 실행

# 기본 실행 및 출력 
$ aws lambda invoke --function-name lambda-jssvs-dev out \
--log-type Tail --query 'LogResult' --output text | base64 -d

lambda 의 출력 포맷은 base64로 인코딩 되어 있기 때문에, 우리가 보려면 base64 로 디코딩 해야 한다.

d) lambda 삭제 

$ aws lambda delete-function --function-name lambda-jssvs-dev

e) lambda 리스트 조회 및 검색

# 검색 
$ aws lambda list-functions --max-items 10
# 조회
$ aws lambda get-function --function-name lambda-jssvs-dev

** lambda 기본적으로 지원할 것 같지만 지원하지 않는 라이브러리들이 (requests 등 ) 있기 때문에, 개발자가 직접 소스를 올려줘야 한다.

** 웹 콘솔에서는 레이어 라는 이름으로 외부 패키지나 라이브러리를 올려서 사용할 수 있다.

** 함수를 생성할때 파이썬 버전은 꼭 맞춰주길 바란다. (다른 언어도 마찬가지. )   

1. Kinesis(키네시스)란?

리얼 타임 스트리밍 데이터를 수집하고, 처리하기 위한 도구. 분산 메시징 시스템

2. 파이프라인 

kinesis -> lambda -> s3 

3. 사전 준비 

• 키네시스 스트림이 생성되어 있어야 한다.
• 다음 권한을 가진 IAM 이 필요하다
  • 키네시스 CRUD
  • 키네시스와 S3에 접근 및 읽기 쓰기가 가능한 Lamda Role
• kiner를 이용하여 테스트 레코드를 전송한다.

4.코드

a) Lambda Code

from __future__ import print_function
import boto3
import base64
import json

print('Loading function')
AWS_BUCKET_NAME = 'jssvs-bucket'
s3 = boto3.client('s3')


# 람다 트리거 발생 시 호출
def lambda_handler(event, context):
    #print("Received event: " + json.dumps(event, indent=2))
    
    for record in event['Records']:
        # Kinesis data is base64 encoded so decode here
        
        #base64 decode 
        payload = base64.b64decode(record['kinesis']['data']).decode("UTF-8")
		
        #json 타입으로 읽어 변수에 저장
        jsonObj = json.loads(payload)
        
        # s3 적재에 필요한 변수 선언
        s3 = boto3.resource('s3')
        bucket=s3.Bucket(AWS_BUCKET_NAME)
        path='kinesis_test/parsed_log/'
        
        
        # 데이터를 S3 에 적재 
        s3.Object(AWS_BUCKET_NAME,path).put(Body=json.dumps(jsonObj[1]))
        
        print(jsonObj[1])
    return 'Successfully processed {} records.'.format(len(event['Records']))

b) 키네시스 Bulk Put Record

from uuid import uuid4
from kiner.producer import KinesisProducer

def on_flush(count, last_flushed_at, Data=b'', PartitionKey='', Metadata=()):
    print("Flushed {count} messages at timestamp {last_flushed_at}\nLast message was {Metadata['id']} paritioned by {PartitionKey} ({len(Data)} bytes)".format(count,last_flushed_at,Data,PartitonKey,Metadata))

p = KinesisProducer('[키네시스 이름]', flush_callback=on_flush)

for i in range(10):
    p.put_record('["dummy_log",{"jsonKey":"jsonValue"}]\n', metadata={'id': uuid4()}, partition_key=f"{i % 2}")



p.close()

** lambda 앞단에 firehose를 배치하여 파이프라인을 구성하는 방법도 있습니다.

** lambda에 키네시스 트리거 이벤트를 바로 붙히는데 레코드를 모아서 처리할 경우, baes64로 decode 된 값이 정상 포맷이 아닐 수 있습니다. deaggregation 과정이 필요하다는 이야기. 

*** kinesis , lambda 생성과정, 트리거 생성 부분등의 과정이 생략되었는데, 추후에 스크린샷을 떠서 올리겠습니다.  :D