AWS Lambda 및 ECS Fargate용 프로덕션 준비 MCP 서버 프레임워크
sipap-mcp는 JSON-RPC 2.0을 구현하고 다음 환경에서 실행할 수 있는 Model Context Protocol(MCP) 서버 구축을 위한 기본 클래스와 인프라를 제공합니다:
AWS Lambda: 가볍고 산발적인 워크로드를 위한 서버리스 함수
ECS Fargate: 장기 실행 상태 저장 워크로드를 위한 컨테이너화된 서비스
이 프레임워크는 Valo(스포츠 인텔리전스 플랫폼) 아키텍처의 5개 데이터 서버를 모두 지원하며, 스포츠 데이터, 배당률 인텔리전스, 뉴스 컨텍스트, 날씨 데이터, 과거 통계를 처리합니다.
Related MCP server: mcp-server-toolkit
✅ MCPServer 기본 클래스: 도구 등록 및 자동 발견 기능이 있는 추상 기본 클래스
✅ @mcp_tool 데코레이터: JSON Schema 검증과 함께 함수를 MCP 도구로 표시
✅ JSON-RPC 2.0 프로토콜: 적절한 오류 처리를 갖춘 완전한 구현
✅ 이중 전송: Lambda 핸들러 및 FastAPI HTTP 서버
✅ 인증: 플러그형 전략 (NoAuth, API 키, AWS SigV4)
✅ 세션 관리: 호출 간 상태 보존을 위한 Redis 백업
✅ 입력 검증: 모든 도구 입력에 대한 JSON Schema 검증
✅ 타입 안전성: 전체 mypy strict 모드 준수 (오류 0건)
✅ 테스트 커버리지: 112개 통과 테스트로 96% 커버리지
✅ 프로덕션 준비: 린트 오류 0건, 포괄적인 오류 처리
pip install sipap-mcp개발용:
pip install sipap-mcp[dev]from sipap_mcp import MCPServer, mcp_tool
class WeatherMCP(MCPServer):
"""Weather data MCP server."""
def __init__(self):
super().__init__(name="weather-mcp", version="1.0.0")
@mcp_tool(
description="Get current weather for a location",
input_schema={
"type": "object",
"properties": {
"location": {"type": "string", "description": "City name"},
"units": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"default": "celsius"
}
},
"required": ["location"]
}
)
def get_weather(self, location: str, units: str = "celsius") -> dict:
"""Get current weather conditions."""
# Your implementation here
return {
"location": location,
"temperature": 22 if units == "celsius" else 72,
"units": units,
"condition": "partly cloudy"
}from sipap_mcp.transport import create_lambda_handler
from sipap_mcp.auth import APIKeyAuth
# Create server instance
server = WeatherMCP()
# Configure authentication
auth = APIKeyAuth(api_keys=["your-api-key"])
# Create Lambda handler (entry point for AWS)
handler = create_lambda_handler(server, auth=auth)AWS CDK 또는 Terraform으로 배포:
핸들러: your_module.handler
런타임: python3.12
타임아웃: 30초
from sipap_mcp.transport import create_http_app
import uvicorn
# Create server instance
server = WeatherMCP()
# Create FastAPI app
app = create_http_app(server, auth=auth)
# Run with uvicorn
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)Docker로 배포:
FROM python:3.12-slim
WORKDIR /app
COPY . .
RUN pip install sipap-mcp
CMD ["uvicorn", "your_module:app", "--host", "0.0.0.0", "--port", "8000"]도구는 @mcp_tool로 데코레이션된 함수로, MCP 프로토콜을 통해 호출 가능해집니다:
@mcp_tool(
description="Description of what this tool does",
input_schema={
"type": "object",
"properties": {
"param": {"type": "string"}
},
"required": ["param"]
}
)
def my_tool(self, param: str) -> dict:
"""Docstring for the tool."""
return {"result": param}지원되는 JSON Schema 타입:
string, number, integer, boolean, array, object
검증: minLength, maxLength, minimum, maximum, pattern, enum
배포 환경에 맞는 인증 전략을 선택하세요:
from sipap_mcp.auth import NoAuth
auth = NoAuth() # No authentication - use for local dev onlyfrom sipap_mcp.auth import APIKeyAuth
auth = APIKeyAuth(api_keys=[
"client-a-key",
"client-b-key",
"client-c-key"
])클라이언트는 X-API-Key 헤더에 API 키를 전송합니다.
from sipap_mcp.auth import SigV4Auth
auth = SigV4Auth(service="lambda", region="us-east-1")IAM 인증이 있는 Lambda Function URL용.
Redis를 사용하여 여러 요청에 걸쳐 상태를 유지합니다:
import redis
from sipap_mcp.session import SessionManager
# Connect to Redis
redis_client = redis.Redis(host="localhost", port=6379)
# Create session manager
session_manager = SessionManager(
redis_client=redis_client,
ttl=3600 # 1 hour default
)
# Create session
session_id = session_manager.create_session(
data={"user_id": "123", "preferences": {...}},
ttl=1800 # 30 minutes custom TTL
)
# Retrieve session
session_data = session_manager.get_session(session_id)
# Update session
session_manager.update_session(session_id, updated_data)
# Extend TTL
session_manager.extend_ttl(session_id, ttl=7200)리소스 관리를 위해 _setup() 및 _cleanup()을 재정의하세요:
class MyServer(MCPServer):
def __init__(self):
super().__init__(name="my-server", version="1.0.0")
self.db_connection = None
def _setup(self) -> None:
"""Called when entering context manager."""
self.db_connection = connect_to_database()
def _cleanup(self) -> None:
"""Called when exiting context manager."""
if self.db_connection:
self.db_connection.close()컨텍스트 관리자와 함께 사용:
with server:
# Server is set up, resources initialized
response = server.handle_request(request)
# Cleanup happens automatically on exit{
"jsonrpc": "2.0",
"id": "req-1",
"method": "tools/list",
"params": {}
}응답:
{
"jsonrpc": "2.0",
"id": "req-1",
"result": {
"tools": [
{
"name": "get_weather",
"description": "Get current weather for a location",
"inputSchema": {
"type": "object",
"properties": {...},
"required": [...]
}
}
]
}
}{
"jsonrpc": "2.0",
"id": "req-2",
"method": "tools/call",
"params": {
"name": "get_weather",
"arguments": {
"location": "London",
"units": "celsius"
}
}
}응답:
{
"jsonrpc": "2.0",
"id": "req-2",
"result": {
"content": [{
"type": "text",
"text": "{\"location\": \"London\", \"temperature\": 15, ...}"
}]
}
}표준 JSON-RPC 2.0 오류 코드:
코드 | 의미 | 발생 시점 |
-32700 | 구문 분석 오류 | 잘못된 JSON |
-32600 | 잘못된 요청 | 필수 필드 누락 |
-32601 | 메서드를 찾을 수 없음 | 알 수 없는 메서드 |
-32602 | 잘못된 매개변수 | 검증 실패 |
-32603 | 내부 오류 | 서버 오류 |
오류 응답:
{
"jsonrpc": "2.0",
"id": "req-3",
"error": {
"code": -32602,
"message": "Invalid params: 'location' is required"
}
}포괄적인 예제는 examples/ 디렉토리를 참조하세요:
예제 | 설명 |
간단한 계산기 서버 | |
API 키 인증이 있는 Lambda 배포 | |
Redis 세션이 있는 HTTP 서버 | |
고급 패턴 및 수명 주기 훅 | |
모든 인증 전략 |
예제 실행:
python examples/01_basic_server.py
python examples/02_lambda_with_auth.py
python examples/03_http_with_sessions.py # Requires Redis이 프레임워크는 Sentinel 아키텍처의 검증된 패턴을 적용합니다:
ExitStack + Generator 패턴: 컨텍스트 관리자를 사용한 리소스 관리
도구 자동 발견: 인트로스펙션 기반 도구 등록
구조화된 출력 강제: 모든 입력/출력에 대한 JSON Schema 검증
ContextVar 기반 로깅: 스레드 안전 컨텍스트 전파
sipap_mcp/
├── core/
│ ├── protocol.py # JSON-RPC 2.0 implementation
│ └── server.py # MCPServer base class
├── decorators/
│ └── tool.py # @mcp_tool decorator & registry
├── transport/
│ ├── lambda_handler.py # AWS Lambda adapter
│ └── http_handler.py # FastAPI adapter
├── auth/
│ └── middleware.py # Authentication strategies
├── session/
│ └── manager.py # Redis session management
└── validation/
└── schema.py # JSON Schema validation# Clone repository
git clone <repo-url>
cd sipap-mcp
# Create virtual environment
python3.12 -m venv .venv
source .venv/bin/activate
# Install in editable mode with dev dependencies
pip install -e ".[dev]"# Run all tests
pytest
# Run with coverage
pytest --cov=src/sipap_mcp --cov-report=html
# Open coverage report
open htmlcov/index.html커밋 전에 모든 품질 게이트를 통과해야 합니다:
# Type checking (strict mode)
mypy src/sipap_mcp --strict
# Linting
ruff check src/sipap_mcp tests/
# Auto-fix linting errors
ruff check --fix src/sipap_mcp tests/
# All gates at once
pytest && mypy src/sipap_mcp --strict && ruff check src/sipap_mcp tests/# Build wheel and source distribution
python -m build
# Install built package
pip install dist/sipap_mcp-0.1.0-py3-none-any.whlPython 3.12, 3.13 또는 3.14
pydantic >= 2.7.0
fastapi >= 0.111.0
uvicorn[standard] >= 0.30.0
jsonschema >= 4.22.0
sipap-common >= 0.1.0
typing-extensions >= 4.12.0
pytest >= 8.0.0
pytest-cov >= 5.0.0
mypy >= 1.10.0
ruff >= 0.4.0
class MCPServer(name: str, version: str)메서드:
handle_request(request_data) -> dict: JSON-RPC 요청 처리
list_tools() -> list[dict]: 등록된 도구 가져오기
get_info() -> dict: 서버 메타데이터 가져오기
_setup() -> None: 초기화를 위해 재정의 (선택 사항)
_cleanup() -> None: 정리를 위해 재정의 (선택 사항)
@mcp_tool(description: str, input_schema: dict)
def tool_function(self, **kwargs) -> dict:
pass매개변수:
description: 사람이 읽을 수 있는 도구 설명
input_schema: 입력 검증을 위한 JSON Schema
class SessionManager(redis_client, ttl: int = 3600)메서드:
create_session(data, ttl=None) -> str: 세션 생성, ID 반환
get_session(session_id) -> dict | None: 세션 데이터 검색
update_session(session_id, data, ttl=None) -> bool: 세션 업데이트
delete_session(session_id) -> bool: 세션 삭제
session_exists(session_id) -> bool: 존재 여부 확인
extend_ttl(session_id, ttl) -> bool: 만료 시간 연장
create_lambda_handler(server, auth=None) -> Callable
create_http_app(server, auth=None) -> FastAPIdef test_my_server():
server = MyServer()
# Test tool listing
tools = server.list_tools()
assert len(tools) > 0
# Test tool execution
request = {
"jsonrpc": "2.0",
"id": "1",
"method": "tools/call",
"params": {
"name": "my_tool",
"arguments": {"param": "value"}
}
}
with server:
response = server.handle_request(request)
assert "result" in responsedef test_lambda_handler():
from sipap_mcp.transport import create_lambda_handler
server = MyServer()
handler = create_lambda_handler(server)
event = {
"headers": {},
"body": json.dumps({
"jsonrpc": "2.0",
"id": "1",
"method": "tools/list",
"params": {}
})
}
response = handler(event, {})
assert response["statusCode"] == 200핸들러 설정:
# app.py
from sipap_mcp import MCPServer, mcp_tool
from sipap_mcp.transport import create_lambda_handler
from sipap_mcp.auth import APIKeyAuth
import os
class MyServer(MCPServer):
# ... server definition ...
server = MyServer()
auth = APIKeyAuth(api_keys=os.getenv("API_KEYS", "").split(","))
handler = create_lambda_handler(server, auth=auth)배포:
핸들러: app.handler
런타임: python3.12
메모리: 512 MB (워크로드에 따라 조정)
타임아웃: 30초 (도구 실행 시간에 따라 조정)
환경 변수: API_KEYS=key1,key2,key3
Dockerfile:
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]app.py:
from sipap_mcp.transport import create_http_app
# ... server definition ...
app = create_http_app(server, auth=auth)작업 정의:
컨테이너 포트: 8000
상태 확인: /health (구현된 경우)
CPU: 256 (.25 vCPU)
메모리: 512 MB
개발:
docker run -d -p 6379:6379 redis:7-alpine프로덕션:
AWS ElastiCache for Redis
버전: Redis 7.x
노드 유형: cache.t4g.micro (또는 더 큰 것)
암호화: 전송 중 및 저장 시
Multi-AZ: 프로덕션용으로 활성화
가져오기 오류:
# Problem
from sipap_mcp import MCPServer # ImportError
# Solution
pip install sipap-mcp인증 실패:
# Check API key header name (must be X-API-Key)
headers = {"X-API-Key": "your-key"} # Correct
headers = {"Api-Key": "your-key"} # Wrong세션을 찾을 수 없음:
# Sessions expire after TTL
session_manager.session_exists(session_id) # Check first
session_manager.extend_ttl(session_id, 3600) # Extend if needed타입 오류:
# Run mypy to catch type issues
mypy your_module.py --strictAWS Lambda (512 MB, Python 3.12)에서 테스트:
작업 | 콜드 스타트 | 웜 스타트 |
tools/list | 850ms | 12ms |
tools/call (simple) | 900ms | 15ms |
tools/call (with DB) | 1200ms | 45ms |
콜드 스타트 줄이기: Lambda 프로비저닝된 동시성 사용
연결 캐시: _setup()에서 초기화하고 호출 간 재사용
의존성 최소화: 필요한 것만 가져오기
async 사용: FastAPI 전송은 async 도구를 지원
세션 TTL: 메모리 사용량과 사용자 경험의 균형
기여를 환영합니다! 다음을 지켜주세요:
기존 코드 스타일 준수 (ruff + mypy strict)
새 기능에 대한 테스트 추가 (80% 이상 커버리지 유지)
문서 업데이트
제출 전에 모든 품질 게이트 실행
Copyright © 2026 Valo Team
다음으로 구축됨:
테스트 주도 개발 (TDD)
타입 안전성 (mypy strict 모드)
96% 테스트 커버리지 (112개 테스트)
프로덕션 준비 오류 처리
포괄적인 문서
Valo 플랫폼의 일부 - 스포츠 인텔리전스 및 결과 확률 평가 플랫폼
This server cannot be deployed
MCP server for progressive tool usage at any scale (see https://klavis.ai)
MCP server for Superserve sandboxes: create, exec, and manage Firecracker microVMs
MCP server for interacting with the Supabase platform
An MCP server that let you interact with Cycloid.io Internal Development Portal and Platform
