社区交流 · 2026-09-21

MCP 开发最佳实践

开源教程转载｜MCP 开发最佳实践。Microsoft 及项目贡献者编写，智链谷平台运营整理。

> 开源教程转载 · 由智链谷平台运营账号整理发布。作者：Microsoft 及项目贡献者。
> [查看原文](https://github.com/microsoft/mcp-for-beginners/blob/2f43408b764326df5fc928984c6dd7db2b8fb63e/translations/zh-CN/08-BestPractices/README.md) · MIT 许可，完整声明见文末。

# MCP 开发最佳实践

[![MCP Development Best Practices](https://raw.githubusercontent.com/microsoft/mcp-for-beginners/2f43408b764326df5fc928984c6dd7db2b8fb63e/translated_images/zh-CN/09.d0f6d86c9d72134c.webp)](https://youtu.be/W56H9W7x-ao)

_(点击上方图片观看本课程视频)_

## 概览

本课程聚焦于在生产环境中开发、测试和部署 MCP 服务器及功能的高级最佳实践。随着 MCP 生态系统复杂性和重要性的增加，遵循既定模式能确保可靠性、可维护性和互操作性。本课程整合了来自真实 MCP 实现中的实用经验，指导您构建具有高效资源、提示和工具的坚固高效服务器。

## 学习目标

完成本课程后，您将能够：

- 应用 MCP 服务器和功能设计的行业最佳实践
- 制定 MCP 服务器的全面测试策略
- 设计复杂 MCP 应用的高效、可复用的工作流模式
- 实现 MCP 服务器中的适当错误处理、日志记录和可观测性
- 优化 MCP 实现，提升性能、安全性和可维护性

## MCP 核心原则

在深入特定实现实践之前，理解指导有效 MCP 开发的核心原则非常重要：

1. 标准化通信：MCP 以 JSON-RPC 2.0 为基础，为所有实现提供一致的请求、响应和错误处理格式。

2. 以用户为中心的设计：始终优先考虑用户的同意、控制和透明度。

3. 安全优先：实施强健的安全措施，包括身份验证、授权、校验和速率限制。

4. 模块化架构：设计 MCP 服务器时采取模块化方法，每个工具和资源拥有明确、专注的职责。

5. 有状态连接：利用 MCP 维持跨多个请求的状态，实现更连贯、有上下文感知的交互。

## 官方 MCP 最佳实践

以下最佳实践来源于官方模型上下文协议文档：

### 安全最佳实践

1. 用户同意与控制：访问数据或执行操作前始终要求用户明确同意。提供清晰的控制权限，说明共享哪些数据和授权哪些操作。

2. 数据隐私：仅在明确同意下暴露用户数据，并采用适当访问控制保护数据。防止未经授权的数据传输。

3. 工具安全：调用任何工具前要求明确用户同意。确保用户理解每个工具的功能并强制实施安全边界。

4. 工具权限控制：配置会话期间模型可使用的工具，确保仅访问明确授权的工具。

5. 身份验证：在授予工具、资源或敏感操作访问权限前进行适当身份验证，使用 API 密钥、OAuth 令牌或其他安全认证方式。

6. 参数校验：对所有工具调用进行校验，防止格式错误或恶意输入进入工具实现。

7. 速率限制：实施速率限制，防止滥用并确保服务器资源的公平使用。

### 实现最佳实践

1. 能力协商：连接建立时交换支持的功能、协议版本、可用工具和资源的信息。

2. 工具设计：创建专注单一职责的工具，而非处理多个问题的单块工具。

3. 错误处理：实施标准化错误消息和代码，帮助诊断问题，优雅处理失败并提供可操作的反馈。

4. 日志记录：配置结构化日志，用于审计、调试和监控协议交互。

5. 进度跟踪：针对长时间运行操作报告进度更新，以支持响应式用户界面。

6. 请求取消：允许客户端取消不再需要或超时的进行中请求。

## 额外参考资料

有关 MCP 最佳实践的最新信息，请参阅：

- [MCP 文档](https://modelcontextprotocol.io/)
- [MCP 规范 (2025-11-25)](https://spec.modelcontextprotocol.io/specification/2025-11-25/)
- [GitHub 仓库](https://github.com/modelcontextprotocol)
- [安全最佳实践](https://modelcontextprotocol.io/specification/draft/basic/security_best_practices)
- [OWASP MCP 前 10](https://microsoft.github.io/mcp-azure-security-guide/mcp/) - 安全风险与缓解措施
- [MCP 安全峰会工作坊 (Sherpa)](https://azure-samples.github.io/sherpa/) - 实操安全培训

## 实践实现示例

### 工具设计最佳实践

#### 1. 单一职责原则

每个 MCP 工具应有明确且专注的目的。避免创建试图处理多重关注点的庞大工具，而应开发专门处理特定任务的工具。

```csharp
// 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. 一致的错误处理

实现健壮的错误处理，提供信息充分的错误消息和适当的恢复机制。

```python
# 带有全面错误处理的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注入检测的实现
pass

def _log_error(self, message, error):
# 错误日志记录的实现
pass
```

#### 3. 参数校验

始终彻底校验参数，防止格式错误或恶意输入。

```javascript
// 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
// 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. 速率限制

```csharp
// 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
// 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. 集成测试

测试从客户端请求到服务器响应的完整流程：

```python
# 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()
```

## 性能优化

### 1. 缓存策略

实施合适的缓存，以降低延迟和资源消耗：

```csharp
// 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. 依赖注入与可测试性

设计工具通过构造函数注入依赖，使其可测试且可配置：

```java
// 使用依赖注入的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. 可组合工具

设计可组合的工具，以构建更复杂的工作流：

```python
# 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

# 这些工具可以独立使用，也可以作为工作流程的一部分使用
```

### 架构设计最佳实践

架构是模型与工具之间的契约。良好设计的架构可提升工具的可用性。

#### 1. 清晰的参数描述

始终为每个参数包含描述信息：

```csharp
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. 校验约束

包含校验约束以防止无效输入：

```java
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. 一致的返回结构

保持响应结构的一致性，方便模型解读结果：

```python
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. 优雅的错误处理

在适当层级处理错误并提供信息丰富的消息：

```csharp
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. 结构化错误响应

尽可能返回结构化的错误信息：

```java
@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. 重试逻辑

针对临时失败实现合适的重试逻辑：

```python
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. 缓存

为耗费资源的操作实现缓存：

```csharp
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 绑定操作使用异步编程模式：

```java
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. 资源限流

实施资源限流以防止过载：

```python
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. 输入校验

始终彻底校验输入参数：

```csharp
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. 授权检查

实现适当的授权检查：

```java
@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. 敏感数据处理

谨慎处理敏感数据：

```python
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. 单独测试每个工具

为每个工具的功能创建专注测试：

```csharp
[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. 架构校验测试

测试架构的有效性及约束执行：

```java
@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. 错误处理测试

为错误条件编写专门测试：

```python
@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
```

### 集成测试

#### 1. 工具链测试

测试多个工具按预期组合工作：

```csharp
[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 服务器上的完整执行：

```java
@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. 端到端测试

测试从模型提示到工具执行的完整工作流：

```python
@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 服务器可处理的并发请求数：

```csharp
[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. 压力测试

测试系统在极端负载下的表现：

```java
@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); // 平均响应时间 < 200毫秒
assertTrue(summaryReport.getPercentile(90.0) < 500); // 第90百分位 < 500毫秒
}
```

#### 3. 监控与性能剖析

设置监控用于长期性能分析：

```python
# 配置 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. 工具链模式

将多个工具按序连接，每个工具的输出作为下一个工具的输入：

```python
# Python 工具链实现
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. 分发器模式

使用中央工具根据输入分发到专门工具：

```csharp
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. 并行处理模式

同时执行多个工具以提高效率：

```java
public class ParallelDataProcessingWorkflow {
private final McpClient mcpClient;

public ParallelDataProcessingWorkflow(McpClient mcpClient) {
this.mcpClient = mcpClient;
}

public WorkflowResult execute(String datasetId) {
// 第一步：获取数据集元数据（同步）
ToolResponse metadataResponse = mcpClient.executeTool("datasetMetadata",
Map.of("datasetId", datasetId));

// 第二步：并行启动多个分析
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(); // 等待完成

// 第三步：合并结果
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());

// 第四步：生成总结报告
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. 错误恢复模式

实现工具失败时的优雅回退：

```python
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. 工作流组合模式

通过组合更简单的工作流构建复杂工作流：

```csharp
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 服务器作为 AI 模型与客户端应用之间的重要中间层，彻底测试可确保：

- 生产环境中的可靠性
- 请求和响应的准确处理
- MCP 规范的正确执行
- 对故障和边界情况的韧性
- 各种负载下的性能一致性

## MCP 服务器单元测试

### 单元测试（基础层）

单元测试隔离验证 MCP 服务器各组件。

#### 测试内容

1. 资源处理器：独立测试每个资源处理器的逻辑
2. 工具实现：验证工具在不同输入下的行为
3. 提示模板：确保提示模板渲染正确
4. 架构校验：测试参数校验逻辑
5. 错误处理：验证无效输入的错误响应

#### 单元测试最佳实践

```csharp
// 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
# 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
```

### 集成测试（中间层）

集成测试验证 MCP 服务器组件间的交互。

#### 测试内容

1. 服务器初始化：测试带不同配置的服务器启动
2. 路由注册：验证所有端点正确注册
3. 请求处理：测试完整请求-响应流程
4. 错误传播：确保错误在组件间恰当处理
5. 身份验证与授权：测试安全机制

#### 集成测试最佳实践

```csharp
// 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();
}
```

### 端到端测试（顶层）

端到端测试验证从客户端到服务器的完整系统行为。

#### 测试内容

1. 客户端与服务器通信：测试完整请求-响应周期
2. **真实客户端 SDK**：使用实际客户端实现测试
3. 负载下性能：验证多并发请求下的行为
4. 错误恢复：测试系统从故障中恢复
5. 长时间运行操作：验证流处理和长操作支持

#### 端到端测试最佳实践

```typescript
// 使用 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');
});
});
```

## MCP 测试中的模拟策略

模拟对于隔离组件测试必不可少。

### 需要模拟的组件

1. **外部 AI 模型**：模拟模型响应以实现可预测测试
2. 外部服务：模拟 API 依赖（数据库、第三方服务）
3. 身份验证服务：模拟身份提供者
4. 资源提供者：模拟高代价资源处理器

### 示例：模拟 AI 模型响应

```csharp
// 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
# 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)
# 继续测试
```

## 性能测试

性能测试对生产 MCP 服务器至关重要。

### 测量内容

1. 延迟：请求响应时间
2. 吞吐量：每秒处理请求数
3. 资源利用：CPU、内存、网络使用情况
4. 并发处理：并行请求下的行为
5. 扩展特性：负载增加时的性能表现

### 性能测试工具

- **k6**：开源负载测试工具
- **JMeter**：综合性能测试工具
- **Locust**：基于 Python 的负载测试
- **Azure 负载测试**：云端性能测试

### 示例：使用 k6 的基础负载测试

```javascript
// 用于负载测试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 集成

1. 在拉取请求时运行单元测试：确保代码变更不破坏现有功能
2. 预发布环境的集成测试：在预生产环境中运行集成测试
3. 性能基准：维护性能基准以捕捉回归
4. 安全扫描：将安全测试自动化作为流水线的一部分

### 示例 CI 流水线 (GitHub Actions)

```yaml
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 规范。

### 关键合规领域

1. **API 端点**：测试必需的端点（/resources，/tools 等）
2. **请求/响应格式**：验证模式合规性
3. 错误代码：验证各种场景的正确状态码
4. 内容类型：测试不同内容类型的处理
5. 认证流程：验证符合规范的认证机制

### 合规测试套件

```csharp
[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 服务器测试的十大建议

1. 单独测试工具定义：独立验证模式定义，而非工具逻辑
2. 使用参数化测试：使用多种输入（包括边缘情况）测试工具
3. 检查错误响应：验证所有可能错误情况的正确错误处理
4. 测试授权逻辑：确保不同用户角色的访问控制正确
5. 监控测试覆盖率：争取关键路径代码的高覆盖率
6. 测试流式响应：验证流式内容的正确处理
7. 模拟网络问题：测试差网络条件下的行为
8. 测试资源限制：验证达到配额或速率限制时的表现
9. 自动化回归测试：建立在每次代码变更时运行的测试套件
10. 记录测试用例：维护清晰的测试场景文档

## 常见测试陷阱

- **过度依赖“顺畅路径”测试**：确保彻底测试错误情况
- 忽视性能测试：在影响生产前识别瓶颈
- 仅孤立测试：结合单元测试、集成测试和端到端测试
- **API 覆盖不全面**：确保所有端点和功能均被测试
- 测试环境不一致：使用容器确保一致的测试环境

## 结论

制定全面的测试策略对于开发可靠、高质量的 MCP 服务器至关重要。通过实施本指南中列出的最佳实践和提示，您可以确保 MCP 实现达到最高的质量、可靠性和性能标准。

## 主要收获

1. 工具设计：遵循单一职责原则，使用依赖注入，设计为可组合
2. 模式设计：创建清晰、文档完善且带有适当验证约束的模式
3. 错误处理：实现优雅的错误处理、结构化错误响应和重试逻辑
4. 性能：使用缓存、异步处理和资源限制
5. 安全：应用彻底的输入验证、授权检查和敏感数据处理
6. 测试：创建全面的单元测试、集成测试和端到端测试
7. 工作流模式：应用诸如链、调度器和平行处理等成熟模式

## 练习

设计一个针对文档处理系统的 MCP 工具和工作流，该系统需要：

1. 接受多种格式的文档（PDF、DOCX、TXT）
2. 从文档中提取文本和关键信息
3. 根据类型和内容对文档进行分类
4. 生成每个文档的摘要

实现工具模式、错误处理和适合此场景的工作流模式。考虑如何测试该实现。

## 资源

1. 加入 [Microsoft Foundry Discord Community](https://aka.ms/foundrydevs) 上的 MCP 社区，获取最新动态
2. 参与开源的 [MCP 项目](https://github.com/modelcontextprotocol)
3. 在您组织的 AI 计划中应用 MCP 原则
4. 探索针对您行业的专门 MCP 实现
5. 考虑参加关于特定 MCP 主题的高级课程，例如多模态集成或企业应用集成
6. 通过 [Hands on Lab](https://github.com/microsoft/mcp-for-beginners/blob/2f43408b764326df5fc928984c6dd7db2b8fb63e/translations/zh-CN/10-StreamliningAIWorkflowsBuildingAnMCPServerWithAIToolkit/README.md) 学习的原则，尝试构建自己的 MCP 工具和工作流

## 下一步

下一章：[案例研究](https://github.com/microsoft/mcp-for-beginners/blob/2f43408b764326df5fc928984c6dd7db2b8fb63e/translations/zh-CN/09-CaseStudy/README.md)

---

**免责声明**：
本文件由 AI 翻译服务 [Co-op Translator](https://github.com/Azure/co-op-translator) 翻译完成。尽管我们力求准确，但请注意，自动翻译可能包含错误或不准确之处。原始语言版文件应视为权威来源。对于重要信息，建议使用专业人工翻译。我们对因使用本翻译而产生的任何误解或误释不承担责任。

---

## 版权与许可

```text
MIT License

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE
```