1. MCP 客户端的角色
在 MCP 架构中，客户端是连接 AI 宿主（Host）和 MCP 服务器的桥梁。它负责：
- 服务器发现与连接：启动外部进程或建立网络连接
- 能力同步：获取服务器声明的工具、资源和提示词
- 工具调用的编排：将 AI 的工具调用意图转换为 MCP 请求
- 结果处理与格式化：将服务器返回的结果展示给 AI 或用户
- 生命周期管理：监控连接健康、处理重连和优雅关闭
与服务器端相比，客户端开发面临更多复杂场景：需要同时管理多个服务器的连接，处理跨服务器的工具依赖，以及在大上下文窗口中高效组织工具结果。
2. 客户端初始化
2.1 stdio 客户端
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";
async function createStdioClient() {
const client = new Client(
{
name: "my-ai-app",
version: "1.0.0",
},
{
capabilities: {
tools: {},
resources: {},
prompts: {},
},
}
);
const transport = new StdioClientTransport({
command: "npx",
args: ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/docs"],
env: process.env as Record<string, string>,
});
await client.connect(transport);
// 初始化握手完成后获取服务器能力
const tools = await client.listTools();
const resources = await client.listResources();
return { client, tools, resources };
}
2.2 HTTP 客户端
import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";
async function createHttpClient() {
const client = new Client({ name: "web-app", version: "1.0" });
const transport = new SSEClientTransport(
new URL("http://localhost:3000/sse")
);
await client.connect(transport);
return client;
}
3. 工具调用链
3.1 基本调用流程
async function callToolWithRetry(
client: Client,
toolName: string,
args: Record<string, unknown>,
maxRetries: number = 3
): Promise<string> {
let lastError: Error | null = null;
for (let attempt = 0; attempt < maxRetries; attempt++) {
try {
const result = await client.callTool({
name: toolName,
arguments: args,
});
// 检查工具返回的错误
if (result.isError) {
const errorText = result.content
.filter((c) => c.type === "text")
.map((c) => c.text)
.join("\n");
throw new Error(`工具执行失败: ${errorText}`);
}
return result.content
.filter((c) => c.type === "text")
.map((c) => c.text)
.join("\n");
} catch (error) {
lastError = error instanceof Error ? error : new Error(String(error));
if (attempt < maxRetries - 1) {
const delay = Math.pow(2, attempt) * 1000; // 指数退避
await new Promise((r) => setTimeout(r, delay));
}
}
}
throw lastError ?? new Error("工具调用失败");
}
3.2 多步骤工具链
许多 AI 任务需要顺序调用多个工具，后一个工具的输入依赖前一个工具的输出：
interface ToolChainStep {
tool: string;
args: (prevResults: Record<string, unknown>) => Record<string, unknown>;
}
async function executeToolChain(
client: Client,
steps: ToolChainStep[]
): Promise<Record<string, unknown>> {
const results: Record<string, unknown> = {};
for (let i = 0; i < steps.length; i++) {
const step = steps[i];
const args = step.args(results);
console.log(`步骤 ${i + 1}/${steps.length}: 调用 ${step.tool}`);
const result = await callToolWithRetry(client, step.tool, args);
results[step.tool] = result;
}
return results;
}
// 使用示例：读取文件 -> 分析内容 -> 生成报告
const analysisChain: ToolChainStep[] = [
{
tool: "read_file",
args: () => ({ path: "/path/to/data.json" }),
},
{
tool: "analyze_data",
args: (prev) => ({
data: prev["read_file"],
metrics: ["mean", "median", "stddev"],
}),
},
{
tool: "generate_report",
args: (prev) => ({
analysis: prev["analyze_data"],
format: "markdown",
}),
},
];
4. 多服务器协调
4.1 服务器注册表
在复杂的 AI 应用中，可能需要同时连接数十个 MCP 服务器。设计一个中央注册表来管理它们：
interface ServerConfig {
name: string;
transport: "stdio" | "sse";
command?: string;
args?: string[];
url?: string;
env?: Record<string, string>;
}
class MCPClientRegistry {
private clients: Map<string, Client> = new Map();
private toolsIndex: Map<string, { server: string; tool: Tool }> = new Map();
async addServer(config: ServerConfig): Promise<void> {
const client = new Client(
{ name: "registry-client", version: "1.0" },
{ capabilities: { tools: {}, resources: {} } }
);
let transport;
if (config.transport === "stdio") {
transport = new StdioClientTransport({
command: config.command!,
args: config.args ?? [],
env: config.env ?? {},
});
} else {
transport = new SSEClientTransport(new URL(config.url!));
}
await client.connect(transport);
// 建立工具索引
const tools = await client.listTools();
for (const tool of tools.tools) {
this.toolsIndex.set(tool.name, { server: config.name, tool });
}
this.clients.set(config.name, client);
}
async callTool(toolName: string, args: Record<string, unknown>): Promise<string> {
const entry = this.toolsIndex.get(toolName);
if (!entry) {
throw new Error(`未找到工具: ${toolName}`);
}
const client = this.clients.get(entry.server);
if (!client) {
throw new Error(`服务器未连接: ${entry.server}`);
}
const result = await client.callTool({
name: toolName,
arguments: args,
});
if (result.isError) {
throw new Error(`工具调用失败: ${toolName}`);
}
return result.content
.filter((c) => c.type === "text")
.map((c) => c.text)
.join("\n");
}
getAvailableTools(): string[] {
return Array.from(this.toolsIndex.keys());
}
async closeAll(): Promise<void> {
for (const [name, client] of this.clients) {
await client.close();
console.log(`已关闭服务器连接: ${name}`);
}
this.clients.clear();
this.toolsIndex.clear();
}
}
4.2 工具路由
当多个服务器提供同名工具时，需要路由策略：
class ToolRouter {
private routes: Map<string, string> = new Map(); // toolName -> serverName
private fallbackRules: Array<(toolName: string) => string | null> = [];
register(toolName: string, serverName: string): void {
this.routes.set(toolName, serverName);
}
addFallbackRule(rule: (toolName: string) => string | null): void {
this.fallbackRules.push(rule);
}
resolve(toolName: string): string | null {
if (this.routes.has(toolName)) {
return this.routes.get(toolName)!;
}
for (const rule of this.fallbackRules) {
const result = rule(toolName);
if (result) return result;
}
return null;
}
}
5. 上下文管理
5.1 工具结果截断
大量工具调用结果可能超出模型的上下文窗口限制。需要智能截断策略：
class ResultCompressor {
private maxTokens: number;
constructor(maxTokens: number = 8000) {
this.maxTokens = maxTokens;
}
estimateTokens(text: string): number {
// 粗略估计：英文约 1 token / 4 chars，中文约 1 token / 1.5 chars
let tokens = 0;
for (const char of text) {
tokens += char.charCodeAt(0) > 127 ? 0.7 : 0.3;
}
return Math.ceil(tokens);
}
compress(result: string, priority: "head" | "tail" | "both" = "head"): string {
if (this.estimateTokens(result) <= this.maxTokens) {
return result;
}
const charsBudget = Math.floor(this.maxTokens * 3.5);
if (priority === "head") {
return result.slice(0, charsBudget) + "\n... [内容已截断]";
} else if (priority === "tail") {
return "[内容前部已截断] ...\n" + result.slice(-charsBudget);
} else {
const half = Math.floor(charsBudget / 2);
return (
result.slice(0, half) +
"\n... [中间内容已截断] ...\n" +
result.slice(-half)
);
}
}
}
5.2 结果缓存
对于幂等性工具调用，添加客户端缓存减少重复请求：
class ToolResultCache {
private cache: Map<string, { result: string; expiry: number }> = new Map();
private defaultTTL: number;
constructor(defaultTTLMs: number = 60000) {
this.defaultTTL = defaultTTLMs;
}
private makeKey(toolName: string, args: Record<string, unknown>): string {
return `${toolName}:${JSON.stringify(args)}`;
}
get(toolName: string, args: Record<string, unknown>): string | null {
const key = this.makeKey(toolName, args);
const entry = this.cache.get(key);
if (!entry) return null;
if (Date.now() > entry.expiry) {
this.cache.delete(key);
return null;
}
return entry.result;
}
set(
toolName: string,
args: Record<string, unknown>,
result: string,
ttlMs?: number
): void {
const key = this.makeKey(toolName, args);
this.cache.set(key, {
result,
expiry: Date.now() + (ttlMs ?? this.defaultTTL),
});
}
invalidate(toolName: string): void {
for (const key of this.cache.keys()) {
if (key.startsWith(`${toolName}:`)) {
this.cache.delete(key);
}
}
}
}
6. 与 LLM 集成
6.1 Function Calling 桥接
将 LLM 的 Function Calling 输出桥接到 MCP 工具调用：
interface LLMFunctionCall {
name: string;
arguments: string; // JSON string
}
class MCPFunctionBridge {
constructor(private registry: MCPClientRegistry) {}
async executeLLMFunctionCall(call: LLMFunctionCall): Promise<string> {
const args = JSON.parse(call.arguments);
return this.registry.callTool(call.name, args);
}
// 将 MCP 工具列表转换为 LLM Function Calling 格式
convertToolsToOpenAIFormat(tools: Tool[]): Array<{
type: "function";
function: {
name: string;
description: string;
parameters: object;
};
}> {
return tools.map((tool) => ({
type: "function" as const,
function: {
name: tool.name,
description: tool.description ?? "",
parameters: tool.inputSchema,
},
}));
}
}
6.2 完整对话流程
async function runConversation(
llm: OpenAI,
registry: MCPClientRegistry,
userMessage: string
): Promise<string> {
const messages: Array<{
role: "user" | "assistant" | "function";
content?: string;
name?: string;
}> = [{ role: "user", content: userMessage }];
// 获取可用工具
const tools = await registry.getAllTools();
const openaiTools = registry.convertToOpenAIFormat(tools);
let response = await llm.chat.completions.create({
model: "gpt-4",
messages,
tools: openaiTools,
tool_choice: "auto",
});
// 处理可能的工具调用循环
while (response.choices[0].message.tool_calls) {
const assistantMessage = response.choices[0].message;
messages.push({
role: "assistant",
content: assistantMessage.content ?? "",
});
for (const toolCall of assistantMessage.tool_calls) {
const functionCall = toolCall.function;
console.log(`LLM 请求调用: ${functionCall.name}`);
const result = await registry.callTool(
functionCall.name,
JSON.parse(functionCall.arguments)
);
messages.push({
role: "function" as const,
name: functionCall.name,
content: result,
});
}
response = await llm.chat.completions.create({
model: "gpt-4",
messages,
tools: openaiTools,
});
}
return response.choices[0].message.content ?? "";
}
7. 监控与日志
class MCPClientMetrics {
private callCounts: Map<string, number> = new Map();
private callLatencies: Map<string, number[]> = new Map();
private errors: Map<string, number> = new Map();
recordCall(toolName: string, latencyMs: number, success: boolean): void {
const count = this.callCounts.get(toolName) ?? 0;
this.callCounts.set(toolName, count + 1);
const latencies = this.callLatencies.get(toolName) ?? [];
latencies.push(latencyMs);
this.callLatencies.set(toolName, latencies);
if (!success) {
const errCount = this.errors.get(toolName) ?? 0;
this.errors.set(toolName, errCount + 1);
}
}
getReport(): string {
const lines = ["=== MCP 客户端调用报告 ==="];
for (const [tool, count] of this.callCounts) {
const latencies = this.callLatencies.get(tool) ?? [];
const avgLatency =
latencies.reduce((a, b) => a + b, 0) / latencies.length;
const errCount = this.errors.get(tool) ?? 0;
lines.push(
`${tool}: ${count} 次调用, 平均 ${avgLatency.toFixed(0)}ms, 错误 ${errCount} 次`
);
}
return lines.join("\n");
}
}
7.2 异常检测告警
class MCPAnomalyDetector {
private latencyHistory: Map<string, number[]> = new Map();
private errorRateHistory: Map<string, { total: number; errors: number }> = new Map();
private windowSize: number = 60; // 最近 60 次调用
record(toolName: string, latencyMs: number, success: boolean): void {
// 记录延迟历史
let latencies = this.latencyHistory.get(toolName) ?? [];
latencies.push(latencyMs);
if (latencies.length > this.windowSize) {
latencies = latencies.slice(-this.windowSize);
}
this.latencyHistory.set(toolName, latencies);
// 记录错误率
let stats = this.errorRateHistory.get(toolName) ?? { total: 0, errors: 0 };
stats.total++;
if (!success) stats.errors++;
this.errorRateHistory.set(toolName, stats);
}
// 基于统计的异常检测
checkAnomalies(toolName: string): Array<{ type: string; severity: string; message: string }> {
const alerts: Array<{ type: string; severity: string; message: string }> = [];
// 延迟异常
const latencies = this.latencyHistory.get(toolName) ?? [];
if (latencies.length >= 10) {
const mean = latencies.reduce((a, b) => a + b, 0) / latencies.length;
const recent = latencies.slice(-5);
const recentMean = recent.reduce((a, b) => a + b, 0) / recent.length;
if (recentMean > mean * 3) {
alerts.push({
type: "latency_spike",
severity: "warning",
message: `${toolName} 最近延迟 ${recentMean.toFixed(0)}ms，远高于平均值 ${mean.toFixed(0)}ms`,
});
}
}
// 错误率异常
const stats = this.errorRateHistory.get(toolName) ?? { total: 0, errors: 0 };
if (stats.total >= 20) {
const errorRate = stats.errors / stats.total;
if (errorRate > 0.2) {
alerts.push({
type: "high_error_rate",
severity: "critical",
message: `${toolName} 错误率高达 ${(errorRate * 100).toFixed(1)}%，需要立即排查`,
});
}
}
return alerts;
}
}
7.3 链路追踪
class MCPTracer {
private spans: Map<string, Span> = new Map();
startSpan(operation: string, parentId?: string): string {
const id = crypto.randomUUID();
this.spans.set(id, {
id,
parentId,
operation,
startTime: Date.now(),
tags: {},
});
return id;
}
addTag(spanId: string, key: string, value: string): void {
const span = this.spans.get(spanId);
if (span) span.tags[key] = value;
}
finishSpan(spanId: string, error?: Error): void {
const span = this.spans.get(spanId);
if (!span) return;
span.endTime = Date.now();
span.duration = span.endTime - span.startTime;
span.error = error?.message;
// 发送到追踪系统（如 Jaeger、Zipkin）
this.exportSpan(span);
}
private exportSpan(span: Span): void {
console.log(`[TRACE] ${span.operation} - ${span.duration}ms`, span.tags);
}
}
// 使用
const tracer = new MCPTracer();
async function tracedToolCall(registry: MCPClientRegistry, toolName: string, args: any) {
const spanId = tracer.startSpan(`tool_call:${toolName}`);
try {
tracer.addTag(spanId, "tool", toolName);
const result = await registry.callTool(toolName, args);
tracer.finishSpan(spanId);
return result;
} catch (error) {
tracer.finishSpan(spanId, error as Error);
throw error;
}
}
7.3 分布式追踪集成
// OpenTelemetry 集成
import { trace, context, propagation } from "@opentelemetry/api";
class OpenTelemetryMCPWrapper {
private tracer = trace.getTracer("mcp-client");
async callToolWithTrace(
registry: MCPClientRegistry,
toolName: string,
args: any,
parentContext?: any
): Promise<string> {
const span = this.tracer.startSpan(`mcp.tool.${toolName}`, {
attributes: {
"mcp.tool.name": toolName,
"mcp.tool.args": JSON.stringify(args),
},
}, parentContext);
try {
const result = await context.with(
trace.setSpan(context.active(), span),
() => registry.callTool(toolName, args)
);
span.setStatus({ code: 1 }); // OK
return result;
} catch (error) {
span.recordException(error as Error);
span.setStatus({ code: 2, message: (error as Error).message }); // ERROR
throw error;
} finally {
span.end();
}
}
}
8. 总结
MCP 客户端的开发核心是将 “AI 模型的工具调用意图” 转化为 “服务器的实际执行”。关键要点包括：
- 连接池管理：同时维护多个服务器连接，按需路由工具调用
- 容错设计：重试、退避、优雅降级确保服务可用性
- 上下文控制：结果截断、缓存和摘要避免超出模型上下文限制
- 监控可见性：记录调用频率、延迟和错误率，及时发现异常
- 桥接标准化：将兼容 MCP 的服务器无缝接入任何支持 Function Calling 的模型
MCP 客户端的错误处理模式
MCP 服务器可能随时不可用或返回异常，客户端需实现健壮的错误处理：
// 带退避重试的工具调用
func callWithRetry(ctx context.Context, client *mcp.Client, tool string, args map[string]any, maxRetries int) (*mcp.CallResult, error) {
var lastErr error
for i := 0; i < maxRetries; i++ {
result, err := client.CallTool(ctx, tool, args)
if err == nil {
return result, nil
}
lastErr = err
// 指数退避
time.Sleep(time.Duration(i*i) * 100 * time.Millisecond)
}
return nil, fmt.Errorf("max retries exceeded: %w", lastErr)
}
| 错误类型 | 处理方式 | 示例 |
|---|---|---|
| 网络超时 | 退避重试 | context.DeadlineExceeded |
| 服务未就绪 | 连接池健康检查 | 定期 ping initialize |
| 参数无效 | 向前端返回校验错误 | InvalidParams |
| 工具调用失败 | 降级到默认行为 | 跳过该工具，继续对话 |
MCP 的开放工具生态意味着不可信服务器的风险。客户端应始终校验返回数据的 schema，防止恶意服务器注入伪造结果。
在下一篇文章中，我们将探讨 MCP 的安全实践——如何在开放工具生态的同时确保数据和系统安全。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。