79 KiB
MCP 개발 모범 사례
(위 이미지 클릭 시 이 수업의 영상 시청)
개요
이 수업은 MCP 서버와 기능을 프로덕션 환경에서 개발, 테스트, 배포하기 위한 고급 모범 사례에 중점을 둡니다. MCP 생태계가 점점 복잡해지고 중요해짐에 따라, 확립된 패턴을 따르면 신뢰성, 유지 관리성 및 상호 운용성을 확보할 수 있습니다. 이 수업은 실제 MCP 구현에서 얻은 실용적인 지혜를 모아 견고하고 효율적인 서버를 효과적인 리소스, 프롬프트, 도구와 함께 만드는 데 안내합니다.
학습 목표
이 수업을 마치면 다음을 할 수 있습니다:
- MCP 서버 및 기능 설계에서 업계 모범 사례 적용
- MCP 서버에 대한 포괄적인 테스트 전략 수립
- 복잡한 MCP 애플리케이션을 위한 효율적이고 재사용 가능한 워크플로 패턴 설계
- MCP 서버에서 적절한 오류 처리, 로깅 및 가시성 구현
- 성능, 보안 및 유지 관리를 위한 MCP 구현 최적화
MCP 핵심 원칙
특정 구현 관행을 다루기 전에, 효과적인 MCP 개발을 이끄는 핵심 원칙을 이해하는 것이 중요합니다:
-
표준화된 통신: MCP는 JSON-RPC 2.0을 기반으로 하여 모든 구현에서 요청, 응답, 오류 처리를 위한 일관된 형식을 제공합니다.
-
사용자 중심 설계: 항상 사용자 동의, 제어 및 투명성을 우선시하십시오.
-
보안 우선: 인증, 권한 부여, 검증 및 속도 제한을 포함한 강력한 보안 조치를 구현하십시오.
-
모듈식 아키텍처: 각 도구와 리소스가 명확하고 집중된 목적을 가지도록 모듈식 접근법으로 MCP 서버를 설계하십시오.
-
상태 유지 연결: 다중 요청 간 상태를 유지하는 MCP의 능력을 활용하여 보다 일관되고 맥락 인식이 가능한 상호작용을 제공합니다.
공식 MCP 모범 사례
다음 모범 사례는 공식 모델 컨텍스트 프로토콜 문서에서 파생되었습니다:
보안 모범 사례
-
사용자 동의 및 제어: 데이터 접근이나 작업 수행 전 반드시 명시적인 사용자 동의를 요구하십시오. 공유되는 데이터와 허용된 작업에 대해 명확한 제어를 제공하십시오.
-
데이터 프라이버시: 명시적인 동의가 있는 경우에만 사용자 데이터를 노출하고 적절한 접근 제어로 보호하십시오. 무단 데이터 전송을 방지하십시오.
-
도구 안전성: 어떤 도구를 호출하기 전에 명시적인 사용자 동의를 요구하십시오. 사용자에게 각 도구의 기능을 이해시키고 강력한 보안 경계를 적용하십시오.
-
도구 권한 제어: 세션 중 모델이 사용할 수 있는 도구를 구성하여 명시적으로 허가된 도구만 액세스할 수 있도록 하십시오.
-
인증: API 키, OAuth 토큰 또는 기타 안전한 인증 방법을 사용하여 도구, 리소스 또는 민감한 작업에 접근하기 전에 적절한 인증을 요구하십시오.
-
매개변수 검증: 잘못되었거나 악의적인 입력이 도구 구현에 도달하지 않도록 모든 도구 호출에 대해 검증을 시행하십시오.
-
속도 제한: 남용을 방지하고 서버 자원의 공정한 사용을 보장하기 위해 속도 제한을 구현하십시오.
구현 모범 사례
-
능력 협상: 연결 설정 중 지원 기능, 프로토콜 버전, 사용 가능한 도구 및 리소스에 관한 정보를 교환하십시오.
-
도구 설계: 여러 문제를 처리하는 단일 대형 도구 대신 하나의 작업에 집중하는 도구를 만드십시오.
-
오류 처리: 표준화된 오류 메시지와 코드를 구현하여 문제를 진단하고 오류를 우아하게 처리하며 실행 가능한 피드백을 제공하십시오.
-
로깅: 프로토콜 상호작용을 감사, 디버깅, 모니터링하기 위해 구조화된 로그를 구성하십시오.
-
진행 추적: 장기 실행 작업의 경우, 반응형 사용자 인터페이스를 가능하게 하기 위해 진행 상황 업데이트를 보고하십시오.
-
요청 취소: 더 이상 필요하지 않거나 너무 오래 걸리는 대기 중인 요청을 클라이언트가 취소할 수 있도록 허용하십시오.
추가 참고 자료
최신 MCP 모범 사례 정보는 다음을 참고하십시오:
- MCP Documentation
- MCP Specification (2025-11-25)
- GitHub Repository
- Security Best Practices
- OWASP MCP Top 10 - 보안 위험 및 완화책
- MCP Security Summit Workshop (Sherpa) - 실습 보안 교육
실용적인 구현 예제
도구 설계 모범 사례
1. 단일 책임 원칙
각 MCP 도구는 명확하고 집중된 목적을 가져야 합니다. 여러 문제를 다루려고 하는 일체형 도구보다 특정 작업에 탁월한 전문 도구를 개발하십시오.
// A focused tool that does one thing well
public class WeatherForecastTool : ITool
{
private readonly IWeatherService _weatherService;
public WeatherForecastTool(IWeatherService weatherService)
{
_weatherService = weatherService;
}
public string Name => "weatherForecast";
public string Description => "Gets weather forecast for a specific location";
public ToolDefinition GetDefinition()
{
return new ToolDefinition
{
Name = Name,
Description = Description,
Parameters = new Dictionary<string, ParameterDefinition>
{
["location"] = new ParameterDefinition
{
Type = ParameterType.String,
Description = "City or location name"
},
["days"] = new ParameterDefinition
{
Type = ParameterType.Integer,
Description = "Number of forecast days",
Default = 3
}
},
Required = new[] { "location" }
};
}
public async Task<ToolResponse> ExecuteAsync(IDictionary<string, object> parameters)
{
var location = parameters["location"].ToString();
var days = parameters.ContainsKey("days")
? Convert.ToInt32(parameters["days"])
: 3;
var forecast = await _weatherService.GetForecastAsync(location, days);
return new ToolResponse
{
Content = new List<ContentItem>
{
new TextContent(JsonSerializer.Serialize(forecast))
}
};
}
}
2. 일관된 오류 처리
정보성 오류 메시지와 적절한 복구 메커니즘으로 강력한 오류 처리를 구현하십시오.
# 포괄적인 오류 처리가 포함된 파이썬 예제
class DataQueryTool:
def get_name(self):
return "dataQuery"
def get_description(self):
return "Queries data from specified database tables"
async def execute(self, parameters):
try:
# 매개변수 검증
if "query" not in parameters:
raise ToolParameterError("Missing required parameter: query")
query = parameters["query"]
# 보안 검증
if self._contains_unsafe_sql(query):
raise ToolSecurityError("Query contains potentially unsafe SQL")
try:
# 타임아웃이 있는 데이터베이스 작업
async with timeout(10): # 10초 타임아웃
result = await self._database.execute_query(query)
return ToolResponse(
content=[TextContent(json.dumps(result))]
)
except asyncio.TimeoutError:
raise ToolExecutionError("Database query timed out after 10 seconds")
except DatabaseConnectionError as e:
# 연결 오류는 일시적일 수 있음
self._log_error("Database connection error", e)
raise ToolExecutionError(f"Database connection error: {str(e)}")
except DatabaseQueryError as e:
# 쿼리 오류는 클라이언트 오류일 가능성 높음
self._log_error("Database query error", e)
raise ToolExecutionError(f"Invalid query: {str(e)}")
except ToolError:
# 도구별 오류는 그대로 통과시킴
raise
except Exception as e:
# 예기치 않은 오류에 대한 포괄적 처리
self._log_error("Unexpected error in DataQueryTool", e)
raise ToolExecutionError(f"An unexpected error occurred: {str(e)}")
def _contains_unsafe_sql(self, query):
# SQL 인젝션 탐지 구현
pass
def _log_error(self, message, error):
# 오류 로그 기록 구현
pass
3. 매개변수 검증
오류가 있거나 악의적인 입력을 방지하기 위해 매개변수를 철저히 검증하십시오.
// 자세한 매개변수 검증이 포함된 JavaScript/TypeScript 예제
class FileOperationTool {
getName() {
return "fileOperation";
}
getDescription() {
return "Performs file operations like read, write, and delete";
}
getDefinition() {
return {
name: this.getName(),
description: this.getDescription(),
parameters: {
operation: {
type: "string",
description: "Operation to perform",
enum: ["read", "write", "delete"]
},
path: {
type: "string",
description: "File path (must be within allowed directories)"
},
content: {
type: "string",
description: "Content to write (only for write operation)",
optional: true
}
},
required: ["operation", "path"]
};
}
async execute(parameters) {
// 1. 매개변수 존재 여부 검증
if (!parameters.operation) {
throw new ToolError("Missing required parameter: operation");
}
if (!parameters.path) {
throw new ToolError("Missing required parameter: path");
}
// 2. 매개변수 타입 검증
if (typeof parameters.operation !== "string") {
throw new ToolError("Parameter 'operation' must be a string");
}
if (typeof parameters.path !== "string") {
throw new ToolError("Parameter 'path' must be a string");
}
// 3. 매개변수 값 검증
const validOperations = ["read", "write", "delete"];
if (!validOperations.includes(parameters.operation)) {
throw new ToolError(`Invalid operation. Must be one of: ${validOperations.join(", ")}`);
}
// 4. 쓰기 작업을 위한 내용 존재 여부 검증
if (parameters.operation === "write" && !parameters.content) {
throw new ToolError("Content parameter is required for write operation");
}
// 5. 경로 안전성 검증
if (!this.isPathWithinAllowedDirectories(parameters.path)) {
throw new ToolError("Access denied: path is outside of allowed directories");
}
// 검증된 매개변수를 기반으로 한 구현
// ...
}
isPathWithinAllowedDirectories(path) {
// 경로 안전성 검사 구현
// ...
}
}
보안 구현 예제
1. 인증 및 권한 부여
// 인증 및 권한 부여가 포함된 Java 예제
public class SecureDataAccessTool implements Tool {
private final AuthenticationService authService;
private final AuthorizationService authzService;
private final DataService dataService;
// 의존성 주입
public SecureDataAccessTool(
AuthenticationService authService,
AuthorizationService authzService,
DataService dataService) {
this.authService = authService;
this.authzService = authzService;
this.dataService = dataService;
}
@Override
public String getName() {
return "secureDataAccess";
}
@Override
public ToolResponse execute(ToolRequest request) {
// 1. 인증 컨텍스트 추출
String authToken = request.getContext().getAuthToken();
// 2. 사용자 인증
UserIdentity user;
try {
user = authService.validateToken(authToken);
} catch (AuthenticationException e) {
return ToolResponse.error("Authentication failed: " + e.getMessage());
}
// 3. 특정 작업에 대한 권한 확인
String dataId = request.getParameters().get("dataId").getAsString();
String operation = request.getParameters().get("operation").getAsString();
boolean isAuthorized = authzService.isAuthorized(user, "data:" + dataId, operation);
if (!isAuthorized) {
return ToolResponse.error("Access denied: Insufficient permissions for this operation");
}
// 4. 권한이 부여된 작업 진행
try {
switch (operation) {
case "read":
Object data = dataService.getData(dataId, user.getId());
return ToolResponse.success(data);
case "update":
JsonNode newData = request.getParameters().get("newData");
dataService.updateData(dataId, newData, user.getId());
return ToolResponse.success("Data updated successfully");
default:
return ToolResponse.error("Unsupported operation: " + operation);
}
} catch (Exception e) {
return ToolResponse.error("Operation failed: " + e.getMessage());
}
}
}
2. 속도 제한
// C# rate limiting implementation
public class RateLimitingMiddleware
{
private readonly RequestDelegate _next;
private readonly IMemoryCache _cache;
private readonly ILogger<RateLimitingMiddleware> _logger;
// Configuration options
private readonly int _maxRequestsPerMinute;
public RateLimitingMiddleware(
RequestDelegate next,
IMemoryCache cache,
ILogger<RateLimitingMiddleware> logger,
IConfiguration config)
{
_next = next;
_cache = cache;
_logger = logger;
_maxRequestsPerMinute = config.GetValue<int>("RateLimit:MaxRequestsPerMinute", 60);
}
public async Task InvokeAsync(HttpContext context)
{
// 1. Get client identifier (API key or user ID)
string clientId = GetClientIdentifier(context);
// 2. Get rate limiting key for this minute
string cacheKey = $"rate_limit:{clientId}:{DateTime.UtcNow:yyyyMMddHHmm}";
// 3. Check current request count
if (!_cache.TryGetValue(cacheKey, out int requestCount))
{
requestCount = 0;
}
// 4. Enforce rate limit
if (requestCount >= _maxRequestsPerMinute)
{
_logger.LogWarning("Rate limit exceeded for client {ClientId}", clientId);
context.Response.StatusCode = StatusCodes.Status429TooManyRequests;
context.Response.Headers.Add("Retry-After", "60");
await context.Response.WriteAsJsonAsync(new
{
error = "Rate limit exceeded",
message = "Too many requests. Please try again later.",
retryAfterSeconds = 60
});
return;
}
// 5. Increment request count
_cache.Set(cacheKey, requestCount + 1, TimeSpan.FromMinutes(2));
// 6. Add rate limit headers
context.Response.Headers.Add("X-RateLimit-Limit", _maxRequestsPerMinute.ToString());
context.Response.Headers.Add("X-RateLimit-Remaining", (_maxRequestsPerMinute - requestCount - 1).ToString());
// 7. Continue with the request
await _next(context);
}
private string GetClientIdentifier(HttpContext context)
{
// Implementation to extract API key or user ID
// ...
}
}
테스트 모범 사례
1. MCP 도구 단위 테스트
항상 도구를 독립적으로 테스트하며 외부 의존성은 모킹하십시오:
// TypeScript 도구 단위 테스트 예제
describe('WeatherForecastTool', () => {
let tool: WeatherForecastTool;
let mockWeatherService: jest.Mocked<IWeatherService>;
beforeEach(() => {
// 모의 날씨 서비스 생성
mockWeatherService = {
getForecasts: jest.fn()
} as any;
// 모의 종속성을 사용하여 도구 생성
tool = new WeatherForecastTool(mockWeatherService);
});
it('should return weather forecast for a location', async () => {
// 준비
const mockForecast = {
location: 'Seattle',
forecasts: [
{ date: '2025-07-16', temperature: 72, conditions: 'Sunny' },
{ date: '2025-07-17', temperature: 68, conditions: 'Partly Cloudy' },
{ date: '2025-07-18', temperature: 65, conditions: 'Rain' }
]
};
mockWeatherService.getForecasts.mockResolvedValue(mockForecast);
// 실행
const response = await tool.execute({
location: 'Seattle',
days: 3
});
// 검증
expect(mockWeatherService.getForecasts).toHaveBeenCalledWith('Seattle', 3);
expect(response.content[0].text).toContain('Seattle');
expect(response.content[0].text).toContain('Sunny');
});
it('should handle errors from the weather service', async () => {
// 준비
mockWeatherService.getForecasts.mockRejectedValue(new Error('Service unavailable'));
// 실행 및 검증
await expect(tool.execute({
location: 'Seattle',
days: 3
})).rejects.toThrow('Weather service error: Service unavailable');
});
});
2. 통합 테스트
클라이언트 요청부터 서버 응답까지 전체 흐름을 테스트하십시오:
# 파이썬 통합 테스트 예제
@pytest.mark.asyncio
async def test_mcp_server_integration():
# 테스트 서버 시작
server = McpServer()
server.register_tool(WeatherForecastTool(MockWeatherService()))
await server.start(port=5000)
try:
# 클라이언트 생성
client = McpClient("http://localhost:5000")
# 도구 탐색 테스트
tools = await client.discover_tools()
assert "weatherForecast" in [t.name for t in tools]
# 도구 실행 테스트
response = await client.execute_tool("weatherForecast", {
"location": "Seattle",
"days": 3
})
# 응답 확인
assert response.status_code == 200
assert "Seattle" in response.content[0].text
assert len(json.loads(response.content[0].text)["forecasts"]) == 3
finally:
# 정리 작업
await server.stop()
성능 최적화
1. 캐싱 전략
지연 시간과 자원 사용량을 줄이기 위한 적절한 캐싱을 구현하십시오:
// C# example with caching
public class CachedWeatherTool : ITool
{
private readonly IWeatherService _weatherService;
private readonly IDistributedCache _cache;
private readonly ILogger<CachedWeatherTool> _logger;
public CachedWeatherTool(
IWeatherService weatherService,
IDistributedCache cache,
ILogger<CachedWeatherTool> logger)
{
_weatherService = weatherService;
_cache = cache;
_logger = logger;
}
public string Name => "weatherForecast";
public async Task<ToolResponse> ExecuteAsync(IDictionary<string, object> parameters)
{
var location = parameters["location"].ToString();
var days = Convert.ToInt32(parameters.GetValueOrDefault("days", 3));
// Create cache key
string cacheKey = $"weather:{location}:{days}";
// Try to get from cache
string cachedForecast = await _cache.GetStringAsync(cacheKey);
if (!string.IsNullOrEmpty(cachedForecast))
{
_logger.LogInformation("Cache hit for weather forecast: {Location}", location);
return new ToolResponse
{
Content = new List<ContentItem>
{
new TextContent(cachedForecast)
}
};
}
// Cache miss - get from service
_logger.LogInformation("Cache miss for weather forecast: {Location}", location);
var forecast = await _weatherService.GetForecastAsync(location, days);
string forecastJson = JsonSerializer.Serialize(forecast);
// Store in cache (weather forecasts valid for 1 hour)
await _cache.SetStringAsync(
cacheKey,
forecastJson,
new DistributedCacheEntryOptions
{
AbsoluteExpirationRelativeToNow = TimeSpan.FromHours(1)
});
return new ToolResponse
{
Content = new List<ContentItem>
{
new TextContent(forecastJson)
}
};
}
}
2. 의존성 주입 및 테스트 가능성
생성자 주입을 통해 의존성을 받도록 설계하여 도구를 테스트 가능하고 구성 가능하게 만드십시오:
// 종속성 주입이 적용된 자바 예제
public class CurrencyConversionTool implements Tool {
private final ExchangeRateService exchangeService;
private final CacheService cacheService;
private final Logger logger;
// 생성자를 통해 주입된 종속성
public CurrencyConversionTool(
ExchangeRateService exchangeService,
CacheService cacheService,
Logger logger) {
this.exchangeService = exchangeService;
this.cacheService = cacheService;
this.logger = logger;
}
// 도구 구현
// ...
}
3. 조합 가능한 도구
보다 복잡한 워크플로를 만들 수 있도록 도구를 조합 가능하게 설계하십시오:
# 컴포저블 도구를 보여주는 파이썬 예제
class DataFetchTool(Tool):
def get_name(self):
return "dataFetch"
# 구현...
class DataAnalysisTool(Tool):
def get_name(self):
return "dataAnalysis"
# 이 도구는 dataFetch 도구의 결과를 사용할 수 있습니다
async def execute_async(self, request):
# 구현...
pass
class DataVisualizationTool(Tool):
def get_name(self):
return "dataVisualize"
# 이 도구는 dataAnalysis 도구의 결과를 사용할 수 있습니다
async def execute_async(self, request):
# 구현...
pass
# 이 도구들은 독립적으로 또는 워크플로우의 일부로 사용할 수 있습니다
스키마 설계 모범 사례
스키마는 모델과 도구 간 계약입니다. 잘 설계된 스키마는 도구 사용성을 향상시킵니다.
1. 명확한 매개변수 설명
각 매개변수에 설명 정보를 항상 포함하십시오:
public object GetSchema()
{
return new {
type = "object",
properties = new {
query = new {
type = "string",
description = "Search query text. Use precise keywords for better results."
},
filters = new {
type = "object",
description = "Optional filters to narrow down search results",
properties = new {
dateRange = new {
type = "string",
description = "Date range in format YYYY-MM-DD:YYYY-MM-DD"
},
category = new {
type = "string",
description = "Category name to filter by"
}
}
},
limit = new {
type = "integer",
description = "Maximum number of results to return (1-50)",
default = 10
}
},
required = new[] { "query" }
};
}
2. 검증 제약
잘못된 입력을 방지하기 위해 검증 제약을 포함하십시오:
Map<String, Object> getSchema() {
Map<String, Object> schema = new HashMap<>();
schema.put("type", "object");
Map<String, Object> properties = new HashMap<>();
// 형식 유효성 검사가 있는 이메일 속성
Map<String, Object> email = new HashMap<>();
email.put("type", "string");
email.put("format", "email");
email.put("description", "User email address");
// 숫자 제한이 있는 나이 속성
Map<String, Object> age = new HashMap<>();
age.put("type", "integer");
age.put("minimum", 13);
age.put("maximum", 120);
age.put("description", "User age in years");
// 열거형 속성
Map<String, Object> subscription = new HashMap<>();
subscription.put("type", "string");
subscription.put("enum", Arrays.asList("free", "basic", "premium"));
subscription.put("default", "free");
subscription.put("description", "Subscription tier");
properties.put("email", email);
properties.put("age", age);
properties.put("subscription", subscription);
schema.put("properties", properties);
schema.put("required", Arrays.asList("email"));
return schema;
}
3. 일관된 반환 구조
모델이 결과를 해석하기 쉽게 응답 구조에 일관성을 유지하십시오:
async def execute_async(self, request):
try:
# 요청 처리
results = await self._search_database(request.parameters["query"])
# 항상 일관된 구조를 반환
return ToolResponse(
result={
"matches": [self._format_item(item) for item in results],
"totalCount": len(results),
"queryTime": calculation_time_ms,
"status": "success"
}
)
except Exception as e:
return ToolResponse(
result={
"matches": [],
"totalCount": 0,
"queryTime": 0,
"status": "error",
"error": str(e)
}
)
def _format_item(self, item):
"""Ensures each item has a consistent structure"""
return {
"id": item.id,
"title": item.title,
"summary": item.summary[:100] + "..." if len(item.summary) > 100 else item.summary,
"url": item.url,
"relevance": item.score
}
오류 처리
MCP 도구의 신뢰성을 유지하려면 견고한 오류 처리가 필수적입니다.
1. 우아한 오류 처리
적절한 수준에서 오류를 처리하고 정보성 메시지를 제공하십시오:
public async Task<ToolResponse> ExecuteAsync(ToolRequest request)
{
try
{
string fileId = request.Parameters.GetProperty("fileId").GetString();
try
{
var fileData = await _fileService.GetFileAsync(fileId);
return new ToolResponse {
Result = JsonSerializer.SerializeToElement(fileData)
};
}
catch (FileNotFoundException)
{
throw new ToolExecutionException($"File not found: {fileId}");
}
catch (UnauthorizedAccessException)
{
throw new ToolExecutionException("You don't have permission to access this file");
}
catch (Exception ex) when (ex is IOException || ex is TimeoutException)
{
_logger.LogError(ex, "Error accessing file {FileId}", fileId);
throw new ToolExecutionException("Error accessing file: The service is temporarily unavailable");
}
}
catch (JsonException)
{
throw new ToolExecutionException("Invalid file ID format");
}
catch (Exception ex)
{
_logger.LogError(ex, "Unexpected error in FileAccessTool");
throw new ToolExecutionException("An unexpected error occurred");
}
}
2. 구조화된 오류 응답
가능한 경우 구조화된 오류 정보를 반환하십시오:
@Override
public ToolResponse execute(ToolRequest request) {
try {
// 구현
} catch (Exception ex) {
Map<String, Object> errorResult = new HashMap<>();
errorResult.put("success", false);
if (ex instanceof ValidationException) {
ValidationException validationEx = (ValidationException) ex;
errorResult.put("errorType", "validation");
errorResult.put("errorMessage", validationEx.getMessage());
errorResult.put("validationErrors", validationEx.getErrors());
return new ToolResponse.Builder()
.setResult(errorResult)
.build();
}
// 다른 예외를 ToolExecutionException으로 다시 던지기
throw new ToolExecutionException("Tool execution failed: " + ex.getMessage(), ex);
}
}
3. 재시도 로직
일시적 실패에 대해 적절한 재시도 로직을 구현하십시오:
async def execute_async(self, request):
max_retries = 3
retry_count = 0
base_delay = 1 # 초
while retry_count < max_retries:
try:
# 외부 API 호출
return await self._call_api(request.parameters)
except TransientError as e:
retry_count += 1
if retry_count >= max_retries:
raise ToolExecutionException(f"Operation failed after {max_retries} attempts: {str(e)}")
# 지수 백오프
delay = base_delay * (2 ** (retry_count - 1))
logging.warning(f"Transient error, retrying in {delay}s: {str(e)}")
await asyncio.sleep(delay)
except Exception as e:
# 비일시적 오류, 재시도하지 않음
raise ToolExecutionException(f"Operation failed: {str(e)}")
성능 최적화
1. 캐싱
고비용 작업에 대해 캐싱을 구현하십시오:
public class CachedDataTool : IMcpTool
{
private readonly IDatabase _database;
private readonly IMemoryCache _cache;
public CachedDataTool(IDatabase database, IMemoryCache cache)
{
_database = database;
_cache = cache;
}
public async Task<ToolResponse> ExecuteAsync(ToolRequest request)
{
var query = request.Parameters.GetProperty("query").GetString();
// Create cache key based on parameters
var cacheKey = $"data_query_{ComputeHash(query)}";
// Try to get from cache first
if (_cache.TryGetValue(cacheKey, out var cachedResult))
{
return new ToolResponse { Result = cachedResult };
}
// Cache miss - perform actual query
var result = await _database.QueryAsync(query);
// Store in cache with expiration
var cacheOptions = new MemoryCacheEntryOptions()
.SetAbsoluteExpiration(TimeSpan.FromMinutes(15));
_cache.Set(cacheKey, JsonSerializer.SerializeToElement(result), cacheOptions);
return new ToolResponse { Result = JsonSerializer.SerializeToElement(result) };
}
private string ComputeHash(string input)
{
// Implementation to generate stable hash for cache key
}
}
2. 비동기 처리
I/O 바운드 작업에 비동기 프로그래밍 패턴을 사용하십시오:
public class AsyncDocumentProcessingTool implements Tool {
private final DocumentService documentService;
private final ExecutorService executorService;
@Override
public ToolResponse execute(ToolRequest request) {
String documentId = request.getParameters().get("documentId").asText();
// 장시간 실행되는 작업의 경우 즉시 처리 ID를 반환합니다
String processId = UUID.randomUUID().toString();
// 비동기 처리 시작
CompletableFuture.runAsync(() -> {
try {
// 장시간 실행 작업 수행
documentService.processDocument(documentId);
// 상태 업데이트 (일반적으로 데이터베이스에 저장됨)
processStatusRepository.updateStatus(processId, "completed");
} catch (Exception ex) {
processStatusRepository.updateStatus(processId, "failed", ex.getMessage());
}
}, executorService);
// 처리 ID와 함께 즉시 응답 반환
Map<String, Object> result = new HashMap<>();
result.put("processId", processId);
result.put("status", "processing");
result.put("estimatedCompletionTime", ZonedDateTime.now().plusMinutes(5));
return new ToolResponse.Builder().setResult(result).build();
}
// 상태 확인 도구 동반
public class ProcessStatusTool implements Tool {
@Override
public ToolResponse execute(ToolRequest request) {
String processId = request.getParameters().get("processId").asText();
ProcessStatus status = processStatusRepository.getStatus(processId);
return new ToolResponse.Builder().setResult(status).build();
}
}
}
3. 자원 제한
과부하 방지를 위해 자원 제한을 구현하십시오:
class ThrottledApiTool(Tool):
def __init__(self):
self.rate_limiter = TokenBucketRateLimiter(
tokens_per_second=5, # 초당 5개의 요청 허용
bucket_size=10 # 최대 10개의 요청 폭발 허용
)
async def execute_async(self, request):
# 진행 가능한지 또는 대기해야 하는지 확인
delay = self.rate_limiter.get_delay_time()
if delay > 0:
if delay > 2.0: # 대기 시간이 너무 길면
raise ToolExecutionException(
f"Rate limit exceeded. Please try again in {delay:.1f} seconds."
)
else:
# 적절한 지연 시간 동안 대기
await asyncio.sleep(delay)
# 토큰을 소모하고 요청을 진행
self.rate_limiter.consume()
# API 호출
result = await self._call_api(request.parameters)
return ToolResponse(result=result)
class TokenBucketRateLimiter:
def __init__(self, tokens_per_second, bucket_size):
self.tokens_per_second = tokens_per_second
self.bucket_size = bucket_size
self.tokens = bucket_size
self.last_refill = time.time()
self.lock = asyncio.Lock()
async def get_delay_time(self):
async with self.lock:
self._refill()
if self.tokens >= 1:
return 0
# 다음 토큰이 사용 가능해질 때까지 시간 계산
return (1 - self.tokens) / self.tokens_per_second
async def consume(self):
async with self.lock:
self._refill()
self.tokens -= 1
def _refill(self):
now = time.time()
elapsed = now - self.last_refill
# 경과 시간에 따라 새 토큰 추가
new_tokens = elapsed * self.tokens_per_second
self.tokens = min(self.bucket_size, self.tokens + new_tokens)
self.last_refill = now
보안 모범 사례
1. 입력 검증
입력 매개변수를 항상 철저히 검증하십시오:
public async Task<ToolResponse> ExecuteAsync(ToolRequest request)
{
// Validate parameters exist
if (!request.Parameters.TryGetProperty("query", out var queryProp))
{
throw new ToolExecutionException("Missing required parameter: query");
}
// Validate correct type
if (queryProp.ValueKind != JsonValueKind.String)
{
throw new ToolExecutionException("Query parameter must be a string");
}
var query = queryProp.GetString();
// Validate string content
if (string.IsNullOrWhiteSpace(query))
{
throw new ToolExecutionException("Query parameter cannot be empty");
}
if (query.Length > 500)
{
throw new ToolExecutionException("Query parameter exceeds maximum length of 500 characters");
}
// Check for SQL injection attacks if applicable
if (ContainsSqlInjection(query))
{
throw new ToolExecutionException("Invalid query: contains potentially unsafe SQL");
}
// Proceed with execution
// ...
}
2. 권한 확인
적절한 권한 검사를 구현하십시오:
@Override
public ToolResponse execute(ToolRequest request) {
// 요청에서 사용자 컨텍스트를 가져옵니다
UserContext user = request.getContext().getUserContext();
// 사용자가 필요한 권한을 가지고 있는지 확인합니다
if (!authorizationService.hasPermission(user, "documents:read")) {
throw new ToolExecutionException("User does not have permission to access documents");
}
// 특정 리소스의 경우 해당 리소스에 대한 접근을 확인합니다
String documentId = request.getParameters().get("documentId").asText();
if (!documentService.canUserAccess(user.getId(), documentId)) {
throw new ToolExecutionException("Access denied to the requested document");
}
// 도구 실행을 진행합니다
// ...
}
3. 민감 데이터 처리
민감한 데이터를 신중히 처리하십시오:
class SecureDataTool(Tool):
def get_schema(self):
return {
"type": "object",
"properties": {
"userId": {"type": "string"},
"includeSensitiveData": {"type": "boolean", "default": False}
},
"required": ["userId"]
}
async def execute_async(self, request):
user_id = request.parameters["userId"]
include_sensitive = request.parameters.get("includeSensitiveData", False)
# 사용자 데이터 가져오기
user_data = await self.user_service.get_user_data(user_id)
# 명시적으로 요청되고 권한이 부여된 경우가 아니면 민감한 필드 필터링
if not include_sensitive or not self._is_authorized_for_sensitive_data(request):
user_data = self._redact_sensitive_fields(user_data)
return ToolResponse(result=user_data)
def _is_authorized_for_sensitive_data(self, request):
# 요청 컨텍스트에서 권한 수준 확인
auth_level = request.context.get("authorizationLevel")
return auth_level == "admin"
def _redact_sensitive_fields(self, user_data):
# 원본을 수정하지 않도록 복사본 생성
redacted = user_data.copy()
# 특정 민감한 필드 삭제
sensitive_fields = ["ssn", "creditCardNumber", "password"]
for field in sensitive_fields:
if field in redacted:
redacted[field] = "REDACTED"
# 중첩된 민감한 데이터 삭제
if "financialInfo" in redacted:
redacted["financialInfo"] = {"available": True, "accessRestricted": True}
return redacted
MCP 도구 테스트 모범 사례
포괄적인 테스트는 MCP 도구가 올바르게 작동하고, 엣지 케이스를 처리하며, 시스템과 적절히 통합되도록 보장합니다.
단위 테스트
1. 각 도구를 독립적으로 테스트
각 도구 기능에 집중한 테스트를 만드십시오:
[Fact]
public async Task WeatherTool_ValidLocation_ReturnsCorrectForecast()
{
// Arrange
var mockWeatherService = new Mock<IWeatherService>();
mockWeatherService
.Setup(s => s.GetForecastAsync("Seattle", 3))
.ReturnsAsync(new WeatherForecast(/* test data */));
var tool = new WeatherForecastTool(mockWeatherService.Object);
var request = new ToolRequest(
toolName: "weatherForecast",
parameters: JsonSerializer.SerializeToElement(new {
location = "Seattle",
days = 3
})
);
// Act
var response = await tool.ExecuteAsync(request);
// Assert
Assert.NotNull(response);
var result = JsonSerializer.Deserialize<WeatherForecast>(response.Result);
Assert.Equal("Seattle", result.Location);
Assert.Equal(3, result.DailyForecasts.Count);
}
[Fact]
public async Task WeatherTool_InvalidLocation_ThrowsToolExecutionException()
{
// Arrange
var mockWeatherService = new Mock<IWeatherService>();
mockWeatherService
.Setup(s => s.GetForecastAsync("InvalidLocation", It.IsAny<int>()))
.ThrowsAsync(new LocationNotFoundException("Location not found"));
var tool = new WeatherForecastTool(mockWeatherService.Object);
var request = new ToolRequest(
toolName: "weatherForecast",
parameters: JsonSerializer.SerializeToElement(new {
location = "InvalidLocation",
days = 3
})
);
// Act & Assert
var exception = await Assert.ThrowsAsync<ToolExecutionException>(
() => tool.ExecuteAsync(request)
);
Assert.Contains("Location not found", exception.Message);
}
2. 스키마 검증 테스트
스키마가 유효하며 제약을 적절히 강제하는지 테스트하십시오:
@Test
public void testSchemaValidation() {
// 도구 인스턴스 생성
SearchTool searchTool = new SearchTool();
// 스키마 가져오기
Object schema = searchTool.getSchema();
// 검증을 위해 스키마를 JSON으로 변환
String schemaJson = objectMapper.writeValueAsString(schema);
// 스키마가 유효한 JSONSchema인지 검증
JsonSchemaFactory factory = JsonSchemaFactory.byDefault();
JsonSchema jsonSchema = factory.getJsonSchema(schemaJson);
// 유효한 매개변수 테스트
JsonNode validParams = objectMapper.createObjectNode()
.put("query", "test query")
.put("limit", 5);
ProcessingReport validReport = jsonSchema.validate(validParams);
assertTrue(validReport.isSuccess());
// 필수 매개변수 누락 테스트
JsonNode missingRequired = objectMapper.createObjectNode()
.put("limit", 5);
ProcessingReport missingReport = jsonSchema.validate(missingRequired);
assertFalse(missingReport.isSuccess());
// 잘못된 매개변수 유형 테스트
JsonNode invalidType = objectMapper.createObjectNode()
.put("query", "test")
.put("limit", "not-a-number");
ProcessingReport invalidReport = jsonSchema.validate(invalidType);
assertFalse(invalidReport.isSuccess());
}
3. 오류 처리 테스트
오류 조건에 대한 특정 테스트를 생성하십시오:
@pytest.mark.asyncio
async def test_api_tool_handles_timeout():
# 정렬
tool = ApiTool(timeout=0.1) # 매우 짧은 타임아웃
# 타임아웃될 요청을 모의(mock) 처리
with aioresponses() as mocked:
mocked.get(
"https://api.example.com/data",
callback=lambda *args, **kwargs: asyncio.sleep(0.5) # 타임아웃보다 긴 시간
)
request = ToolRequest(
tool_name="apiTool",
parameters={"url": "https://api.example.com/data"}
)
# 실행 및 검증
with pytest.raises(ToolExecutionException) as exc_info:
await tool.execute_async(request)
# 예외 메시지 확인
assert "timed out" in str(exc_info.value).lower()
@pytest.mark.asyncio
async def test_api_tool_handles_rate_limiting():
# 정렬
tool = ApiTool()
# 요청 제한된 응답을 모의(mock) 처리
with aioresponses() as mocked:
mocked.get(
"https://api.example.com/data",
status=429,
headers={"Retry-After": "2"},
body=json.dumps({"error": "Rate limit exceeded"})
)
request = ToolRequest(
tool_name="apiTool",
parameters={"url": "https://api.example.com/data"}
)
# 실행 및 검증
with pytest.raises(ToolExecutionException) as exc_info:
await tool.execute_async(request)
# 예외에 요청 제한 정보가 포함되었는지 확인
error_msg = str(exc_info.value).lower()
assert "rate limit" in error_msg
assert "try again" in error_msg
통합 테스트
1. 도구 체인 테스트
예상 조합에서 도구가 함께 작동하는지 테스트하십시오:
[Fact]
public async Task DataProcessingWorkflow_CompletesSuccessfully()
{
// Arrange
var dataFetchTool = new DataFetchTool(mockDataService.Object);
var analysisTools = new DataAnalysisTool(mockAnalysisService.Object);
var visualizationTool = new DataVisualizationTool(mockVisualizationService.Object);
var toolRegistry = new ToolRegistry();
toolRegistry.RegisterTool(dataFetchTool);
toolRegistry.RegisterTool(analysisTools);
toolRegistry.RegisterTool(visualizationTool);
var workflowExecutor = new WorkflowExecutor(toolRegistry);
// Act
var result = await workflowExecutor.ExecuteWorkflowAsync(new[] {
new ToolCall("dataFetch", new { source = "sales2023" }),
new ToolCall("dataAnalysis", ctx => new {
data = ctx.GetResult("dataFetch"),
analysis = "trend"
}),
new ToolCall("dataVisualize", ctx => new {
analysisResult = ctx.GetResult("dataAnalysis"),
type = "line-chart"
})
});
// Assert
Assert.NotNull(result);
Assert.True(result.Success);
Assert.NotNull(result.GetResult("dataVisualize"));
Assert.Contains("chartUrl", result.GetResult("dataVisualize").ToString());
}
2. MCP 서버 테스트
도구 등록 및 실행이 포함된 MCP 서버를 테스트하십시오:
@SpringBootTest
@AutoConfigureMockMvc
public class McpServerIntegrationTest {
@Autowired
private MockMvc mockMvc;
@Autowired
private ObjectMapper objectMapper;
@Test
public void testToolDiscovery() throws Exception {
// 디스커버리 엔드포인트 테스트
mockMvc.perform(get("/mcp/tools"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.tools").isArray())
.andExpect(jsonPath("$.tools[*].name").value(hasItems(
"weatherForecast", "calculator", "documentSearch"
)));
}
@Test
public void testToolExecution() throws Exception {
// 도구 요청 생성
Map<String, Object> request = new HashMap<>();
request.put("toolName", "calculator");
Map<String, Object> parameters = new HashMap<>();
parameters.put("operation", "add");
parameters.put("a", 5);
parameters.put("b", 7);
request.put("parameters", parameters);
// 요청을 보내고 응답 검증
mockMvc.perform(post("/mcp/execute")
.contentType(MediaType.APPLICATION_JSON)
.content(objectMapper.writeValueAsString(request)))
.andExpect(status().isOk())
.andExpect(jsonPath("$.result.value").value(12));
}
@Test
public void testToolValidation() throws Exception {
// 잘못된 도구 요청 생성
Map<String, Object> request = new HashMap<>();
request.put("toolName", "calculator");
Map<String, Object> parameters = new HashMap<>();
parameters.put("operation", "divide");
parameters.put("a", 10);
// 파라미터 "b" 누락
request.put("parameters", parameters);
// 요청을 보내고 오류 응답 검증
mockMvc.perform(post("/mcp/execute")
.contentType(MediaType.APPLICATION_JSON)
.content(objectMapper.writeValueAsString(request)))
.andExpect(status().isBadRequest())
.andExpect(jsonPath("$.error").exists());
}
}
3. 종단 간 테스트
모델 프롬프트부터 도구 실행까지 전체 워크플로를 테스트하십시오:
@pytest.mark.asyncio
async def test_model_interaction_with_tool():
# 정렬 - MCP 클라이언트 및 모의 모델 설정
mcp_client = McpClient(server_url="http://localhost:5000")
# 모의 모델 응답
mock_model = MockLanguageModel([
MockResponse(
"What's the weather in Seattle?",
tool_calls=[{
"tool_name": "weatherForecast",
"parameters": {"location": "Seattle", "days": 3}
}]
),
MockResponse(
"Here's the weather forecast for Seattle:\n- Today: 65°F, Partly Cloudy\n- Tomorrow: 68°F, Sunny\n- Day after: 62°F, Rain",
tool_calls=[]
)
])
# 모의 날씨 도구 응답
with aioresponses() as mocked:
mocked.post(
"http://localhost:5000/mcp/execute",
payload={
"result": {
"location": "Seattle",
"forecast": [
{"date": "2023-06-01", "temperature": 65, "conditions": "Partly Cloudy"},
{"date": "2023-06-02", "temperature": 68, "conditions": "Sunny"},
{"date": "2023-06-03", "temperature": 62, "conditions": "Rain"}
]
}
}
)
# 실행
response = await mcp_client.send_prompt(
"What's the weather in Seattle?",
model=mock_model,
allowed_tools=["weatherForecast"]
)
# 확인
assert "Seattle" in response.generated_text
assert "65" in response.generated_text
assert "Sunny" in response.generated_text
assert "Rain" in response.generated_text
assert len(response.tool_calls) == 1
assert response.tool_calls[0].tool_name == "weatherForecast"
성능 테스트
1. 부하 테스트
MCP 서버가 처리할 수 있는 동시 요청 수를 테스트하십시오:
[Fact]
public async Task McpServer_HandlesHighConcurrency()
{
// Arrange
var server = new McpServer(
name: "TestServer",
version: "1.0",
maxConcurrentRequests: 100
);
server.RegisterTool(new FastExecutingTool());
await server.StartAsync();
var client = new McpClient("http://localhost:5000");
// Act
var tasks = new List<Task<McpResponse>>();
for (int i = 0; i < 1000; i++)
{
tasks.Add(client.ExecuteToolAsync("fastTool", new { iteration = i }));
}
var results = await Task.WhenAll(tasks);
// Assert
Assert.Equal(1000, results.Length);
Assert.All(results, r => Assert.NotNull(r));
}
2. 스트레스 테스트
극한 부하 상태에서 시스템을 테스트하십시오:
@Test
public void testServerUnderStress() {
int maxUsers = 1000;
int rampUpTimeSeconds = 60;
int testDurationSeconds = 300;
// 스트레스 테스트를 위해 JMeter 설정
StandardJMeterEngine jmeter = new StandardJMeterEngine();
// JMeter 테스트 플랜 구성
HashTree testPlanTree = new HashTree();
// 테스트 플랜, 스레드 그룹, 샘플러 등 생성
TestPlan testPlan = new TestPlan("MCP Server Stress Test");
testPlanTree.add(testPlan);
ThreadGroup threadGroup = new ThreadGroup();
threadGroup.setNumThreads(maxUsers);
threadGroup.setRampUp(rampUpTimeSeconds);
threadGroup.setScheduler(true);
threadGroup.setDuration(testDurationSeconds);
testPlanTree.add(threadGroup);
// 도구 실행을 위한 HTTP 샘플러 추가
HTTPSampler toolExecutionSampler = new HTTPSampler();
toolExecutionSampler.setDomain("localhost");
toolExecutionSampler.setPort(5000);
toolExecutionSampler.setPath("/mcp/execute");
toolExecutionSampler.setMethod("POST");
toolExecutionSampler.addArgument("toolName", "calculator");
toolExecutionSampler.addArgument("parameters", "{\"operation\":\"add\",\"a\":5,\"b\":7}");
threadGroup.add(toolExecutionSampler);
// 리스너 추가
SummaryReport summaryReport = new SummaryReport();
threadGroup.add(summaryReport);
// 테스트 실행
jmeter.configure(testPlanTree);
jmeter.run();
// 결과 검증
assertEquals(0, summaryReport.getErrorCount());
assertTrue(summaryReport.getAverage() < 200); // 평균 응답 시간 < 200ms
assertTrue(summaryReport.getPercentile(90.0) < 500); // 90번째 백분위수 < 500ms
}
3. 모니터링 및 프로파일링
장기 성능 분석을 위해 모니터링을 설정하십시오:
# MCP 서버 모니터링 구성
def configure_monitoring(server):
# Prometheus 메트릭 설정
prometheus_metrics = {
"request_count": Counter("mcp_requests_total", "Total MCP requests"),
"request_latency": Histogram(
"mcp_request_duration_seconds",
"Request duration in seconds",
buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.5, 5.0, 10.0]
),
"tool_execution_count": Counter(
"mcp_tool_executions_total",
"Tool execution count",
labelnames=["tool_name"]
),
"tool_execution_latency": Histogram(
"mcp_tool_duration_seconds",
"Tool execution duration in seconds",
labelnames=["tool_name"],
buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.5, 5.0, 10.0]
),
"tool_errors": Counter(
"mcp_tool_errors_total",
"Tool execution errors",
labelnames=["tool_name", "error_type"]
)
}
# 타이밍 및 메트릭 기록을 위한 미들웨어 추가
server.add_middleware(PrometheusMiddleware(prometheus_metrics))
# 메트릭 엔드포인트 노출
@server.router.get("/metrics")
async def metrics():
return generate_latest()
return server
MCP 워크플로 설계 패턴
잘 설계된 MCP 워크플로는 효율성, 신뢰성 및 유지 관리를 개선합니다. 다음은 따라야 할 주요 패턴입니다:
1. 도구 체인 패턴
여러 도구를 순차적으로 연결하여 각 도구의 출력이 다음 도구의 입력이 되도록 하십시오:
# 파이썬 체인 오브 툴 구현
class ChainWorkflow:
def __init__(self, tools_chain):
self.tools_chain = tools_chain # 순서대로 실행할 도구 이름 목록
async def execute(self, mcp_client, initial_input):
current_result = initial_input
all_results = {"input": initial_input}
for tool_name in self.tools_chain:
# 이전 결과를 전달하며 체인의 각 도구 실행
response = await mcp_client.execute_tool(tool_name, current_result)
# 결과를 저장하고 다음 도구의 입력으로 사용
all_results[tool_name] = response.result
current_result = response.result
return {
"final_result": current_result,
"all_results": all_results
}
# 사용 예제
data_processing_chain = ChainWorkflow([
"dataFetch",
"dataCleaner",
"dataAnalyzer",
"dataVisualizer"
])
result = await data_processing_chain.execute(
mcp_client,
{"source": "sales_database", "table": "transactions"}
)
2. 디스패처 패턴
입력에 따라 특화된 도구로 전달하는 중앙 도구를 사용하십시오:
public class ContentDispatcherTool : IMcpTool
{
private readonly IMcpClient _mcpClient;
public ContentDispatcherTool(IMcpClient mcpClient)
{
_mcpClient = mcpClient;
}
public string Name => "contentProcessor";
public string Description => "Processes content of various types";
public object GetSchema()
{
return new {
type = "object",
properties = new {
content = new { type = "string" },
contentType = new {
type = "string",
enum = new[] { "text", "html", "markdown", "csv", "code" }
},
operation = new {
type = "string",
enum = new[] { "summarize", "analyze", "extract", "convert" }
}
},
required = new[] { "content", "contentType", "operation" }
};
}
public async Task<ToolResponse> ExecuteAsync(ToolRequest request)
{
var content = request.Parameters.GetProperty("content").GetString();
var contentType = request.Parameters.GetProperty("contentType").GetString();
var operation = request.Parameters.GetProperty("operation").GetString();
// Determine which specialized tool to use
string targetTool = DetermineTargetTool(contentType, operation);
// Forward to the specialized tool
var specializedResponse = await _mcpClient.ExecuteToolAsync(
targetTool,
new { content, options = GetOptionsForTool(targetTool, operation) }
);
return new ToolResponse { Result = specializedResponse.Result };
}
private string DetermineTargetTool(string contentType, string operation)
{
return (contentType, operation) switch
{
("text", "summarize") => "textSummarizer",
("text", "analyze") => "textAnalyzer",
("html", _) => "htmlProcessor",
("markdown", _) => "markdownProcessor",
("csv", _) => "csvProcessor",
("code", _) => "codeAnalyzer",
_ => throw new ToolExecutionException($"No tool available for {contentType}/{operation}")
};
}
private object GetOptionsForTool(string toolName, string operation)
{
// Return appropriate options for each specialized tool
return toolName switch
{
"textSummarizer" => new { length = "medium" },
"htmlProcessor" => new { cleanUp = true, operation },
// Options for other tools...
_ => new { }
};
}
}
3. 병렬 처리 패턴
효율성을 위해 여러 도구를 동시에 실행하십시오:
public class ParallelDataProcessingWorkflow {
private final McpClient mcpClient;
public ParallelDataProcessingWorkflow(McpClient mcpClient) {
this.mcpClient = mcpClient;
}
public WorkflowResult execute(String datasetId) {
// 1단계: 데이터셋 메타데이터 가져오기(동기식)
ToolResponse metadataResponse = mcpClient.executeTool("datasetMetadata",
Map.of("datasetId", datasetId));
// 2단계: 여러 분석을 병렬로 실행
CompletableFuture<ToolResponse> statisticalAnalysis = CompletableFuture.supplyAsync(() ->
mcpClient.executeTool("statisticalAnalysis", Map.of(
"datasetId", datasetId,
"type", "comprehensive"
))
);
CompletableFuture<ToolResponse> correlationAnalysis = CompletableFuture.supplyAsync(() ->
mcpClient.executeTool("correlationAnalysis", Map.of(
"datasetId", datasetId,
"method", "pearson"
))
);
CompletableFuture<ToolResponse> outlierDetection = CompletableFuture.supplyAsync(() ->
mcpClient.executeTool("outlierDetection", Map.of(
"datasetId", datasetId,
"sensitivity", "medium"
))
);
// 모든 병렬 작업이 완료될 때까지 대기
CompletableFuture<Void> allAnalyses = CompletableFuture.allOf(
statisticalAnalysis, correlationAnalysis, outlierDetection
);
allAnalyses.join(); // 완료될 때까지 대기
// 3단계: 결과 결합
Map<String, Object> combinedResults = new HashMap<>();
combinedResults.put("metadata", metadataResponse.getResult());
combinedResults.put("statistics", statisticalAnalysis.join().getResult());
combinedResults.put("correlations", correlationAnalysis.join().getResult());
combinedResults.put("outliers", outlierDetection.join().getResult());
// 4단계: 요약 보고서 생성
ToolResponse summaryResponse = mcpClient.executeTool("reportGenerator",
Map.of("analysisResults", combinedResults));
// 전체 워크플로우 결과 반환
WorkflowResult result = new WorkflowResult();
result.setDatasetId(datasetId);
result.setAnalysisResults(combinedResults);
result.setSummaryReport(summaryResponse.getResult());
return result;
}
}
4. 오류 복구 패턴
도구 실패에 대비한 우아한 대체 경로를 구현하십시오:
class ResilientWorkflow:
def __init__(self, mcp_client):
self.client = mcp_client
async def execute_with_fallback(self, primary_tool, fallback_tool, parameters):
try:
# 먼저 기본 도구를 시도합니다
response = await self.client.execute_tool(primary_tool, parameters)
return {
"result": response.result,
"source": "primary",
"tool": primary_tool
}
except ToolExecutionException as e:
# 실패를 기록합니다
logging.warning(f"Primary tool '{primary_tool}' failed: {str(e)}")
# 보조 도구로 대체합니다
try:
# 보조 도구를 위해 매개변수를 변환해야 할 수도 있습니다
fallback_params = self._adapt_parameters(parameters, primary_tool, fallback_tool)
response = await self.client.execute_tool(fallback_tool, fallback_params)
return {
"result": response.result,
"source": "fallback",
"tool": fallback_tool,
"primaryError": str(e)
}
except ToolExecutionException as fallback_error:
# 두 도구 모두 실패했습니다
logging.error(f"Both primary and fallback tools failed. Fallback error: {str(fallback_error)}")
raise WorkflowExecutionException(
f"Workflow failed: primary error: {str(e)}; fallback error: {str(fallback_error)}"
)
def _adapt_parameters(self, params, from_tool, to_tool):
"""Adapt parameters between different tools if needed"""
# 이 구현은 특정 도구에 따라 달라질 것입니다
# 이 예에서는 원래 매개변수를 그대로 반환합니다
return params
# 사용 예
async def get_weather(workflow, location):
return await workflow.execute_with_fallback(
"premiumWeatherService", # 기본 (유료) 날씨 API
"basicWeatherService", # 보조 (무료) 날씨 API
{"location": location}
)
5. 워크플로 조합 패턴
더 단순한 워크플로를 조합하여 복잡한 작업을 만드십시오:
public class CompositeWorkflow : IWorkflow
{
private readonly List<IWorkflow> _workflows;
public CompositeWorkflow(IEnumerable<IWorkflow> workflows)
{
_workflows = new List<IWorkflow>(workflows);
}
public async Task<WorkflowResult> ExecuteAsync(WorkflowContext context)
{
var results = new Dictionary<string, object>();
foreach (var workflow in _workflows)
{
var workflowResult = await workflow.ExecuteAsync(context);
// Store each workflow's result
results[workflow.Name] = workflowResult;
// Update context with the result for the next workflow
context = context.WithResult(workflow.Name, workflowResult);
}
return new WorkflowResult(results);
}
public string Name => "CompositeWorkflow";
public string Description => "Executes multiple workflows in sequence";
}
// Example usage
var documentWorkflow = new CompositeWorkflow(new IWorkflow[] {
new DocumentFetchWorkflow(),
new DocumentProcessingWorkflow(),
new InsightGenerationWorkflow(),
new ReportGenerationWorkflow()
});
var result = await documentWorkflow.ExecuteAsync(new WorkflowContext {
Parameters = new { documentId = "12345" }
});
MCP 서버 테스트: 모범 사례 및 주요 팁
개요
테스트는 신뢰할 수 있고 고품질 MCP 서버를 개발하는 데 중요한 요소입니다. 이 가이드는 단위 테스트부터 통합 테스트, 종단 간 검증까지 개발 라이프사이클 전반에 걸쳐 MCP 서버 테스트를 위한 포괄적인 모범 사례와 팁을 제공합니다.
MCP 서버 테스트가 중요한 이유
MCP 서버는 AI 모델과 클라이언트 애플리케이션 간 중요한 미들웨어 역할을 합니다. 철저한 테스트는 다음을 보장합니다:
- 프로덕션 환경에서의 신뢰성
- 요청과 응답의 정확한 처리
- MCP 명세의 올바른 구현
- 실패 및 예외 상황에 대한 탄력성
- 다양한 부하 하에서의 일관된 성능
MCP 서버 단위 테스트
단위 테스트 (기초)
단위 테스트는 MCP 서버 구성 요소를 독립적으로 검증합니다.
테스트 대상
- 리소스 핸들러: 각 리소스 핸들러 로직을 개별적으로 테스트
- 도구 구현: 다양한 입력에 따른 도구 동작 검증
- 프롬프트 템플릿: 프롬프트 템플릿이 올바르게 렌더링되는지 확인
- 스키마 검증: 매개변수 검증 로직 테스트
- 오류 처리: 잘못된 입력에 대해 오류 응답 검증
단위 테스트 모범 사례
// Example unit test for a calculator tool in C#
[Fact]
public async Task CalculatorTool_Add_ReturnsCorrectSum()
{
// Arrange
var calculator = new CalculatorTool();
var parameters = new Dictionary<string, object>
{
["operation"] = "add",
["a"] = 5,
["b"] = 7
};
// Act
var response = await calculator.ExecuteAsync(parameters);
var result = JsonSerializer.Deserialize<CalculationResult>(response.Content[0].ToString());
// Assert
Assert.Equal(12, result.Value);
}
# 파이썬에서 계산기 도구에 대한 예제 단위 테스트
def test_calculator_tool_add():
# 준비
calculator = CalculatorTool()
parameters = {
"operation": "add",
"a": 5,
"b": 7
}
# 실행
response = calculator.execute(parameters)
result = json.loads(response.content[0].text)
# 검증
assert result["value"] == 12
통합 테스트 (중간 계층)
통합 테스트는 MCP 서버 구성요소 간 상호작용을 검증합니다.
테스트 대상
- 서버 초기화: 다양한 설정으로 서버 시작 테스트
- 경로 등록: 모든 엔드포인트가 올바르게 등록되었는지 검증
- 요청 처리: 전체 요청-응답 주기 테스트
- 오류 전파: 구성 요소 간 오류가 적절히 처리되는지 확인
- 인증 및 권한 부여: 보안 메커니즘 테스트
통합 테스트 모범 사례
// Example integration test for MCP server in C#
[Fact]
public async Task Server_ProcessToolRequest_ReturnsValidResponse()
{
// Arrange
var server = new McpServer();
server.RegisterTool(new CalculatorTool());
await server.StartAsync();
var request = new McpRequest
{
Tool = "calculator",
Parameters = new Dictionary<string, object>
{
["operation"] = "multiply",
["a"] = 6,
["b"] = 7
}
};
// Act
var response = await server.ProcessRequestAsync(request);
// Assert
Assert.NotNull(response);
Assert.Equal(McpStatusCodes.Success, response.StatusCode);
// Additional assertions for response content
// Cleanup
await server.StopAsync();
}
종단 간 테스트 (최상위 계층)
종단 간 테스트는 클라이언트부터 서버까지 전체 시스템 동작을 검증합니다.
테스트 대상
- 클라이언트-서버 통신: 전체 요청-응답 주기 테스트
- 실제 클라이언트 SDK: 실제 클라이언트 구현으로 테스트
- 부하 하 성능: 여러 동시 요청 시 동작 검증
- 오류 복구: 실패에서 시스템 복구 테스트
- 장기 실행 작업: 스트리밍 및 장기 작업 처리 검증
종단 간 테스트 모범 사례
// TypeScript로 작성된 클라이언트와 함께하는 예제 E2E 테스트
describe('MCP Server E2E Tests', () => {
let client: McpClient;
beforeAll(async () => {
// 테스트 환경에서 서버 시작
await startTestServer();
client = new McpClient('http://localhost:5000');
});
afterAll(async () => {
await stopTestServer();
});
test('Client can invoke calculator tool and get correct result', async () => {
// 동작
const response = await client.invokeToolAsync('calculator', {
operation: 'divide',
a: 20,
b: 4
});
// 검증
expect(response.statusCode).toBe(200);
expect(response.content[0].text).toContain('5');
});
});
MCP 테스트를 위한 모킹 전략
모킹은 테스트 중 구성 요소를 분리하는 데 필수적입니다.
모킹 대상 구성 요소
- 외부 AI 모델: 예측 가능한 테스트를 위해 모델 응답 모킹
- 외부 서비스: API 의존성(데이터베이스, 타사 서비스) 모킹
- 인증 서비스: 인증 공급자 모킹
- 리소스 공급자: 비용이 많이 드는 리소스 핸들러 모킹
예시: AI 모델 응답 모킹
// C# example with Moq
var mockModel = new Mock<ILanguageModel>();
mockModel
.Setup(m => m.GenerateResponseAsync(
It.IsAny<string>(),
It.IsAny<McpRequestContext>()))
.ReturnsAsync(new ModelResponse {
Text = "Mocked model response",
FinishReason = FinishReason.Completed
});
var server = new McpServer(modelClient: mockModel.Object);
# unittest.mock를 사용한 파이썬 예제
@patch('mcp_server.models.OpenAIModel')
def test_with_mock_model(mock_model):
# 목 구성하기
mock_model.return_value.generate_response.return_value = {
"text": "Mocked model response",
"finish_reason": "completed"
}
# 테스트에서 목 사용하기
server = McpServer(model_client=mock_model)
# 테스트 계속하기
성능 테스트
성능 테스트는 프로덕션 MCP 서버에 필수적입니다.
측정 대상
- 지연 시간: 요청 응답 시간
- 처리량: 초당 처리 요청 수
- 자원 사용량: CPU, 메모리, 네트워크 사용량
- 동시성 처리: 병렬 요청 시 동작
- 확장 특성: 부하 증가에 따른 성능
성능 테스트 도구
- k6: 오픈 소스 부하 테스트 도구
- JMeter: 종합 성능 테스트
- Locust: 파이썬 기반 부하 테스트
- Azure Load Testing: 클라우드 기반 성능 테스트
예시: k6로 기본 부하 테스트
// MCP 서버 부하 테스트용 k6 스크립트
import http from 'k6/http';
import { check, sleep } from 'k6';
export const options = {
vus: 10, // 10 가상 사용자
duration: '30s',
};
export default function () {
const payload = JSON.stringify({
tool: 'calculator',
parameters: {
operation: 'add',
a: Math.floor(Math.random() * 100),
b: Math.floor(Math.random() * 100)
}
});
const params = {
headers: {
'Content-Type': 'application/json',
'Authorization': 'Bearer test-token'
},
};
const res = http.post('http://localhost:5000/api/tools/invoke', payload, params);
check(res, {
'status is 200': (r) => r.status === 200,
'response time < 500ms': (r) => r.timings.duration < 500,
});
sleep(1);
}
MCP 서버 테스트 자동화
테스트 자동화는 일관된 품질과 빠른 피드백 루프를 보장합니다.
CI/CD 통합
- 풀 리퀘스트에서 단위 테스트 실행: 코드 변경이 기존 기능을 손상시키지 않도록 보장
- 스테이징에서의 통합 테스트: 사전 프로덕션 환경에서 통합 테스트 실행
- 성능 기준 유지: 회귀를 잡기 위해 성능 벤치마크 유지
- 보안 검사: 파이프라인의 일부로 보안 테스트 자동화
예시 CI 파이프라인 (GitHub Actions)
name: MCP Server Tests
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
- name: Set up Runtime
uses: actions/setup-dotnet@v1
with:
dotnet-version: '8.0.x'
- name: Restore dependencies
run: dotnet restore
- name: Build
run: dotnet build --no-restore
- name: Unit Tests
run: dotnet test --no-build --filter Category=Unit
- name: Integration Tests
run: dotnet test --no-build --filter Category=Integration
- name: Performance Tests
run: dotnet run --project tests/PerformanceTests/PerformanceTests.csproj
MCP 사양 준수 테스트
서버가 MCP 사양을 올바르게 구현했는지 확인하세요.
주요 준수 영역
- API 엔드포인트: 필수 엔드포인트 (/resources, /tools 등) 테스트
- 요청/응답 형식: 스키마 준수 여부 검증
- 오류 코드: 다양한 시나리오에서 올바른 상태 코드 확인
- 콘텐츠 유형: 다양한 콘텐츠 유형 처리 테스트
- 인증 흐름: 사양에 맞는 인증 메커니즘 검증
준수 테스트 스위트
[Fact]
public async Task Server_ResourceEndpoint_ReturnsCorrectSchema()
{
// Arrange
var client = new HttpClient();
client.DefaultRequestHeaders.Add("Authorization", "Bearer test-token");
// Act
var response = await client.GetAsync("http://localhost:5000/api/resources");
var content = await response.Content.ReadAsStringAsync();
var resources = JsonSerializer.Deserialize<ResourceList>(content);
// Assert
Assert.Equal(HttpStatusCode.OK, response.StatusCode);
Assert.NotNull(resources);
Assert.All(resources.Resources, resource =>
{
Assert.NotNull(resource.Id);
Assert.NotNull(resource.Type);
// Additional schema validation
});
}
효과적인 MCP 서버 테스트를 위한 10가지 팁
- 툴 정의를 별도로 테스트: 툴 로직과 독립적으로 스키마 정의 확인
- 매개변수화된 테스트 사용: 다양한 입력, 특히 경계 조건으로 툴 테스트
- 오류 응답 확인: 가능한 모든 오류 조건에서 적절한 오류 처리 검증
- 인가 로직 테스트: 다양한 사용자 역할에 대한 적절한 접근 제어 보장
- 테스트 커버리지 모니터링: 중요한 경로 코드의 높은 커버리지 목표
- 스트리밍 응답 테스트: 스트리밍 콘텐츠의 적절한 처리 검증
- 네트워크 문제 시뮬레이션: 열악한 네트워크 조건 하에서 동작 테스트
- 리소스 한도 테스트: 할당량 또는 속도 제한 도달 시 동작 확인
- 회귀 테스트 자동화: 코드 변경 시마다 실행되는 테스트 스위트 구축
- 테스트 케이스 문서화: 테스트 시나리오를 명확히 문서화 유지
일반적인 테스트 실패 사례
- 정상 경로 테스트에 과도한 의존: 오류 케이스도 철저히 테스트할 것
- 성능 테스트 무시: 프로덕션 영향 이전에 병목 현상 확인
- 단독 테스트만 진행: 단위, 통합, E2E 테스트 결합 필요
- 불완전한 API 커버리지: 모든 엔드포인트와 기능 테스트 보장
- 불일치한 테스트 환경: 일관된 테스트 환경 확보를 위해 컨테이너 사용
결론
신뢰성 있고 고품질의 MCP 서버 개발을 위해서는 포괄적인 테스트 전략이 필수적입니다. 이 가이드에 제시된 모범 사례와 팁을 구현하면 MCP 구현이 최고 수준의 품질, 신뢰성, 성능을 충족할 수 있습니다.
주요 사항 요약
- 툴 설계: 단일 책임 원칙 준수, 의존성 주입 사용, 조합 가능하도록 설계
- 스키마 설계: 명확하고 문서화가 잘 된 스키마 생성 및 적절한 검증 제약 조건 포함
- 오류 처리: 우아한 오류 처리, 구조화된 오류 응답 및 재시도 로직 구현
- 성능: 캐싱, 비동기 처리, 리소스 제한 조절 사용
- 보안: 철저한 입력 검증, 인가 검사, 민감 데이터 처리 적용
- 테스트: 포괄적인 단위, 통합, 엔드투엔드 테스트 생성
- 워크플로우 패턴: 체인, 디스패처, 병렬 처리 같은 확립된 패턴 적용
연습 문제
문서 처리 시스템을 위한 MCP 툴과 워크플로우를 설계하세요:
- 여러 포맷(PDF, DOCX, TXT)의 문서 수락
- 문서에서 텍스트 및 주요 정보 추출
- 문서 유형과 내용에 따른 분류
- 각 문서 요약 생성
이 시나리오에 가장 적합한 툴 스키마, 오류 처리, 워크플로우 패턴을 구현하세요. 이 구현을 어떻게 테스트할지도 고려하세요.
자료
- 최신 개발 동향을 확인하기 위해 Microsoft Foundry Discord Community의 MCP 커뮤니티에 참여하세요
- 오픈 소스 MCP 프로젝트에 기여하세요
- 조직의 AI 이니셔티브에 MCP 원칙 적용
- 업계 특화 MCP 구현 탐색
- 다중 모드 통합이나 엔터프라이즈 애플리케이션 통합 등 특정 MCP 주제의 고급 과정 수강 고려
- Hands on Lab를 통해 배운 원칙으로 자체 MCP 툴과 워크플로우 실험
다음 단계
다음: 사례 연구
면책 조항: 이 문서는 AI 번역 서비스 Co-op Translator를 사용하여 번역되었습니다. 정확성을 기하기 위해 노력하고 있으나, 자동 번역은 오류나 부정확한 부분이 있을 수 있음을 유의하시기 바랍니다. 원본 문서의 원어본이 권위 있는 자료로 간주되어야 합니다. 중요한 정보의 경우, 전문가의 인간 번역을 권장합니다. 이 번역 사용으로 인해 발생하는 오해나 잘못된 해석에 대해 당사는 책임을 지지 않습니다.
