构建 MCP 服务器并通过 Claude API 集成,是让 AI 编程助手拥有自定义工具能力的关键。本文将分步指导你:1) 使用 TypeScript SDK 和 Zod 注册 Tool/Resource/Prompt,并选择合适的传输协议;2) 配置 Claude API 的消息、流式输出及工具调用;3) 将两者结合,构建从单次工具调用到多轮推理的多步 Agent 工作流。内容基于 Everything Claude Code 的 MCP Server Patterns 与 Claude API Skill。
MCP 服务器开发与 Claude API 集成:从 Tool 注册到多步 Agent 工作流
在 AI 辅助编程中,MCP(Model Context Protocol) 服务器是连接 AI 助手与外部工具、数据的桥梁。而 Claude API 则是驱动这些智能体的核心引擎。本文将结合 Everything Claude Code 的 MCP Server Patterns Skill 与 Claude API Skill,为你提供一套从服务器构建、API 集成到实现复杂工作流的完整操作指南。这能帮助你为 Claude Code、Cursor 等 AI 编程助手扩展强大的自定义能力。
第一部分:构建 MCP 服务器:注册 Tool 与传输选择
MCP 服务器的核心是声明 AI 助手可以使用的“能力”,并处理其调用。
1. 环境准备与服务器初始化
首先,安装官方的 MCP TypeScript SDK 和用于参数校验的 Zod 库。
npm install @modelcontextprotocol/sdk zod
随后,初始化一个 MCP 服务器实例。
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
const server = new McpServer({ name: "my-tool-server", version: "1.0.0" });
2. 使用 Zod Schema 注册 Tool
注册 Tool 是最关键的一步。推荐采用 schema first 模式,将参数校验与业务逻辑解耦。
// 注册一个搜索工具
server.tool(
"search_codebase", // 工具名称
"在代码库中搜索指定关键词,并返回相关文件路径和行号。", // 工具描述,AI 据此判断何时调用
// 使用 Zod 定义严格的输入参数 schema
z.object({
query: z.string().describe("要搜索的关键词或正则表达式"),
file_pattern: z.string().optional().describe("可选的文件通配符,例如 '*.ts'"),
}),
// 处理函数,接收经过 Zod 校验后的参数
async ({ query, file_pattern }) => {
// 实际执行搜索逻辑的代码
const results = await performSearch(query, file_pattern);
// 返回结构化结果
return { content: [{ type: "text", text: JSON.stringify(results) }] };
}
);
同样的模式也适用于注册 Resource(读取只读资源,如文件内容)和 Prompt(参数化的提示模板)。
3. 选择传输协议:stdio vs Streamable HTTP
MCP 服务器需要通过特定协议与客户端(如 Claude Desktop)通信。
- stdio 传输:适用于本地单机场景,例如在 Claude Desktop 中直接运行的本地工具服务器。通信通过标准输入输出进行。
import { createStdioTransport } from "@modelcontextprotocol/sdk/transport/stdio"; server.connect(createStdioTransport()); - Streamable HTTP 传输:适用于远程或需要网络访问的场景,例如部署在云端或供多个客户端共享的服务。
import { createHttpTransport } from "@modelcontextprotocol/sdk/transport/http"; server.connect(createHttpTransport({ port: 8080 }));
选择建议:开发调试或为单个本地 IDE 提供工具时,用 stdio。需要部署为服务、支持多客户端或与远程 Claude Code 会话集成时,用 Streamable HTTP。务必查阅最新文档,避免 SDK 版本差异导致的 API 变动。
第二部分:集成 Claude API:消息、工具调用与成本优化
服务器就绪后,需要通过 Claude API 让模型能够理解并调用这些工具。
1. 基础消息与流式输出
使用 Python 或 TypeScript SDK 发起请求,并开启流式输出以改善用户体验。
# Python 示例:基础消息与流式
import anthropic
client = anthropic.Anthropic()
# 流式输出
with client.messages.stream(
model="claude-sonnet-4-0",
max_tokens=1024,
messages=[{"role": "user", "content": "解释一下 MCP 协议"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)
2. 将 MCP Tool 声明为 Claude 的可用工具
为了让 Claude 知道可以调用你 MCP 服务器上的工具,需要在 API 请求中明确定义工具 schema。
tools = [
{
"name": "search_codebase",
"description": "在代码库中搜索指定关键词,并返回相关文件路径和行号。",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "要搜索的关键词或正则表达式"},
"file_pattern": {"type": "string", "description": "可选的文件通配符,例如 '*.ts'"}
},
"required": ["query"]
}
}
# ... 可以在此声明其他工具
]
message = client.messages.create(
model="claude-sonnet-4-0",
max_tokens=1024,
tools=tools,
messages=[{"role": "user", "content": "帮我找到项目里所有使用了 useEffect 的地方"}]
)
3. 成本优化策略
频繁调用 API 会产生显著成本。以下是基于 Everything Claude Code 实践总结的核心优化策略:
| 策略 | 节省潜力 | 适用场景与说明 |
|---|---|---|
| Prompt 缓存 | 最高90% | 对重复的 System Prompt 或大型上下文启用缓存。适用于会话初期、多轮对话间共享基础指令。 |
| Batches API | 50% | 将大批量、非实时任务(如代码注释生成、数据提取)打包异步处理,成本减半。 |
| 选择轻量模型 | ~75% | 对于简单分类、信息抽取等任务,用 claude-3-5-haiku-latest 替代 claude-sonnet-4-0。 |
控制 max_tokens |
可变 | 根据预期输出长度合理设置,避免为简短回答支付长输出费用。 |
实现 Prompt 缓存的代码示例:
message = client.messages.create(
model="claude-sonnet-4-0",
max_tokens=1024,
system=[
{"type": "text", "text": large_system_prompt, "cache_control": {"type": "ephemeral"}}
],
messages=[{"role": "user", "content": "基于上面缓存的上下文提问..."}]
)
# 检查缓存命中情况
print(f"缓存读取 tokens: {message.usage.cache_read_input_tokens}")
第三部分:实现多步 Agent 工作流
结合前两部分,我们可以构建一个能让 Claude 自主调用工具、观察结果并持续推理的 Agent 循环。
import anthropic
import json
client = anthropic.Anthropic()
# 假设 search_codebase 函数已实现,能调用你的 MCP 服务器
from my_mcp_client import search_codebase
tools = [ ... ] # 同上定义
messages = [{"role": "user", "content": "检查 auth 模块的安全性"}]
# 多步 Agent 循环
while True:
response = client.messages.create(
model="claude-sonnet-4-0",
max_tokens=4096,
tools=tools,
messages=messages,
)
# 如果模型决定结束对话,则跳出循环
if response.stop_reason == "end_turn":
final_text = "".join(block.text for block in response.content if hasattr(block, 'text'))
print(f"最终分析:{final_text}")
break
# 处理工具调用
tool_calls = [block for block in response.content if block.type == "tool_use"]
if tool_calls:
# 将模型的回复(包含工具调用)加入历史
messages.append({"role": "assistant", "content": response.content})
# 执行所有工具调用并收集结果
tool_results = []
for tool_call in tool_calls:
result = None
if tool_call.name == "search_codebase":
result = search_codebase(**tool_call.input)
# ... 处理其他工具
tool_results.append({
"type": "tool_result",
"tool_use_id": tool_call.id,
"content": json.dumps(result) if result else "工具执行失败或无结果"
})
# 将工具结果作为用户消息的一部分反馈给模型
messages.append({"role": "user", "content": tool_results})
else:
# 如果模型既没有结束也没有调用工具,可能需要中断或记录日志
print("警告:模型未调用工具也未结束,检查提示词或上下文。")
break
这个工作流实现了“思考-行动-观察”的经典 Agent 模式。Claude 分析任务,调用你通过 MCP 注册的工具(如代码搜索、数据库查询),根据返回的结果进行下一轮推理,直至完成任务。
结合 Everything Claude Code 的 Agent Harness Construction Skill 和 Autonomous Agent Harness Skill,你可以将这种模式工程化,构建包含持久记忆、任务队列和质量门控的全自主 Agent 系统。
FAQ
Q: MCP Server Patterns Skill 与直接使用 MCP SDK 有什么区别? A: Skill 提供了经过验证的注册模式、错误处理模板、传输选择建议和最佳实践,能帮你避免 SDK 版本差异和通信细节上的坑,快速搭建生产级服务器。
Q: 如何在 MCP 服务器中选择 stdio 还是 HTTP 传输? A: 如果你的工具服务器仅在本机为单个客户端(如 Claude Desktop)服务,选择 stdio。如果需要被远程访问、被多个客户端共享或部署在服务器上,则选择 Streamable HTTP。
Q: 降低 Claude API 调用成本最有效的方法是什么? A: 对重复上下文启用 Prompt 缓存潜力最大。其次,对非实时大批量任务使用 Batches API。最后,根据任务复杂度灵活选择模型,简单任务用 Haiku,复杂任务才用 Sonnet 或 Opus。