wehub-resource-sync
108 KiB
108 KiB
MCP Development Best Practices
(คลิกที่รูปภาพด้านบนเพื่อดูวิดีโอบทเรียนนี้)
Overview
บทเรียนนี้มุ่งเน้นไปที่แนวปฏิบัติที่ดีที่สุดขั้นสูงสำหรับการพัฒนา ทดสอบ และปรับใช้เซิร์ฟเวอร์และฟีเจอร์ MCP ในสภาพแวดล้อมการผลิต ขณะที่ระบบนิเวศ MCP เพิ่มความซับซ้อนและความสำคัญ การปฏิบัติตามรูปแบบที่กำหนดไว้ช่วยให้มั่นใจในความน่าเชื่อถือ การบำรุงรักษา และการทำงานร่วมกัน บทเรียนนี้รวบรวมปัญญาจากการใช้งาน MCP ในโลกจริงเพื่อแนะนำคุณในการสร้างเซิร์ฟเวอร์ที่แข็งแกร่ง ประสิทธิภาพสูง ด้วยทรัพยากร ตัวกระตุ้น และเครื่องมือที่มีประสิทธิผล
Learning Objectives
เมื่อจบบทเรียนนี้ คุณจะสามารถ:
- ใช้แนวปฏิบัติที่ดีที่สุดในอุตสาหกรรมในการออกแบบเซิร์ฟเวอร์และฟีเจอร์ MCP
- สร้างกลยุทธ์การทดสอบที่ครอบคลุมสำหรับเซิร์ฟเวอร์ MCP
- ออกแบบรูปแบบงานที่มีประสิทธิภาพ และนำกลับมาใช้ใหม่ได้สำหรับแอปพลิเคชัน MCP ที่ซับซ้อน
- นำแนวทางจัดการข้อผิดพลาด การบันทึก และการสังเกตการณ์ที่เหมาะสมในเซิร์ฟเวอร์ MCP
- ปรับแต่งการใช้งาน MCP เพื่อประสิทธิภาพ ความปลอดภัย และการบำรุงรักษา
MCP Core Principles
ก่อนลงลึกสู่การปฏิบัติการใช้งานเฉพาะ ควรเข้าใจหลักการสำคัญที่ชี้นำการพัฒนา MCP ที่มีประสิทธิภาพ:
-
การสื่อสารที่ได้มาตรฐาน: MCP ใช้ JSON-RPC 2.0 เป็นฐาน เพื่อจัดรูปแบบที่สอดคล้องกันสำหรับคำขอ คำตอบ และการจัดการข้อผิดพลาดในทุกการใช้งาน
-
การออกแบบที่ยึดผู้ใช้เป็นศูนย์กลาง: ให้ความสำคัญกับความยินยอม การควบคุม และความโปร่งใสของผู้ใช้ในการใช้งาน MCP
-
ความปลอดภัยเป็นอันดับแรก: นำมาตรการความปลอดภัยที่แข็งแกร่งรวมถึงการตรวจสอบสิทธิ์ การอนุญาต การตรวจสอบ และจำกัดอัตราการใช้งาน
-
สถาปัตยกรรมแบบโมดูลาร์: ออกแบบเซิร์ฟเวอร์ MCP ด้วยแนวทางโมดูลาร์ โดยให้แต่ละเครื่องมือและทรัพยากรมีวัตถุประสงค์ที่ชัดเจนและมุ่งเน้น
-
การเชื่อมต่อแบบมีสถานะ: ใช้ความสามารถของ MCP ในการรักษาสถานะข้ามคำขอหลายรายการเพื่อให้มีปฏิสัมพันธ์ที่สอดคล้องและรับรู้บริบทได้ดีขึ้น
Official MCP Best Practices
แนวปฏิบัติที่ดีที่สุดต่อไปนี้ได้มาจากเอกสาร Model Context Protocol อย่างเป็นทางการ:
Security Best Practices
-
ความยินยอมและการควบคุมของผู้ใช้: ต้องขอความยินยอมจากผู้ใช้อย่างชัดเจนก่อนเข้าถึงข้อมูลหรือดำเนินการใด ๆ ให้อำนาจควบคุมที่ชัดเจนว่าส่งแชร์ข้อมูลใดและอนุญาตให้ทำการใดบ้าง
-
ความเป็นส่วนตัวของข้อมูล: เปิดเผยข้อมูลของผู้ใช้เฉพาะเมื่อได้รับความยินยอมอย่างชัดเจนและคุ้มครองด้วยการควบคุมการเข้าถึงที่เหมาะสม ป้องกันการส่งข้อมูลโดยไม่ได้รับอนุญาต
-
ความปลอดภัยของเครื่องมือ: ต้องมีความยินยอมจากผู้ใช้อย่างชัดเจนก่อนเรียกใช้เครื่องมือใด ๆ ให้ผู้ใช้เข้าใจฟังก์ชันการทำงานของแต่ละเครื่องมือและบังคับใช้ขอบเขตความปลอดภัยที่เข้มงวด
-
การควบคุมสิทธิ์เครื่องมือ: ตั้งค่ากำหนดว่าเครื่องมือใดที่โมเดลได้รับอนุญาตให้ใช้ในระหว่างการเซสชัน เพื่อให้เข้าถึงได้เฉพาะเครื่องมือที่ได้รับสิทธิ์ชัดเจนเท่านั้น
-
การตรวจสอบสิทธิ์: ต้องมีการตรวจสอบสิทธิ์ที่ถูกต้องก่อนอนุญาตให้เข้าถึงเครื่องมือ ทรัพยากร หรือการดำเนินการที่มีความอ่อนไหว เช่น ใช้ API keys, OAuth tokens หรือวิธีการตรวจสอบสิทธิ์ที่ปลอดภัยอื่น ๆ
-
การตรวจสอบพารามิเตอร์: บังคับใช้การตรวจสอบความถูกต้องของพารามิเตอร์ทุกการเรียกใช้เครื่องมือเพื่อป้องกันข้อมูลป้อนเข้าที่ผิดรูปแบบหรือเป็นอันตราย
-
การจำกัดอัตรา: นำการจำกัดอัตราการใช้งานมาใช้เพื่อป้องกันการละเมิดและรักษาการใช้งานทรัพยากรเซิร์ฟเวอร์อย่างเป็นธรรม
Implementation Best Practices
-
การเจรจาความสามารถ: ในระหว่างการตั้งค่าการเชื่อมต่อ ให้แลกเปลี่ยนข้อมูลเกี่ยวกับฟีเจอร์ที่รองรับ เวอร์ชันโปรโตคอล เครื่องมือและทรัพยากรที่มี
-
การออกแบบเครื่องมือ: สร้างเครื่องมือที่มุ่งเน้นปฏิบัติหน้าที่หนึ่งอย่างดี แทนที่จะสร้างเครื่องมือรวมที่จัดการหลายแง่มุม
-
การจัดการข้อผิดพลาด: นำข้อความและรหัสข้อผิดพลาดที่เป็นมาตรฐานมาใช้เพื่อช่วยวินิจฉัยปัญหา จัดการความล้มเหลวอย่างเหมาะสม และให้ข้อมูลตอบกลับที่นำไปใช้ได้จริง
-
การบันทึกข้อมูล: ตั้งค่าการบันทึกแบบมีโครงสร้างสำหรับตรวจสอบ แก้ไขปัญหา และติดตามการโต้ตอบโปรโตคอล
-
การติดตามความคืบหน้า: สำหรับการดำเนินการที่ใช้เวลานาน ให้รายงานอัพเดตความคืบหน้าเพื่อเปิดใช้งานส่วนติดต่อผู้ใช้ที่ตอบสนองได้ดี
-
การยกเลิกคำขอ: อนุญาตให้ไคลเอนต์ยกเลิกคำขอกำลังดำเนินการที่ไม่จำเป็นอีกต่อไปหรือใช้เวลานานเกินควร
Additional References
สำหรับข้อมูลที่อัพเดตล่าสุดเกี่ยวกับแนวปฏิบัติที่ดีที่สุดของ MCP โปรดดูที่:
- MCP Documentation
- MCP Specification (2025-11-25)
- GitHub Repository
- Security Best Practices
- OWASP MCP Top 10 - ความเสี่ยงด้านความปลอดภัยและการป้องกัน
- MCP Security Summit Workshop (Sherpa) - การฝึกอบรมความปลอดภัยแบบปฏิบัติ
Practical Implementation Examples
Tool Design Best Practices
1. Single Responsibility Principle
เครื่องมือ 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. Consistent Error Handling
นำการจัดการข้อผิดพลาดที่แข็งแกร่งพร้อมข้อความข้อผิดพลาดที่ให้ข้อมูลและกลไกการกู้คืนที่เหมาะสมมาใช้
# ตัวอย่าง Python พร้อมการจัดการข้อผิดพลาดอย่างครอบคลุม
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 injection
pass
def _log_error(self, message, error):
# การใช้งานบันทึกข้อผิดพลาด
pass
3. Parameter Validation
ตรวจสอบพารามิเตอร์อย่างละเอียดเสมอเพื่อป้องกันการป้อนข้อมูลที่ผิดรูปแบบหรือเป็นอันตราย
// ตัวอย่าง 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) {
// การดำเนินการตรวจสอบความปลอดภัยของเส้นทาง
// ...
}
}
Security Implementation Examples
1. Authentication and Authorization
// ตัวอย่าง 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. Rate Limiting
// 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
// ...
}
}
Testing Best Practices
1. Unit Testing MCP Tools
ทดสอบเครื่องมือของคุณในสภาพแวดล้อมแยก โดยใช้การจำลอง (mock) ของการพึ่งพาภายนอก:
// ตัวอย่างการทดสอบหน่วยเครื่องมือโดยใช้ 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. Integration Testing
ทดสอบการทำงานครบวงจรตั้งแต่คำขอของไคลเอนต์จนถึงการตอบกลับของเซิร์ฟเวอร์:
# ตัวอย่างการทดสอบรวม Python
@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()
Performance Optimization
1. Caching Strategies
นำกลยุทธ์การแคชที่เหมาะสมมาใช้เพื่อลดความหน่วงและการใช้ทรัพยากร:
// 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. Dependency Injection and Testability
ออกแบบเครื่องมือให้รับการพึ่งพาผ่านการฉีดค่าตอนสร้าง (constructor injection) เพื่อให้ทดสอบและตั้งค่าได้ง่าย:
// ตัวอย่าง Java พร้อมการฉีดพึ่งพา
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. Composable Tools
ออกแบบเครื่องมือที่สามารถผสมผสานเข้าด้วยกันเพื่อสร้างเวิร์กโฟลว์ที่ซับซ้อนกว่า:
# ตัวอย่าง Python แสดงเครื่องมือที่สามารถประกอบกันได้
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
# เครื่องมือเหล่านี้สามารถใช้ได้อย่างอิสระหรือเป็นส่วนหนึ่งของเวิร์กโฟลว์
Schema Design Best Practices
สคีมาเป็นสัญญาระหว่างโมเดลกับเครื่องมือของคุณ สคีมาที่ออกแบบดีนำไปสู่การใช้งานเครื่องมือที่ง่ายขึ้น
1. Clear Parameter Descriptions
รวมข้อมูลคำอธิบายสำหรับแต่ละพารามิเตอร์เสมอ:
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. Validation Constraints
รวมขอบเขตการตรวจสอบความถูกต้องเพื่อป้องกันข้อมูลป้อนเข้าที่ไม่ถูกต้อง:
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. Consistent Return Structures
รักษาความสอดคล้องในโครงสร้างผลลัพธ์เพื่อง่ายแก่การตีความโดยโมเดล:
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
}
Error Handling
การจัดการข้อผิดพลาดที่แข็งแรงมีความสำคัญสำหรับเครื่องมือ MCP เพื่อรักษาความน่าเชื่อถือ
1. Graceful Error Handling
จัดการข้อผิดพลาดในระดับที่เหมาะสมและให้ข้อความที่ให้ข้อมูล:
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. Structured Error Responses
ส่งคืนข้อมูลข้อผิดพลาดในรูปแบบโครงสร้างเมื่อเป็นไปได้:
@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. Retry Logic
นำตรรกะการลองซ้ำที่เหมาะสมสำหรับความล้มเหลวชั่วคราวมาใช้:
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)}")
Performance Optimization
1. Caching
ใช้การแคชสำหรับการดำเนินการที่มีค่าใช้จ่ายสูง:
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. Asynchronous Processing
ใช้รูปแบบการเขียนโปรแกรมแบบอะซิงโครนัสสำหรับการปฏิบัติการที่ผูกกับ 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();
// สำหรับการดำเนินการที่ใช้เวลานาน ให้ส่งคืนรหัสการประมวลผลทันที
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);
// ส่งคืนการตอบสนองทันทีพร้อมกับรหัสกระบวนการ
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. Resource Throttling
นำการจำกัดทรัพยากรมาป้องกันการทำงานหนักเกินไป:
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
Security Best Practices
1. Input Validation
ตรวจสอบพารามิเตอร์ป้อนเข้าทุกครั้งอย่างละเอียด:
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. Authorization Checks
นำการตรวจสอบการอนุญาตที่เหมาะสมมาใช้:
@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. Sensitive Data Handling
จัดการข้อมูลที่อ่อนไหวอย่างรอบคอบ:
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
Testing Best Practices for MCP Tools
การทดสอบที่ครอบคลุมช่วยให้มั่นใจว่าเครื่องมือ MCP ทำงานถูกต้อง จัดการกรณีขอบ และรวมเข้าได้ดีกับระบบ
Unit Testing
1. Test Each Tool in Isolation
สร้างการทดสอบที่มุ่งเน้นสำหรับฟังก์ชันแต่ละของเครื่องมือ:
[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. Schema Validation Testing
ทดสอบให้แน่ใจว่าสคีมาถูกต้องและบังคับใช้ข้อจำกัดได้อย่างเหมาะสม:
@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. Error Handling Tests
สร้างการทดสอบเฉพาะสำหรับสถานการณ์ข้อผิดพลาด:
@pytest.mark.asyncio
async def test_api_tool_handles_timeout():
# จัดเรียง
tool = ApiTool(timeout=0.1) # หมดเวลารวดเร็วมาก
# จำลองคำร้องขอที่จะหมดเวลา
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()
# จำลองการตอบกลับที่จำกัดอัตรา
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
Integration Testing
1. Tool Chain Testing
ทดสอบเครื่องมือที่ทำงานร่วมกันในชุดค่าที่คาดหวัง:
[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 Server Testing
ทดสอบเซิร์ฟเวอร์ 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. End-to-End Testing
ทดสอบเวิร์กโฟลว์ครบถ้วนตั้งแต่การกระตุ้นโมเดลจนถึงการทำงานของเครื่องมือ:
@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"
Performance Testing
1. Load Testing
ทดสอบจำนวนคำขอพร้อมกันที่เซิร์ฟเวอร์ 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. Stress Testing
ทดสอบระบบภายใต้การโหลดสูงสุด:
@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 sampler สำหรับการเรียกใช้เครื่องมือ
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. Monitoring and Profiling
ตั้งค่าการตรวจสอบเพื่อวิเคราะห์ประสิทธิภาพในระยะยาว:
# กำหนดการตรวจสอบสำหรับเซิร์ฟเวอร์ 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 Workflow Design Patterns
เวิร์กโฟลว์ MCP ที่ออกแบบดีช่วยเพิ่มประสิทธิภาพ ความน่าเชื่อถือ และการบำรุงรักษา ต่อไปนี้เป็นรูปแบบสำคัญที่ควรปฏิบัติตาม:
1. Chain of Tools Pattern
เชื่อมเครื่องมือหลายตัวในลำดับที่ผลลัพธ์ของเครื่องมือหนึ่งเป็นข้อมูลนำเข้าของถัดไป:
# การใช้งาน Python Chain of Tools
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. Dispatcher Pattern
ใช้เครื่องมือกลางที่จัดการส่งงานไปยังเครื่องมือเฉพาะทางตามข้อมูลนำเข้า:
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. Parallel Processing Pattern
ดำเนินการเครื่องมือหลายตัวพร้อมกันเพื่อประสิทธิภาพ:
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. Error Recovery Pattern
ใช้การสำรองแบบนุ่มนวลเมื่อเครื่องมือเกิดข้อผิดพลาด:
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. Workflow Composition Pattern
สร้างเวิร์กโฟลว์ที่ซับซ้อนโดยการประสานเวิร์กโฟลว์ที่ง่ายขึ้น:
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" }
});
Testing MCP Servers: Best Practices and Top Tips
Overview
การทดสอบเป็นส่วนสำคัญของการพัฒนาเซิร์ฟเวอร์ MCP ที่น่าเชื่อถือและคุณภาพสูง คู่มือนี้ให้แนวปฏิบัติที่ดีที่สุดและเคล็ดลับสำหรับการทดสอบเซิร์ฟเวอร์ MCP ของคุณตลอดวงจรชีวิตการพัฒนา ตั้งแต่การทดสอบหน่วยถึงการทดสอบแบบผสานรวมและการตรวจสอบตั้งแต่ต้นจนจบ
Why Testing Matters for MCP Servers
เซิร์ฟเวอร์ MCP ทำหน้าที่เป็นมิดเดิลแวร์สำคัญระหว่างโมเดล AI และแอปพลิเคชันของไคลเอนต์ การทดสอบอย่างละเอียดช่วยให้มั่นใจใน:
- ความน่าเชื่อถือในสภาพแวดล้อมการผลิต
- การจัดการคำขอและคำตอบที่ถูกต้อง
- การใช้งานสเปค MCP อย่างเหมาะสม
- ความต้านทานต่อความล้มเหลวและกรณีขอบ
- ประสิทธิภาพที่สม่ำเสมอภายใต้ภาระงานต่าง ๆ
Unit Testing for MCP Servers
Unit Testing (Foundation)
การทดสอบหน่วยตรวจสอบองค์ประกอบแต่ละส่วนของเซิร์ฟเวอร์ MCP ของคุณอย่างแยกส่วน
What to Test
- Resource Handlers: ทดสอบตรรกะของตัวจัดการทรัพยากรแต่ละตัวอย่างอิสระ
- Tool Implementations: ตรวจสอบพฤติกรรมของเครื่องมือด้วยข้อมูลเข้าแบบต่าง ๆ
- Prompt Templates: ตรวจสอบให้แน่ใจว่าแบบร่างคำกระตุ้นแสดงผลถูกต้อง
- Schema Validation: ทดสอบตรรกะการตรวจสอบความถูกต้องของพารามิเตอร์
- Error Handling: ตรวจสอบการตอบสนองข้อผิดพลาดสำหรับข้อมูลเข้าไม่ถูกต้อง
Best Practices for Unit Testing
// 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);
}
# ตัวอย่างการทดสอบหน่วยสำหรับเครื่องมือเครื่องคิดเลขใน Python
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
Integration Testing (Middle Layer)
การทดสอบแบบผสานรวมตรวจสอบการทำงานร่วมกันระหว่างส่วนประกอบของเซิร์ฟเวอร์ MCP ของคุณ
What to Test
- Server Initialization: ทดสอบการเริ่มต้นเซิร์ฟเวอร์ด้วยการตั้งค่าต่าง ๆ
- Route Registration: ตรวจสอบว่าเส้นทางทั้งหมดถูกลงทะเบียนอย่างถูกต้อง
- Request Processing: ทดสอบวงจรคำขอ-คำตอบเต็มรูปแบบ
- Error Propagation: ตรวจสอบให้แน่ใจว่าข้อผิดพลาดถูกจัดการอย่างถูกต้องข้ามส่วนประกอบ
- Authentication & Authorization: ทดสอบกลไกความปลอดภัย
Best Practices for Integration Testing
// 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();
}
End-to-End Testing (Top Layer)
การทดสอบตั้งแต่ต้นจนจบตรวจสอบพฤติกรรมของระบบทั้งหมดจากไคลเอนต์ถึงเซิร์ฟเวอร์
What to Test
- Client-Server Communication: ทดสอบวงจรคำขอ-คำตอบครบถ้วน
- Real Client SDKs: ทดสอบกับการใช้งานไคลเอนต์จริง
- Performance Under Load: ตรวจสอบพฤติกรรมกับคำขอพร้อมกันหลายรายการ
- Error Recovery: ทดสอบการกู้คืนระบบจากความล้มเหลว
- Long-Running Operations: ตรวจสอบการจัดการการสตรีมและการดำเนินการยาว
Best Practices for E2E Testing
// ตัวอย่างการทดสอบ E2E กับไคลเอนต์ใน TypeScript
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');
});
});
Mocking Strategies for MCP Testing
การจำลองจำเป็นสำหรับการแยกส่วนประกอบในระหว่างการทดสอบ
Components to Mock
- External AI Models: จำลองการตอบกลับของโมเดลเพื่อการทดสอบที่คาดการณ์ได้
- External Services: จำลองการพึ่งพา API (ฐานข้อมูล บริการภายนอก)
- Authentication Services: จำลองผู้ให้บริการพิสูจน์ตัวตน
- Resource Providers: จำลองตัวจัดการทรัพยากรที่มีค่าใช้จ่ายสูง
Example: Mocking an AI Model Response
// 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);
# ตัวอย่าง Python กับ unittest.mock
@patch('mcp_server.models.OpenAIModel')
def test_with_mock_model(mock_model):
# กำหนดค่า mock
mock_model.return_value.generate_response.return_value = {
"text": "Mocked model response",
"finish_reason": "completed"
}
# ใช้ mock ในการทดสอบ
server = McpServer(model_client=mock_model)
# ดำเนินการทดสอบต่อ
Performance Testing
การทดสอบประสิทธิภาพมีความสำคัญสำหรับเซิร์ฟเวอร์ MCP ในการผลิต
What to Measure
- Latency: เวลาตอบสนองคำขอ
- Throughput: จำนวนคำขอที่จัดการต่อวินาที
- Resource Utilization: การใช้ CPU, หน่วยความจำ, เครือข่าย
- Concurrency Handling: พฤติกรรมภายใต้คำขอพร้อมกัน
- Scaling Characteristics: ประสิทธิภาพตามภาระการทำงานที่เพิ่มขึ้น
Tools for Performance Testing
- k6: เครื่องมือทดสอบโหลดแบบโอเพนซอร์ส
- JMeter: การทดสอบประสิทธิภาพครบวงจร
- Locust: ทดสอบโหลดด้วย Python
- Azure Load Testing: การทดสอบประสิทธิภาพบนคลาวด์
Example: Basic Load Test with k6
// สคริปต์ k6 สำหรับทดสอบโหลดเซิร์ฟเวอร์ MCP
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);
}
Test Automation for MCP Servers
การทำอัตโนมัติการทดสอบช่วยให้คุณภาพสม่ำเสมอและการตอบกลับรวดเร็วขึ้น
CI/CD Integration
- Run Unit Tests on Pull Requests: ตรวจสอบว่าโค้ดที่เปลี่ยนไม่ทำให้ฟังก์ชันการทำงานเสียหาย
- การทดสอบการรวมระบบในสเตจจิ้ง: รันการทดสอบการรวมระบบในสภาพแวดล้อมก่อนการผลิต
- มาตรฐานประสิทธิภาพ: รักษาเกณฑ์มาตรฐานประสิทธิภาพเพื่อจับความถดถอย
- การสแกนความปลอดภัย: ทำการทดสอบความปลอดภัยโดยอัตโนมัติเป็นส่วนหนึ่งของ pipeline
ตัวอย่าง CI Pipeline (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
});
}
10 เคล็ดลับยอดนิยมสำหรับการทดสอบเซิร์ฟเวอร์ MCP อย่างมีประสิทธิภาพ
- ทดสอบการกำหนดเครื่องมือแยกต่างหาก: ตรวจสอบการกำหนดสคีมาอย่างอิสระจากตรรกะของเครื่องมือ
- ใช้การทดสอบแบบมีพารามิเตอร์: ทดสอบเครื่องมือด้วยอินพุตหลากหลาย รวมถึงกรณีขอบเขต
- ตรวจสอบการตอบกลับข้อผิดพลาด: ตรวจสอบการจัดการข้อผิดพลาดที่เหมาะสมสำหรับทุกเงื่อนไขข้อผิดพลาด
- ทดสอบตรรกะการอนุญาต: ตรวจสอบการควบคุมการเข้าถึงที่เหมาะสมสำหรับบทบาทผู้ใช้ต่าง ๆ
- ตรวจสอบการครอบคลุมการทดสอบ: ตั้งเป้าความครอบคลุมสูงของโค้ดเส้นทางสำคัญ
- ทดสอบการตอบสนองแบบสตรีมมิ่ง: ตรวจสอบการจัดการเนื้อหาสตรีมมิ่งที่ถูกต้อง
- จำลองปัญหาเครือข่าย: ทดสอบพฤติกรรมภายใต้เครือข่ายที่มีปัญหา
- ทดสอบขีดจำกัดทรัพยากร: ตรวจสอบพฤติกรรมเมื่อถึงโควตาหรือขีดจำกัดอัตรา
- อัตโนมัติการทดสอบถดถอย: สร้างชุดทดสอบที่รันทุกครั้งที่โค้ดเปลี่ยนแปลง
- จัดทำเอกสารกรณีทดสอบ: รักษาเอกสารที่ชัดเจนของสถานการณ์การทดสอบ
ข้อผิดพลาดทั่วไปในการทดสอบ
- พึ่งพาการทดสอบเส้นทางที่สมบูรณ์มากเกินไป: ต้องทดสอบกรณีข้อผิดพลาดอย่างละเอียด
- ละเลยการทดสอบประสิทธิภาพ: ระบุคอขวดก่อนที่จะส่งผลกระทบต่อการผลิต
- ทดสอบแยกกันเท่านั้น: รวมการทดสอบหน่วยการทำงาน การรวม และปลายทางถึงปลายทาง
- การครอบคลุม API ที่ไม่สมบูรณ์: ตรวจสอบให้แน่ใจว่าทดสอบทุกจุดเชื่อมต่อและคุณสมบัติ
- สภาพแวดล้อมทดสอบไม่สอดคล้องกัน: ใช้คอนเทนเนอร์เพื่อให้สภาพแวดล้อมทดสอบมีความสอดคล้อง
บทสรุป
กลยุทธ์การทดสอบที่ครอบคลุมเป็นสิ่งจำเป็นสำหรับการพัฒนาเซิร์ฟเวอร์ MCP ที่น่าเชื่อถือและมีคุณภาพสูง โดยการนำแนวทางปฏิบัติที่ดีที่สุดและเคล็ดลับที่ระบุในคู่มือนี้ไปใช้ คุณจะมั่นใจได้ว่าการใช้งาน MCP ของคุณเป็นไปตามมาตรฐานสูงสุดด้านคุณภาพ ความน่าเชื่อถือ และประสิทธิภาพ
ข้อสรุปสำคัญ
- การออกแบบเครื่องมือ: ปฏิบัติตามหลักการความรับผิดชอบเดี่ยว ใช้การฉีดพึ่งพิง และออกแบบให้สามารถประกอบกันได้
- การออกแบบสคีมา: สร้างสคีมาที่ชัดเจน มีเอกสารครบถ้วน พร้อมข้อจำกัดการตรวจสอบที่เหมาะสม
- การจัดการข้อผิดพลาด: ดำเนินการจัดการข้อผิดพลาดอย่างอ่อนโยน ตอบสนองข้อผิดพลาดอย่างมีโครงสร้าง และตรรกะการลองใหม่
- ประสิทธิภาพ: ใช้การแคช การประมวลผลแบบอะซิงโครนัส และการควบคุมทรัพยากร
- ความปลอดภัย: ตรวจสอบข้อมูลนำเข้าอย่างละเอียด ตรวจสอบการอนุญาต และจัดการข้อมูลที่ละเอียดอ่อน
- การทดสอบ: สร้างชุดทดสอบหน่วย การรวม และปลายทางถึงปลายทางที่ครอบคลุม
- รูปแบบการทำงาน: ใช้รูปแบบที่ยอมรับเช่น chains, dispatchers และการประมวลผลคู่ขนาน
แบบฝึกหัด
ออกแบบเครื่องมือและเวิร์กโฟลว์ MCP สำหรับระบบประมวลผลเอกสารที่:
- รับเอกสารในหลายรูปแบบ (PDF, DOCX, TXT)
- สกัดข้อความและข้อมูลสำคัญจากเอกสาร
- จำแนกเอกสารตามประเภทและเนื้อหา
- สร้างสรุปของแต่ละเอกสาร
นำเสนอสคีมาของเครื่องมือ การจัดการข้อผิดพลาด และรูปแบบเวิร์กโฟลว์ที่เหมาะสมกับสถานการณ์นี้ พิจารณาวิธีการทดสอบการใช้งานนี้
แหล่งข้อมูล
- เข้าร่วมชุมชน MCP บน Microsoft Foundry Discord Community เพื่อรับข่าวสารการพัฒนาล่าสุด
- มีส่วนร่วมใน โครงการ MCP แบบโอเพนซอร์ส
- ใช้หลักการ MCP ในโครงการ AI ขององค์กรของคุณ
- สำรวจการใช้งาน MCP เฉพาะทางสำหรับอุตสาหกรรมของคุณ
- พิจารณาการเรียนหลักสูตรขั้นสูงในหัวข้อ MCP เฉพาะ เช่น การรวมโมดูลหลายรูปแบบ หรือการรวมแอปพลิเคชันองค์กร
- ทดลองสร้างเครื่องมือและเวิร์กโฟลว์ MCP ของคุณเองโดยใช้หลักการจาก Hands on Lab
ถัดไป
ต่อไป: กรณีศึกษา
ปฏิเสธความรับผิดชอบ: เอกสารนี้ได้รับการแปลโดยใช้บริการแปลภาษา AI Co-op Translator ขณะที่เราพยายามให้ความถูกต้อง โปรดทราบว่าการแปลโดยอัตโนมัติอาจมีข้อผิดพลาดหรือความไม่ถูกต้อง เอกสารต้นฉบับในภาษาต้นทางควรถูกพิจารณาเป็นแหล่งข้อมูลที่เชื่อถือได้ สำหรับข้อมูลที่สำคัญ แนะนำให้ใช้การแปลโดยมนุษย์มืออาชีพ เราไม่รับผิดชอบต่อความเข้าใจผิดหรือการตีความที่ผิดพลาดที่เกิดขึ้นจากการใช้การแปลนี้