文章总结： 本文详细介绍了MCP协议的核心概念与Server开发实践，重点阐述其作为标准化AI工具集成协议的解耦设计、三层架构模型和四大原语。内容涵盖MCP与FunctionCalling的互补关系、Client-Host-Server通信流程，以及通过二进制分析MCPServer实例演示工具动态发现与跨进程调用的具体实现方案。
综合评分： 85
文章分类： 技术标准,安全工具,安全开发,解决方案,AI安全
4. MCP协议基础与Server开发
原创
李北辰
李北辰
SPEEDCoding
2026年5月13日 23:25
山西
在小说阅读器读本章
去阅读
完整docx文件关注公众号回复：从零构建AI驱动的二进制安全系统
在第三章中，你已经学会了如何通过Function Calling让Agent调用外部工具，并在命令行中实现了一个具备基础工具能力的Agent。但那种模式有一个根本性局限：工具的定义、注册和执行都发生在同一个代码库中，Agent和工具之间是代码级耦合。当你需要让不同的Agent共享同一套工具，或者想让工具运行在不同的进程甚至不同的机器上时，这种模式就显得力不从心。
这正是MCP（Model Context Protocol，模型上下文协议）要解决的核心问题。MCP将工具从Agent的代码中解耦出来，使其成为独立的、可通过标准协议发现的服务。想象你走进一个图书馆：Function Calling就像是你自带了一箱子书（工具定义嵌在代码里），而MCP则是图书馆本身的检索系统——你只需要知道如何查询，就能找到馆内的任何一本书（动态发现），而且新书不断上架而不需要你重新编译你的程序1^。
在本章中，你将深入理解MCP协议的设计哲学，掌握其核心原语，并最终实现一个二进制分析MCP Server——一个能够暴露反汇编、字符串提取、文件信息分析等能力的独立服务。这个Server将通过stdio模式运行，任何兼容MCP的客户端（包括Claude Desktop、Cursor、或你将在第五章中构建的MCP Client）都可以动态发现并调用它的能力。
4.1 MCP协议概述
4.1.1 MCP是什么：协议定义、发起方与设计目标
MCP（Model Context Protocol，模型上下文协议） 是一个开放的、标准化的应用层协议，用于连接AI模型与外部数据源和工具2^。它由Anthropic于2024年11月首次提出并开源， rapidly evolving into an open industry standard backed by major technology companies including Microsoft、Google、AWS、Cloudflare、Figma和Stripe3^。截至2025年中期，社区已经发布了超过10,000个共享MCP Server4^。这些Server覆盖了从文件系统操作、数据库查询、API集成到专业领域工具（如生物信息学分析、金融数据获取、网络安全扫描）的广泛场景。MCP的快速增长反映了行业对标准化AI工具集成协议的迫切需求——每个团队都不想重复造轮子，而MCP提供了一个”一次编写，到处使用”的解决方案。
MCP的设计目标可以用五个关键词来概括：
标准化（Standardization）。MCP消除了每个AI应用重复编写工具集成代码的需求。在没有MCP之前，如果你想让Agent调用Slack API、查询GitHub Issues、读取本地文件，你需要为每个工具编写自定义的适配层——80%的AI Agent开发时间都花在了这种”管道工程”上5^。MCP通过统一协议让工具只需集成一次，就能被任何兼容客户端使用。这种标准化带来的效益是巨大的：一个团队开发的文件系统MCP Server可以立即被使用Claude Desktop、Cursor IDE或自研Agent的其他团队使用，无需任何代码修改。
解耦（Decoupling）。工具的实现细节（用什么语言编写、运行在哪个进程、使用什么认证凭证）与工具的消费方（Agent或LLM客户端）完全分离。工具提供者发布一个MCP Server，工具消费者通过MCP Client连接——两者通过标准协议通信，互不关心对方的内部实现。这种解耦使得工具可以独立开发、独立部署、独立更新。你的二进制分析团队可以用Python开发分析引擎，而Agent开发团队用TypeScript构建Client——两者通过MCP协议无缝对接。
可插拔（Pluggability）。MCP常被比喻为”AI的USB-C接口”6^。就像USB-C统一了充电和数据传输标准一样，MCP统一了AI工具与模型之间的通信标准。你可以随时插拔新的工具，而不需要修改Agent的代码。当你需要为Agent添加数据库查询能力时，只需让Client连接一个数据库MCP Server即可，Agent本身的代码完全不需要改动。
安全隔离（Security Isolation）。每个MCP Server运行在独立的进程中，凭证隔离在Server级别而非应用级别。这意味着Agent不需要直接访问你的GitHub Token或数据库密码——这些敏感信息只保存在对应的MCP Server中7^。此外，Host层可以实施细粒度的访问控制策略，决定哪些Client可以连接哪些Server，哪些操作需要用户确认。这种多层安全模型使得MCP特别适合企业级部署。
跨模型兼容（Model Agnostic）。MCP不绑定特定的LLM提供商。无论是Claude、GPT、Gemini还是本地运行的Llama，只要客户端实现了MCP协议，就能使用任何MCP Server暴露的工具8^。这种模型无关性保护了你在工具集成上的投资——即使未来更换了底层LLM，所有的MCP Server和工具集成都可以直接复用。
4.1.2 MCP与Function Calling的区别
理解了MCP是什么之后，一个自然的问题是：MCP和第三章学到的Function Calling是什么关系？它们是竞争关系还是互补关系？
答案是：互补关系。MCP和Function Calling位于AI工具调用栈的不同层次，各自解决不同的问题9^。
Function Calling是模型API的特性，属于工具调用的”第一阶段”——意图生成。当LLM决定需要调用某个工具时，它会输出一个包含工具名称和参数的JSON对象。但Function Calling本身并不规定：工具定义存储在哪里、工具由谁来执行、执行环境如何管理、多个工具如何共享。这些都是应用层需要解决的问题。
MCP是应用层协议，属于工具调用的”第二阶段”——标准化执行。它规定了一套完整的通信协议，让工具的发现、调用和执行可以在不同进程、不同语言、不同机器之间标准化进行10^。
以下对比表清晰地展示了两者的核心差异：
| 对比维度 | Function Calling（传统模式） | MCP（Model Context Protocol） |
| — | — | — |
| 协议层级 | 模型API特性（意图生成层） | 应用层协议（执行标准化层） |
| 工具定义位置 | 嵌入每次API请求的tools数组中 | 在独立的MCP Server上定义，运行时动态发现 |
| 工具执行环境 | 应用代码内联执行，与Agent同进程 | MCP Server独立进程执行，进程级隔离 |
| 发现机制 | 手动硬编码，每次请求传递工具列表 | 自动动态发现，Client连接Server后获取工具列表 |
| 可移植性 | 模型特定格式（OpenAI、Anthropic格式不同） | 通用开放标准，跨模型兼容 |
| 状态管理 | 每次请求无状态，需重复传递上下文 | 持久化Server连接，维护会话状态 |
| 部署模式 | 内联代码，与Agent同进程部署 | 独立Server进程，可本地或远程部署 |
| 多客户端共享 | 不支持，需复制代码到每个客户端 | 原生支持，任何MCP客户端可连接 |
| 凭证隔离 | 应用级（all-or-nothing） | Server级（最小权限原则） |
| 语言限制 | 与Agent同语言 | 跨语言互操作，Server可用任何语言实现 |
| 生态规模 | 每个应用自定义 | 10,000+共享服务器 |
从架构角度看，Function Calling的完整循环是：
应用代码 → 定义工具Schema → 发送给LLM → LLM决定调用 →
应用执行函数 → 结果返回LLM → LLM生成最终响应
在这个循环中，所有环节——工具Schema定义、执行逻辑、错误处理、凭证管理——都发生在同一个应用进程中。当你的Agent需要调用50个不同的工具时，这些工具的代码、依赖和凭证都会堆积在你的主应用中。
MCP的Client-Server模型则完全不同：
AI应用(Host) → MCP Client → 连接到MCP Server →
Server暴露tools/resources/prompts → Client动态发现 →
LLM决定调用 → Client发送请求 → Server执行 → 结果返回
工具在独立进程中运行。AI应用通过MCP Client连接到各个Server，动态发现可用能力。当LLM决定调用某个工具时，Client将请求转发给对应的Server执行，执行结果再返回给LLM。
那么什么时候用Function Calling，什么时候用MCP？一个实用的判断标准是：
- 使用Function Calling：工具数量少于5个、单一模型和单一应用、延迟敏感路径（每毫秒都重要）、快速原型开发阶段。如果你正在构建一个个人使用的天气查询Agent，只有两个工具（获取天气、搜索城市），Function Calling是最简单直接的选择。
- 使用MCP：工具数量超过10个、多团队共享工具、多模型架构、需要生产级治理（认证、审计日志、访问控制）、构建工具生态11^。当你正在构建一个企业的AI助手平台，需要连接GitHub、Jira、Slack、内部数据库、文件系统等十几种工具时，MCP的解耦和标准化优势就会充分体现。
在实际项目中，推荐采用混合架构：对应用特有的、不需要共享的工具使用Function Calling内联实现；对通用的、多客户端共享的基础设施工具使用MCP Server暴露。例如，一个Agent可能通过Function Calling直接调用内部的业务逻辑API，同时通过MCP Server连接文件系统操作工具、代码搜索工具和外部API。这种分层方法在保持简单性的同时获得了MCP生态的复用优势。
4.1.3 MCP架构模型：Client-Host-Server三层架构
MCP采用Client-Host-Server三层架构模型，这是理解MCP通信流程的基础12^。
Host（宿主）是MCP架构中的顶层容器。它通常是你直接与之交互的AI应用程序——比如Claude Desktop、Cursor IDE、或你自己开发的AI应用。Host的职责包括：创建和管理多个Client实例、控制Client的连接权限和生命周期、强制执行安全策略和同意要求、处理用户授权决策、协调LLM集成和上下文聚合13^。
Client（客户端）由Host创建，每个Client维护与单个Server之间的隔离连接。Client的核心职责是：为每个Server建立有状态的会话、处理协议协商和能力交换、双向路由协议消息、管理订阅和通知、维护Server之间的安全边界。关键点：Host应用创建和管理多个Client，每个Client与特定Server保持1:1关系14^。
Server（服务端）通过MCP原语向客户端暴露资源、工具和提示模板。每个Server独立运行，职责聚焦，可以是本地进程（通过stdio通信）或远程服务（通过HTTP通信）。Server还可以向Client请求采样（Sampling），即在需要时”使用”LLM的能力15^。
MCP的会话生命周期分为三个阶段：
初始化阶段（Initialization）。Client发送initialize请求，包含支持的协议版本（如2025-03-26）和声明的能力（如sampling、roots）。Server响应匹配的协议版本和自身能力（如prompts、tools、resources）。双方完成能力协商后，Client发送initialized通知确认就绪。只有在初始化完成后，有意义的操作才能开始16^。
初始化阶段的设计体现了MCP协议的前向兼容性哲学。协议版本遵循YYYY-MM-DD的日期格式，每次规范更新都会修改版本号。当Client和Server的版本不一致时，双方选择都支持的最新版本进行通信。这意味着2025年6月的Client可以连接2024年11月的Server，只要两者在核心协议上兼容。能力协商（Capability Negotiation）则进一步细化了兼容性——Client声明自己支持的功能（如能否处理Server发起的采样请求），Server声明自己提供的功能（如暴露了哪些原语），双方只在交集范围内操作。
操作阶段（Operation）。Client和Server根据协商的能力交换请求、通知和响应。Client可以调用tools/list发现Server的工具列表，调用tools/call执行工具，调用resources/read读取资源，调用prompts/get获取提示模板。如果协商了采样能力，Server也可以主动发起sampling/createMessage请求来使用LLM17^。
关闭阶段（Shutdown）。Client或Server可以随时关闭底层传输连接，无需特定的协议消息。建议实现超时机制和健壮的错误处理，确保资源正确释放。
4.1.4 MCP核心原语：Resources、Tools、Prompts、Sampling
MCP定义了四个核心原语（Primitives），它们是Server向LLM客户端暴露能力的统一接口18^。理解这些原语的性质和用途，是设计MCP Server的关键。
Tools（工具）——主动执行原语。Tools是LLM可以调用的函数或服务，用于执行外部操作。它们的性质是主动的（Active）：模型决定何时调用。类比来说，Tools就像公司中的服务工作者——快递员、客服代理、支付系统。当你需要发送包裹时，你主动呼叫快递员；当你需要查询天气时，Agent主动调用天气API。
每个Tool包含三个要素：唯一标识的名称、描述输入参数的JSON Schema、执行实际逻辑的处理函数。Tool的调用数据流是：Client发送请求 → Server执行 → 返回结果。2025年3月的规范新增了Tool注解（Annotations），允许标注工具的行为特征，如readOnlyHint（只读操作）、destructiveHint（破坏性操作）、idempotentHint（幂等）、openWorldHint（可能访问外部世界）19^。这些注解帮助LLM更智能地选择工具。
Resources（资源）——被动读取原语。Resources是AI模型可以访问的只读数据源，用于获取上下文信息。它们的性质是被动的（Passive）：模型可以读取但不需要触发函数调用。类比来说，Resources就像图书馆中的书籍——你可以随时翻阅获取信息。
Resource通过URI寻址（如file:///config.json、db://users/123、weather://cities），Server在注册时声明支持的URI模式和MIME类型。Client通过resources/read方法读取资源内容20^。
Prompts（提示模板）——标准化交互原语。Prompts是可复用的交互模板，Server可以预定义标准化的交互方式。它们的性质是模板化的（Template）：参数化的消息序列。类比来说，Prompts就像客户服务脚本或标准化操作流程——它们确保每次交互的一致性和质量。
Prompts的独特之处在于，Server既了解数据（Resources的内容），也了解让模型处理数据的最佳方式（Prompt的设计）。因此，Prompts可以组合Resources和Tools来创建动态工作流21^。
Sampling（采样）——LLM能力委托原语。Sampling是MCP中最独特的原语。它允许Server通过Client请求LLM进行推理——换句话说，Server可以在需要时”使用”LLM的能力。工作机制是：Server发送sampling/createMessage请求 → Client转发给LLM → LLM响应 → Client返回结果22^。
Sampling的安全性设计非常重要：Host可以审核、修改或拒绝任何采样请求，控制使用的模型和token限制。这使得Server能够在安全约束下利用LLM的能力进行复杂推理，比如递归分析Agent工作流或工具链中的LLM调用23^。
这四个原语的组合创造了强大的动态工作流能力。例如，一个日志分析Server可以这样工作：首先暴露一个分析日志的Prompt模板，当用户选择该Prompt后，Server调用Sampling让LLM分析日志内容，LLM返回”日志显示重复认证失败，建议检查OAuth配置”，然后Server调用Toolcheck_auth_config来验证OAuth配置，最后LLM审查验证结果并生成总结24^。这种Prompt驱动的Agentic工作流是MCP区别于简单工具调用的核心特征——Server不仅是被动的工具提供者，更是能够主动编排LLM能力的工作流引擎。
理解这四个原语的关系，可以用一个类比：想象你是一位厨师（LLM），Resources是你冰箱里的食材（只读数据），Tools是你的厨具（可执行操作），Prompts是菜谱（标准化流程），Sampling是你在烹饪过程中品尝味道并根据反馈调整的能力。这四个要素的协同使得AI Agent能够处理远比简单Q&A复杂的任务。
4.2 MCP传输层与通信
MCP协议建立在JSON-RPC 2.0之上，传输层负责承载这些JSON-RPC消息的传递。MCP当前定义了三种标准传输机制：stdio（标准输入输出）、HTTP/SSE（已弃用）和Streamable HTTP（2025年3月新增的推荐方式）。理解每种传输模式的工作原理和适用场景，对选择合适的部署方案至关重要。
4.2.1 stdio传输模式：标准输入输出的JSON-RPC通信
stdio传输模式是MCP中最基础、最常用的传输方式，尤其适用于本地集成的场景25^。
在这种模式下，MCP Client将Server作为一个子进程启动，Server从标准输入（stdin）读取JSON-RPC消息，处理后将响应写入标准输出（stdout）。所有MCP协议消息以换行符（\n）分隔，不包含嵌入式换行符。Server的日志信息应该写入标准错误（stderr），而非stdout——因为stdout必须保持纯净，只包含MCP协议消息26^。
通信流程如下：
Server STDOUTMCP ServerServer STDINMCP ClientServer STDOUTMCP ServerServer STDINMCP Client{"jsonrpc":"2.0","id":1,"method":"initialize",...}读取JSON-RPC请求处理initialize请求{"jsonrpc":"2.0","id":1,"result":{...}}读取JSON-RPC响应{"jsonrpc":"2.0","id":2,"method":"tools/list"}读取请求查询工具列表{"jsonrpc":"2.0","id":2,"result":{"tools":[...]}}读取响应{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{...}}读取工具调用请求执行工具逻辑{"jsonrpc":"2.0","id":3,"result":{"content":[...]}}读取执行结果
stdio模式具有以下特点：
- 本地性：Client和Server在同一台机器上运行，Server作为Client的子进程启动。
- 低延迟：无网络栈开销，纯进程间通信，延迟通常在微秒级别。
- 简单性：无需网络配置、端口管理或防火墙规则，直接通过操作系统管道通信。
- 一对一：每个Client进程对应一个Server进程，通信是点对点的。
- 安全性：无网络暴露面，通信不经过任何网络接口，天然适合安全敏感操作27^。
stdio模式的典型适用场景包括：本地文件系统操作（文件读写、目录浏览）、本地数据库查询、CLI工具集成、IDE扩展（如VS Code插件）、安全敏感操作（密钥管理、加密解密）。在实际使用中，stdio模式是最常见也是最容易上手的方式——Claude Desktop、Cursor、Windsurf等主流AI应用都优先使用stdio模式连接本地MCP Server27^。
stdio模式的一个关键注意事项是stdout的独占性。由于MCP协议消息通过stdout传递，任何写入stdout的非协议内容都会破坏通信。这意味着你必须避免在Server代码中使用console.log()——所有日志和调试信息应该通过console.error()写入stderr。这是一个常见的新手错误，会导致Client端解析JSON失败、通信中断等诡异问题。
以下是一个stdio传输模式下JSON-RPC消息流的实际示例。当Client调用tools/list时，消息流如下：
// Client → Server (STDIN)
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}
// Server → Client (STDOUT)
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "get_weather",
"description": "Get current weather for a specified city",
"inputSchema": {
"type": "object",
"properties": {
"city": { "type": "string", "description": "City name" }
},
"required": ["city"]
}
},
{
"name": "read_file",
"description": "Read contents of a file",
"inputSchema": {
"type": "object",
"properties": {
"path": { "type": "string", "description": "File path" }
},
"required": ["path"]
}
}
]
}
}
代码示例1：stdio传输模式的基础Server实现
// stdio-server.ts —— 最基础的stdio模式MCP Server
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
// 创建Server实例，声明名称、版本和能力
const server = new Server(
{ name: "example-stdio-server", version: "1.0.0" },
{ capabilities: { tools: {} } }
);
// 注册tools/list处理器：返回Server暴露的所有工具
server.setRequestHandler("tools/list", async () => {
return {
tools: [
{
name: "echo",
description: "Echo back the input message",
inputSchema: {
type: "object",
properties: {
message: { type: "string", description: "Message to echo" }
},
required: ["message"]
}
}
]
};
});
// 注册tools/call处理器：执行工具逻辑
server.setRequestHandler("tools/call", async (request) => {
if (request.params.name === "echo") {
const { message } = request.params.arguments;
return {
content: [{ type: "text", text: message }]
};
}
throw new Error(`Unknown tool: ${request.params.name}`);
});
// 使用StdioServerTransport连接Server
const transport = new StdioServerTransport();
await server.connect(transport);
// 重要：日志必须写入stderr，stdout专用于MCP消息
console.error("stdio MCP Server running on stdin/stdout");
在这个示例中，StdioServerTransport封装了stdio通信的所有细节——它从process.stdin读取JSON-RPC消息，解析后交给Server处理，然后将响应序列化后写入process.stdout。你只需要关注业务逻辑的实现，不需要手动处理字节流的读写。
4.2.2 HTTP/SSE传输模式：服务端推送的流式通信
在Streamable HTTP出现之前，MCP远程传输的标准方式是HTTP + SSE（Server-Sent Events）组合28^。理解这种已被弃用的传输方式仍然有意义，因为许多现有的MCP Server实现和文档仍然基于它，而且理解它的问题有助于理解为什么Streamable HTTP是更好的方案。
HTTP/SSE传输模式使用两个端点：
- SSE端点（通常是
/sse）：Client通过GET请求打开SSE连接，Server通过这个连接推送Server-to-Client的消息（如通知、采样请求）。 - 消息端点（通常是
/messages）：Client通过POST请求发送Client-to-Server的消息（如工具调用、资源读取）。
MCP Server消息端点 (/messages)SSE端点 (/sse)MCP ClientMCP Server消息端点 (/messages)SSE端点 (/sse)MCP ClientGET /sse (建立SSE连接)event: endpoint\ndata: /messages?session_id=xxxPOST /messages?session_id=xxx\n{"method":"initialize",...}转发请求通过SSE推送响应{"jsonrpc":"2.0","id":1,"result":{...}}POST /messages\n{"method":"tools/list"}推送响应{"jsonrpc":"2.0","id":2,"result":{...}}
代码示例2：HTTP/SSE传输模式的Server实现
// sse-server.ts —— HTTP+SSE传输模式（已弃用，仅作参考）
import express from "express";
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
const app = express();
app.use(express.json());
const server = new Server(
{ name: "example-sse-server", version: "1.0.0" },
{ capabilities: { tools: {} } }
);
// 工具注册...
server.setRequestHandler("tools/list", () => ({
tools: [{ name: "get-time", description: "Get current time", inputSchema: { type: "object", properties: {} } }]
}));
// SSE端点：Client通过GET建立SSE连接
app.get("/sse", async (req, res) => {
res.setHeader("Content-Type", "text/event-stream");
res.setHeader("Cache-Control", "no-cache");
res.setHeader("Connection", "keep-alive");
// 发送消息端点URL
res.write(`event: endpoint\ndata: /messages?session_id=${generateSessionId()}\n\n`);
// 后续Server响应通过此连接推送
// ...
});
// 消息端点：Client通过POST发送请求
app.post("/messages", async (req, res) => {
const message = req.body;
// 处理JSON-RPC消息
const response = await server.handleRequest(message);
res.json(response);
});
app.listen(3000, () => console.log("SSE MCP Server on http://localhost:3000"));
HTTP/SSE模式存在以下根本性问题29^：
双端点复杂性。需要维护两个独立的HTTP端点（/sse和/messages），Client需要先连接SSE端点获取消息端点URL，才能开始通信。这增加了初始化的复杂度和出错的可能性。
Serverless不兼容。SSE长连接在Serverless平台（Cloudflare Workers、Vercel Edge Functions、AWS Lambda）上无法正常工作，因为这些平台的设计假设是请求-响应模式的短期执行30^。
负载均衡问题。SSE连接需要粘性会话（Sticky Session），即同一个Client的请求必须路由到同一个Server实例。这在负载均衡环境下增加了配置复杂性。
连接恢复困难。SSE连接断开后，上下文丢失，Client需要重新初始化整个会话，包括能力协商和订阅管理。
4.2.3 Streamable HTTP传输：MCP 2025年3月新增的推荐传输方式
Streamable HTTP是2025年3月26日MCP规范重大更新中引入的新的标准远程传输机制，它在2025年6月18日的规范中完全取代了HTTP/SSE31^。
Streamable HTTP的核心设计理念是简化：所有通信通过一个HTTP端点完成，不再需要双端点架构。
工作原理：
- 单一端点：所有通信通过一个HTTP端点（通常是
/mcp）完成。 - POST请求：Client发送JSON-RPC消息通过POST请求。
- 动态响应：Server根据请求类型选择返回
application/json（单响应）或text/event-stream（流式响应）。 - 可选GET：Client可通过GET请求打开SSE流来接收Server主动推送的消息（如通知）。
- 会话管理：通过
Mcp-Session-IdHTTP Header管理会话状态。 - 可恢复性：通过SSE事件ID支持断点恢复32^。
MCP Server/mcp (单一端点)MCP ClientMCP Server/mcp (单一端点)MCP ClientServer通过Mcp-Session-Id返回会话IDPOST /mcp\n{"method":"initialize",...}处理请求HTTP 200 application/json{"jsonrpc":"2.0","id":1,"result":{...}}POST /mcp\nMcp-Session-Id: xxx\n{"method":"tools/list"}HTTP 200 application/json{"tools":[...]}POST /mcp\n{"method":"tools/call",...}HTTP 200 text/event-stream (流式响应)SSE流推送结果GET /mcp\nMcp-Session-Id: xxx (可选SSE流)SSE流：Server主动推送通知
代码示例3：Streamable HTTP传输模式的Server实现
// streamable-http-server.ts —— Streamable HTTP传输模式（推荐）
import express from "express";
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StreamableHTTPServerTransport } from
"@modelcontextprotocol/sdk/server/streamableHttp.js";
const app = express();
app.use(express.json());
const server = new Server(
{ name: "example-http-server", version: "1.0.0" },
{ capabilities: { tools: {}, resources: {} } }
);
// 注册工具
server.setRequestHandler("tools/list", () => ({
tools: [
{
name: "get-time",
description: "Get current UTC time",
inputSchema: { type: "object", properties: {} }
}
]
}));
server.setRequestHandler("tools/call", (request) => {
if (request.params.name === "get-time") {
return {
content: [{ type: "text", text: new Date().toISOString() }]
};
}
throw new Error("Unknown tool");
});
// 单一/mcp端点处理所有MCP通信
app.post("/mcp", async (req, res) => {
// 创建StreamableHTTPTransport实例
const transport = new StreamableHTTPServerTransport({
sessionIdGenerator: undefined, // 启用会话管理
enableJsonResponse: true // 允许JSON响应（非SSE）
});
// 连接关闭时清理transport
res.on("close", () => transport.close());
// 连接Server和Transport
await server.connect(transport);
// 处理请求：transport自动决定返回JSON或SSE
await transport.handleRequest(req, res, req.body);
});
// 可选：GET端点支持SSE流式推送
app.get("/mcp", async (req, res) => {
const transport = new StreamableHTTPServerTransport({
sessionIdGenerator: undefined,
});
res.on("close", () => transport.close());
await server.connect(transport);
await transport.handleRequest(req, res);
});
app.listen(3000, () => console.log("MCP Server on http://localhost:3000/mcp"));
Streamable HTTP解决了HTTP/SSE模式的所有根本性问题：单一端点简化了架构，兼容Serverless平台（因为POST请求可以是无状态的），原生支持负载均衡（会话通过Header管理而非连接状态），并且支持断点恢复33^。
TypeScript SDK在v1.10.0（2025年4月17日）首次支持Streamable HTTP，这是目前构建远程MCP Server的推荐方式34^。
从实现角度看，Streamable HTTP的关键优势在于其无状态设计。在传统的HTTP/SSE模式中，Server需要维护SSE连接的状态（哪些Client订阅了哪些通知），这使得水平扩展变得困难——你需要使用Redis或类似的共享存储来同步订阅状态，或者使用粘性会话（Sticky Session）将同一Client的请求路由到同一Server实例。Streamable HTTP通过将通知推送拆分为独立的GET请求解决了这个问题：Client可以选择性地通过GET /mcp打开一个SSE流来接收Server主动推送的消息，而常规的POST请求保持完全无状态。这种设计使得每个POST请求都可以被任意Server实例处理，天然兼容负载均衡和Serverless架构。
Streamable HTTP的认证模型也更加成熟。它支持OAuth 2.1（包括PKCE流程）、Bearer Token和API Key三种认证方式32^。这种标准化的认证机制使得MCP Server可以集成到现有的企业身份基础设施中，通过OAuth Provider控制访问权限。对于需要多租户支持的场景（如SaaS化的分析服务），Streamable HTTP是stdio模式无法比拟的选择。
4.2.4 传输层选型指南
选择合适的传输模式取决于你的具体部署场景。下表提供了不同传输方式的详细对比和选择建议：
| 评估因素 | stdio | HTTP/SSE（已弃用） | Streamable HTTP（推荐） |
| — | — | — | — |
| 部署模型 | 本地子进程 | Web服务 | Web服务/Serverless |
| 通信延迟 | 极低（进程管道，微秒级） | 中等（HTTP RTT） | 中等（HTTP RTT） |
| 并发客户端 | 1（一对一） | 多 | 多 |
| 认证机制 | 环境变量、文件权限 | OAuth、API Key | OAuth 2.1、Bearer Token、API Key |
| 负载均衡 | 不适用 | 需粘性会话 | 原生支持，无状态设计 |
| Serverless兼容 | 否 | 否 | 是（Cloudflare Workers、Vercel Edge） |
| 安全边界 | 进程隔离 | 网络隔离+认证 | 网络隔离+认证 |
| 连接恢复 | 进程重启 | 需重新初始化 | SSE事件ID支持断点恢复 |
| 设置复杂度 | 低（无需配置） | 中（双端点配置） | 中（单一端点） |
| 推荐用途 | 本地CLI工具、IDE扩展 | 不推荐（已弃用） | 远程部署、生产环境 |
选型决策流程：
首先判断Server的部署位置。如果是本地部署（与Client在同一台机器上），stdio模式几乎总是最佳选择。它简单、低延迟、安全，且无需网络配置。Claude Desktop、Cursor IDE等本地AI应用都主要使用stdio模式连接本地MCP Server。stdio模式的一个额外优势是自动化的进程生命周期管理——当Client关闭时，Server子进程自动终止，无需显式的关闭协议或心跳检测35^。
如果你的Server需要支持多Client并发访问（比如团队共享的分析服务），stdio模式不再适用，因为它是一对一的进程关系。此时应该选择Streamable HTTP，它可以自然地处理多个并发的HTTP请求。Streamable HTTP的无状态设计还意味着你可以使用标准的负载均衡器（如Nginx、AWS ALB）来分发请求，而无需配置粘性会话。
如果Server需要远程部署（Client通过网络连接），Streamable HTTP是推荐的唯一选择。它在所有维度上都优于已弃用的HTTP/SSE模式，特别是在Serverless和负载均衡场景下。
对于二进制分析这个特定场景，stdio模式是合理的起点——分析脚本通常运行在本地机器上，通过stdio与MCP Client通信既简单又高效。如果你的团队需要多人共享分析服务，或者分析任务需要运行在远程服务器上，可以考虑将Server迁移到Streamable HTTP模式。
一个值得注意的实际考量是：在stdio模式下，Python子进程的生命周期完全由Node.js的spawn管理。这意味着每次工具调用都会启动一个新的Python进程，执行完毕后自动退出。虽然进程创建有一定开销（通常在几十毫秒级别），但对于分析工具这种执行时间通常在秒级别的场景来说，这点开销可以忽略不计。更重要的是，这种”用完即走”的模型避免了内存泄漏和状态污染问题——每个分析任务都在一个干净的进程中执行，即使Python脚本因为异常数据而崩溃，也不会影响MCP Server主进程或其他并行的分析任务35^。
4.3 MCP Server开发入门
本节将带领你从零开始构建一个MCP Server。你将学习如何安装SDK、创建项目骨架、注册工具、暴露资源和定义提示模板。所有示例都使用@modelcontextprotocol/sdk的TypeScript SDK，这是MCP官方维护的SDK，API设计简洁且与协议规范保持同步。
4.3.1 @modelcontextprotocol/sdk安装与项目初始化
MCP TypeScript SDK是官方提供的标准开发工具包。截至2025年，SDK的最新稳定版本是v1.x系列，v2正在积极开发中36^。本教程使用v1.x的API风格，这是目前生产环境的推荐选择。
首先创建项目目录并初始化：
# 创建项目目录
mkdir binary-analysis-mcp-server && cd binary-analysis-mcp-server
# 初始化npm项目
npm init -y
# 安装MCP SDK和依赖
npm install @modelcontextprotocol/sdk zod
# 安装TypeScript开发依赖
npm install -D typescript @types/node
代码示例4：完整的package.json和tsconfig.json配置
// package.json
{
"name": "binary-analysis-mcp-server",
"version": "1.0.0",
"description": "MCP Server for binary analysis - disassembly, strings, file info",
"type": "module",
"main": "dist/index.js",
"bin": {
"binary-analysis-mcp-server": "dist/index.js"
},
"scripts": {
"build": "tsc",
"dev": "tsc --watch",
"start": "node dist/index.js",
"prepare": "npm run build"
},
"keywords": ["mcp", "binary-analysis", "disassembly", "security"],
"author": "",
"license": "MIT",
"dependencies": {
"@modelcontextprotocol/sdk": "^1.10.0",
"zod": "^3.23.0"
},
"devDependencies": {
"@types/node": "^20.0.0",
"typescript": "^5.5.0"
},
"engines": {
"node": ">=18.0.0"
}
}
// tsconfig.json
{
"compilerOptions": {
"target": "ES2022",
"module": "Node16",
"moduleResolution": "Node16",
"outDir": "./dist",
"rootDir": "./src",
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true,
"declaration": true,
"declarationMap": true,
"sourceMap": true
},
"include": ["src/**/*"],
"exclude": ["node_modules", "dist"]
}
关键配置说明：
"type": "module"：启用ES Module支持，这是MCP SDK的要求。"module": "Node16"和"moduleResolution": "Node16"：Node16模块解析策略，正确处理.js扩展名的导入（即使源文件是.ts）。"bin"字段：定义命令行入口，安装后可通过binary-analysis-mcp-server命令启动Server。"engines": { "node": ">=18.0.0" }：Node.js 18+支持原生fetch API和稳定的ESM加载器。
项目目录结构建议如下：
binary-analysis-mcp-server/
├── package.json # 项目配置
├── tsconfig.json # TypeScript配置
├── README.md # 项目说明
├── src/
│ ├── index.ts # 入口文件：Server启动
│ ├── tools/ # 工具实现
│ │ ├── disassemble.ts # 反汇编工具
│ │ ├── strings.ts # 字符串提取工具
│ │ └── file-info.ts # 文件信息工具
│ ├── resources/ # 资源实现
│ │ └── binary-file.ts # 二进制文件资源
│ ├── prompts/ # 提示模板
│ │ └── analyze-binary.ts # 分析提示模板
│ └── bridge/ # Python桥接
│ └── python-runner.ts # Python脚本执行器
└── dist/ # 编译输出（自动生成）
4.3.2 Server基础骨架：创建MCP Server的Hello World示例
MCP SDK提供了两个主要的Server类：Server和McpServer。Server是底层类，需要手动注册每个协议方法的处理函数；McpServer是高级封装，提供了tool()、resource()、prompt()等便捷方法来注册原语37^。对于新开发的项目，推荐使用McpServer——它的API更直观，代码更简洁。
代码示例5：MCP Server Hello World（stdio模式）
// src/index.ts —— MCP Server Hello World
#!/usr/bin/env node
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
// ====== 步骤1：创建McpServer实例 ======
const server = new McpServer({
name: "hello-mcp-server",
version: "1.0.0"
});
// ====== 步骤2：注册第一个Tool ======
// server.tool() 方法的参数：
// 1. name: 工具名称（唯一标识符，LLM通过名称调用）
// 2. config: 工具配置对象（包含描述、输入Schema、注解）
// 3. handler: 异步处理函数（执行实际逻辑）
server.tool(
"hello",
{
description: "Say hello to someone. Use this tool when the user wants a greeting.",
inputSchema: z.object({
name: z.string()
.describe("The name of the person to greet, e.g. 'Alice', 'Bob'")
}),
annotations: {
readOnlyHint: true, // 只读操作，不修改任何状态
destructiveHint: false, // 非破坏性操作
idempotentHint: true, // 幂等：多次调用结果相同
}
},
async ({ name }) => {
// 返回工具执行结果
// content数组支持text、image、audio、video等多种内容类型
return {
content: [
{ type: "text", text: `Hello, ${name}! Welcome to MCP Server development.` }
]
};
}
);
// ====== 步骤3：启动Server ======
async function main() {
// StdioServerTransport处理所有stdio通信细节
const transport = new StdioServerTransport();
// 连接Server到Transport
await server.connect(transport);
// 日志写入stderr（重要：stdout专用于MCP消息）
console.error("Hello MCP Server running on stdio");
}
// 错误处理
main().catch((error) => {
console.error("Fatal error:", error);
process.exit(1);
});
让我们逐行解析这段代码：
#!/usr/bin/env node 是Shebang行，使文件可以直接执行（chmod +x后运行./dist/index.js）。
McpServer是SDK的高级封装类，构造函数接收一个配置对象，包含Server的name和version。这些信息在初始化阶段会被发送给Client，用于识别Server身份。
server.tool()是注册工具的核心方法。它接收三个参数：工具名称（字符串）、工具配置对象（包含description、inputSchema、annotations）、处理函数（接收解析后的参数，返回执行结果）。
inputSchema使用Zod定义参数结构。Zod是一个TypeScript优先的Schema验证库，它的优势在于：类型安全（Schema即类型）、运行时验证、自动类型推断、简洁的API。每个字段通过.describe()添加描述，这些描述会被传递给LLM，帮助它正确理解参数的用途38^。
annotations是2025年3月规范新增的特性。它们不是强制的，但强烈建议添加，因为它们帮助LLM更智能地决策。readOnlyHint: true告诉LLM这个工具不会修改任何状态，在决策时可以放心调用；destructiveHint: false表示操作是安全的，不会删除或损坏数据；idempotentHint: true表示多次调用结果相同，LLM在不确定时可以安全地重试；openWorldHint: false表示工具操作是自包含的，不需要访问外部网络或服务。这些元数据看似微小，但在复杂的Agent工作流中，它们显著提高了LLM工具选择的准确性39^。
返回值的content数组支持多种内容类型：text（文本）、image（图片，需base64编码）、audio（音频）、video（视频）。对于二进制分析工具，text类型通常足够，因为它可以承载JSON序列化的分析结果。但在某些场景下，你可能需要返回图片——例如，如果你实现了控制流图生成功能，可以将图作为base64编码的PNG返回，让LLM直接”看到”函数调用关系。音频和视频类型的应用场景目前较少，但在多媒体分析Server中可能会有用武之地。
编译并测试这个Hello World Server：
# 编译TypeScript
npx tsc
# 使用MCP Inspector测试（先安装Inspector）
npx @anthropics/mcp-inspector node dist/index.js
4.3.3 Tool的定义与注册：使用server.tool()方法
工具（Tool）是MCP Server最常用和最重要的原语。设计良好的工具需要兼顾三个维度：准确的描述（让LLM知道何时调用）、严格的参数验证（防止错误输入）、健壮的执行逻辑（处理各种边界情况）。
代码示例6：工具定义与注册的完整示例
// src/tools/calculator.ts —— 工具定义的最佳实践示例
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
export function registerCalculatorTools(server: McpServer) {
// ====== Tool 1: 基础计算器 ======
server.tool(
"calculate",
{
// 描述至关重要：说明"什么时候使用"，而非仅仅"做什么"
description: `Evaluate a mathematical expression.
Use this tool for ALL mathematical calculations, unit conversions,
and numeric computations. Supports +, -, *, /, ^, parentheses,
and common functions like sqrt, sin, cos, log.
Always use this instead of trying to calculate manually.`,
inputSchema: z.object({
expression: z.string()
.describe("The mathematical expression to evaluate, e.g. '2 + 3 * 4', 'sqrt(256)'")
.min(1, "Expression cannot be empty")
.max(1000, "Expression too long"),
precision: z.number()
.int()
.min(0)
.max(10)
.optional()
.describe("Number of decimal places in result (0-10, default: 2)")
}),
annotations: {
readOnlyHint: true,
destructiveHint: false,
idempotentHint: true,
openWorldHint: false, // 不需要访问外部世界
}
},
async ({ expression, precision = 2 }) => {
try {
// 输入验证：只允许安全的数学表达式
const sanitized = sanitizeExpression(expression);
// 使用Function构造器安全求值（实际项目中应使用math.js等库）
const result = evaluateMath(sanitized);
return {
content: [{
type: "text",
text: JSON.stringify({
expression: expression,
result: Number(result.toFixed(precision)),
precision: precision
}, null, 2)
}]
};
} catch (error) {
// 错误处理：返回有意义的错误信息
return {
content: [{
type: "text",
text: JSON.stringify({
error: "Calculation failed",
expression: expression,
message: error instanceof Error ? error.message : "Unknown error",
suggestion: "Check expression syntax. Supported: +, -, *, /, ^, sqrt(), sin(), cos(), log()"
}, null, 2)
}],
isError: true // 标记为错误结果
};
}
}
);
// ====== Tool 2: 单位转换器（展示枚举参数） ======
server.tool(
"convert_unit",
{
description: "Convert between common units of measurement. " +
"Use for length, weight, temperature, and data size conversions.",
inputSchema: z.object({
value: z.number().describe("The numeric value to convert"),
from: z.enum(["meters", "feet", "kilograms", "pounds", "celsius", "fahrenheit",
"bytes", "kilobytes", "megabytes", "gigabytes"])
.describe("Source unit"),
to: z.enum(["meters", "feet", "kilograms", "pounds", "celsius", "fahrenheit",
"bytes", "kilobytes", "megabytes", "gigabytes"])
.describe("Target unit")
}),
annotations: {
readOnlyHint: true,
idempotentHint: true,
openWorldHint: false,
}
},
async ({ value, from, to }) => {
try {
const result = performConversion(value, from, to);
return {
content: [{
type: "text",
text: `${value} ${from} = ${result} ${to}`
}]
};
} catch (error) {
return {
content: [{
type: "text",
text: `Conversion error: ${error instanceof Error ? error.message : "Unknown"}`
}],
isError: true
};
}
}
);
}
// 辅助函数：清理数学表达式（安全验证）
function sanitizeExpression(expr: string): string {
// 只允许数字、运算符、括号、小数点和已知函数
const allowed = /^[\d\+\-\*\/\^\(\)\.\s\,]+$/;
const allowedFunctions = /(sqrt|sin|cos|tan|log|ln|abs|round|floor|ceil)/g;
// 移除潜在危险的内容
const cleaned = expr
.replace(/;/g, "") // 移除语句分隔符
.replace(/{/g, "") // 移除花括号
.replace(/}/g, "")
.replace(/import/gi, "") // 移除import关键字
.replace(/require/gi, ""); // 移除require关键字
return cleaned;
}
// 辅助函数：安全求值
function evaluateMath(expr: string): number {
// 实际项目中应使用math.js等专业数学库
// 这里使用简化的实现
const normalized = expr
.replace(/\^/g, "**")
.replace(/sqrt\(/g, "Math.sqrt(")
.replace(/sin\(/g, "Math.sin(")
.replace(/cos\(/g, "Math.cos(")
.replace(/log\(/g, "Math.log10(");
// 使用Function构造器而非eval，限制作用域
const fn = new Function(`return (${normalized})`);
return fn();
}
// 辅助函数：单位转换
function performConversion(value: number, from: string, to: string): number {
const conversions: Record<string, Record<string, number>> = {
meters: { feet: 3.28084 },
feet: { meters: 0.3048 },
kilograms: { pounds: 2.20462 },
pounds: { kilograms: 0.453592 },
bytes: { kilobytes: 1 / 1024, megabytes: 1 / 1024 / 1024, gigabytes: 1 / 1024 / 1024 / 1024 },
kilobytes: { bytes: 1024, megabytes: 1 / 1024, gigabytes: 1 / 1024 / 1024 },
megabytes: { bytes: 1024 * 1024, kilobytes: 1024, gigabytes: 1 / 1024 },
gigabytes: { bytes: 1024 * 1024 * 1024, kilobytes: 1024 * 1024, megabytes: 1024 },
};
// 温度转换需要特殊处理
if (from === "celsius" && to === "fahrenheit") return value * 9 / 5 + 32;
if (from === "fahrenheit" && to === "celsius") return (value - 32) * 5 / 9;
if (from === to) return value;
const rate = conversions[from]?.[to];
if (!rate) throw new Error(`Cannot convert from ${from} to ${to}`);
return value * rate;
}
这个示例展示了工具定义的几个最佳实践：
描述质量决定工具调用准确性。calculate工具的description不仅说明了”做什么”（Evaluate a mathematical expression），更重要的是说明了”什么时候使用”（Use this tool for ALL mathematical calculations…）和”为什么用它”（Always use this instead of trying to calculate manually）。高质量的描述直接提高LLM选择正确工具的概率39^。
Zod Schema定义参数的约束和语义。expression参数通过.min()和.max()限制长度，precision通过.int().min(0).max(10)限制取值范围，from和to通过z.enum()限制为预定义的选项。这些约束在运行时自动验证，无效的输入会在到达处理函数之前被拒绝。
错误处理返回有意义的反馈。当工具执行失败时，不是简单地抛出异常，而是返回一个结构化的错误对象，包含error、message和suggestion字段。这样LLM可以根据错误信息调整参数重试，而不是向用户暴露晦涩的技术错误40^。
isError标记区分正常结果和错误结果。返回对象中的isError: true告诉Client这次调用虽然技术上成功了（没有抛出异常），但业务逻辑上产生了错误。这有助于LLM区分空结果和错误结果。
4.3.4 Resource的定义与暴露：使用server.resource()暴露可读数据资源
Resource是MCP中用于暴露只读数据的原语。与Tool不同，Resource不需要LLM触发函数调用——Client可以直接读取Resource的内容，将其作为上下文注入到对话中41^。
Resource通过URI模式寻址。当你注册一个Resource时，你声明一个URI模板（如binary://{file_id}/content），Client可以通过填充模板变量来读取具体的资源。
代码示例7：Resource的定义与暴露
// src/resources/binary-file.ts —— 二进制文件资源暴露
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { readFileSync, statSync } from "fs";
import { resolve } from "path";
// 内存中的文件注册表（实际项目中应使用数据库）
const registeredFiles = new Map<string, {
path: string;
name: string;
size: number;
uploadedAt: Date;
}>();
export function registerBinaryResources(server: McpServer) {
// ====== Resource 1: 二进制文件内容 ======
server.resource(
"binary-content", // Resource名称（内部标识）
"binary://{file_id}/content", // URI模板
{
description: "Raw binary content of an uploaded file. " +
"Use this resource to access the bytes of a binary file for analysis.",
mimeType: "application/octet-stream"
},
async (uri, { file_id }) => {
// file_id从URI模板中自动解析
const fileInfo = registeredFiles.get(file_id);
if (!fileInfo) {
return {
contents: [{
uri: uri.href,
text: JSON.stringify({ error: `File not found: ${file_id}` }),
mimeType: "application/json"
}]
};
}
try {
// 读取二进制文件（以base64编码返回）
const buffer = readFileSync(fileInfo.path);
const base64 = buffer.toString("base64");
return {
contents: [{
uri: uri.href,
text: JSON.stringify({
file_id,
name: fileInfo.name,
size: fileInfo.size,
sha256: "", // 实际项目中计算SHA256
base64: base64.substring(0, 10000) // 限制返回大小
}),
mimeType: "application/json"
}]
};
} catch (error) {
return {
contents: [{
uri: uri.href,
text: JSON.stringify({
error: "Failed to read file",
message: error instanceof Error ? error.message : "Unknown error"
}),
mimeType: "application/json"
}]
};
}
}
);
// ====== Resource 2: 二进制文件元数据 ======
server.resource(
"binary-metadata",
"binary://{file_id}/metadata",
{
description: "Metadata about an uploaded binary file including " +
"file type, architecture, size, and upload time.",
mimeType: "application/json"
},
async (uri, { file_id }) => {
const fileInfo = registeredFiles.get(file_id);
if (!fileInfo) {
return {
contents: [{
uri: uri.href,
text: JSON.stringify({ error: `File not found: ${file_id}` }),
mimeType: "application/json"
}]
};
}
// 获取文件类型信息（简化实现）
const fileType = detectFileType(fileInfo.path);
return {
contents: [{
uri: uri.href,
text: JSON.stringify({
file_id,
name: fileInfo.name,
size: fileInfo.size,
file_type: fileType.type,
architecture: fileType.arch,
bitness: fileType.bitness,
uploaded_at: fileInfo.uploadedAt.toISOString(),
mime_type: "application/octet-stream"
}, null, 2),
mimeType: "application/json"
}]
};
}
);
// ====== Resource 3: 文件列表索引 ======
server.resource(
"binary-index",
"binary://index",
{
description: "List of all uploaded binary files available for analysis.",
mimeType: "application/json"
},
async (uri) => {
const files = Array.from(registeredFiles.entries()).map(
([id, info]) => ({
file_id: id,
name: info.name,
size: info.size,
uploaded_at: info.uploadedAt.toISOString()
})
);
return {
contents: [{
uri: uri.href,
text: JSON.stringify({ files, count: files.length }, null, 2),
mimeType: "application/json"
}]
};
}
);
}
// 辅助函数：检测文件类型
function detectFileType(filePath: string): { type: string; arch: string; bitness: number } {
try {
const buffer = readFileSync(filePath);
// PE文件检测
if (buffer[0] === 0x4D && buffer[1] === 0x5A) { // 'MZ'
const peOffset = buffer.readUInt32LE(0x3C);
if (buffer[peOffset] === 0x50 && buffer[peOffset + 1] === 0x45) {
const machine = buffer.readUInt16LE(peOffset + 4);
if (machine === 0x14c) return { type: "PE", arch: "x86", bitness: 32 };
if (machine === 0x8664) return { type: "PE", arch: "x64", bitness: 64 };
return { type: "PE", arch: "unknown", bitness: 0 };
}
}
// ELF文件检测
if (buffer[0] === 0x7F && buffer[1] === 0x45 && buffer[2] === 0x4C && buffer[3] === 0x46) {
const ei_class = buffer[4]; // 32/64-bit
const ei_machine = buffer.readUInt16LE(18);
const bitness = ei_class === 1 ? 32 : 64;
if (ei_machine === 0x03) return { type: "ELF", arch: "x86", bitness };
if (ei_machine === 0x3E) return { type: "ELF", arch: "x64", bitness };
if (ei_machine === 0x28) return { type: "ELF", arch: "ARM", bitness };
return { type: "ELF", arch: "unknown", bitness };
}
return { type: "Raw", arch: "unknown", bitness: 0 };
} catch {
return { type: "Unknown", arch: "unknown", bitness: 0 };
}
}
// 辅助函数：注册新文件
export function registerFile(fileId: string, filePath: string, name: string): void {
const stats = statSync(filePath);
registeredFiles.set(fileId, {
path: filePath,
name,
size: stats.size,
uploadedAt: new Date()
});
}
Resource注册与Tool注册的关键区别在于：
Resource是被动暴露的，Tool是主动调用的。Client通过resources/read方法直接读取Resource，不需要LLM生成工具调用请求。这使得Resource适合作为上下文的直接来源——比如文件内容、配置文件、知识库文档。在二进制分析场景中，Resource的用途是为LLM提供原始二进制内容（base64编码），让LLM能够”看到”文件的原始字节，这在某些分析场景中非常有用。
Resource使用URI模板寻址。binary://{file_id}/content中的{file_id}是模板变量，当Client请求binary://abc123/content时，SDK自动解析出file_id = "abc123"并传递给处理函数。URI模板可以包含多个变量，如analysis://{file_id}/{analysis_type}/result，使得Resource的组织方式非常灵活。
Resource可以返回多种MIME类型。二进制内容返回application/octet-stream，JSON元数据返回application/json，文本内容返回text/plain。正确的MIME类型帮助Client正确处理Resource内容。在二进制分析Server中，即使是二进制原始内容，我们也返回application/json（包含base64编码的数据），因为纯二进制内容无法直接嵌入JSON42^。
Resource与Tool的一个关键区别体现在LLM的调用流程中。当LLM需要使用某个Tool时，它会主动生成一个tool_call请求；而当LLM需要Resource的信息时，Client会自动在对话上下文中注入Resource内容，LLM不需要显式请求。这种被动注入机制使得Resource特别适合作为”始终可用”的上下文来源。
4.3.5 Prompt模板定义：使用server.prompt()创建可复用的提示模板
Prompts是MCP中最具创造性的原语。它们允许Server预定义标准化的交互模板，确保Agent处理特定任务时的一致性。Prompts的独特价值在于：Server了解数据（Resources的内容），也了解让模型处理数据的最佳方式（Prompt的设计），因此可以创建高质量的预置工作流43^。
代码示例8：Prompt模板的定义与使用
// src/prompts/analyze-binary.ts —— 二进制分析提示模板
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { z } from "zod";
export function registerAnalysisPrompts(server: McpServer) {
// ====== Prompt 1: 二进制安全分析 ======
server.prompt(
"security-analysis",
{
description: "Perform a comprehensive security analysis on a binary file. " +
"Use this prompt when asked to analyze a binary for vulnerabilities, " +
"security risks, or suspicious patterns.",
argsSchema: {
file_path: z.string().describe("Absolute path to the binary file to analyze"),
file_type: z.enum(["PE", "ELF", "Mach-O", "Raw"])
.optional()
.describe("Known file type, if available"),
focus_areas: z.array(z.enum([
"buffer_overflow", "format_string", "integer_overflow",
"use_after_free", "command_injection", "crypto_weaknesses",
"information_disclosure", "all"
])).optional().describe("Specific vulnerability types to focus on")
}
},
async ({ file_path, file_type = "Raw", focus_areas = ["all"] }) => {
// 构建系统级分析指令
const systemPrompt = `You are an expert binary security analyst with deep expertise in:
- Reverse engineering and disassembly analysis
- Vulnerability research and exploit development
- Memory corruption vulnerability detection
- Dangerous API usage pattern recognition
- Cryptographic weakness identification
Analyze the provided binary methodically. For each finding, provide:
1. Vulnerability type with CWE ID
2. Severity level (critical/high/medium/low/info)
3. Confidence score (0.0-1.0)
4. Specific memory address or function name
5. Detailed technical description
6. Actionable remediation advice`;
// 构建用户级分析请求
const focusText = focus_areas.includes("all")
? "all vulnerability categories"
: focus_areas.join(", ");
const userPrompt = `Please perform a comprehensive security analysis on the following binary file:
**File Information:**
- Path: ${file_path}
- File Type: ${file_type}
- Focus Areas: ${focusText}
**Analysis Instructions:**
1. First, examine the binary structure and identify key sections
2. Analyze imported functions for dangerous API usage (strcpy, sprintf, gets, system, etc.)
3. Extract and analyze strings for suspicious patterns (URLs, registry keys, passwords)
4. Review disassembly for memory safety issues (buffer overflows, format strings)
5. Check for cryptographic weaknesses (hardcoded keys, weak algorithms)
6. Assess overall risk score (0-100) and provide prioritized recommendations
**Output Format:**
Return findings as a structured JSON object with the following schema:
{
"findings": [
{
"type": "vulnerability_category",
"severity": "critical|high|medium|low|info",
"confidence": 0.0-1.0,
"address": "0x... or function_name",
"description": "detailed explanation",
"recommendation": "how to fix",
"cwe_id": "CWE-XXX"
}
],
"risk_score": 0-100,
"summary": "executive summary"
}`;
// 返回Prompt模板：消息序列
return {
messages: [
{
role: "user", // Prompt可以包含system和user消息
content: {
type: "text",
text: `${systemPrompt}\n\n${userPrompt}`
}
}
],
// 可选：描述此Prompt的用途
description: `Security analysis prompt for ${file_type} binary at ${file_path}`
};
}
);
// ====== Prompt 2: 快速文件概览 ======
server.prompt(
"file-overview",
{
description: "Get a quick overview of a binary file's structure and characteristics. " +
"Use this for initial triage when first examining an unknown binary.",
argsSchema: {
file_path: z.string().describe("Path to the binary file"),
detail_level: z.enum(["brief", "standard", "detailed"])
.optional()
.describe("Level of detail in the overview")
}
},
async ({ file_path, detail_level = "standard" }) => {
const detailInstructions = {
brief: "Provide a 3-5 sentence summary focusing only on the most important characteristics.",
standard: "Provide a structured overview covering: file type, architecture, size, key sections, imports, and notable strings.",
detailed: "Provide an exhaustive analysis including: file header details, all sections with entropy, complete import table, string analysis, and packing detection."
};
return {
messages: [
{
role: "user",
content: {
type: "text",
text: `Analyze the binary file at \`${file_path}\` and provide an overview.
${detailInstructions[detail_level]}
Use the available tools to gather information:
1. Get file metadata (type, architecture, size)
2. Extract strings (ASCII and Unicode)
3. List imported functions
4. Check for packing/encryption (entropy analysis)
Format the output as a structured report with clear sections.`
}
}
]
};
}
);
// ====== Prompt 3: 反汇编代码审查 ======
server.prompt(
"disassembly-review",
{
description: "Review disassembled code for a specific function or address range. " +
"Use this when asked to examine specific code patterns or function behavior.",
argsSchema: {
file_path: z.string().describe("Path to the binary file"),
target: z.string().describe("Function name or address to examine, e.g. 'sub_401000' or '0x401000'"),
context_lines: z.number().int().min(5).max(50).optional()
.describe("Number of surrounding instructions to include for context")
}
},
async ({ file_path, target, context_lines = 20 }) => {
return {
messages: [
{
role: "user",
content: {
type: "text",
text: `Review the disassembled code for target \`${target}\` in the binary at \`${file_path}\`.
Please:
1. Disassemble the target function/address with ${context_lines} instructions of context
2. Analyze the control flow (branches, loops, function calls)
3. Identify the function's purpose and behavior
4. Flag any suspicious patterns:
- Unusual API calls (e.g., CreateRemoteThread, VirtualAllocEx)
- String manipulation without bounds checking
- Dynamic code execution (eval, exec, shellcode)
- Anti-debugging or anti-analysis techniques
- Network communication patterns
5. Suggest a descriptive name for the function based on its behavior
Format the response with:
- A summary of the function's purpose
- Key observations (security-relevant)
- Control flow description
- Recommended function name
- Risk assessment (low/medium/high)`
}
}
]
};
}
);
}
Prompt模板的设计体现了MCP的核心价值：Server既了解数据，也了解让模型处理数据的最佳方式44^。在security-analysisPrompt中，我们不仅提供了文件路径这个参数，还预定义了完整的分析框架——从检查顺序到输出格式，确保每次安全分析都遵循一致的方法论。
messages数组支持多个消息对象，每个消息有role（可以是user、assistant或system）和content。这使得Prompt可以构建复杂的多轮对话模板，甚至可以包含模拟的助手回复来引导LLM的输出风格45^。
Prompt返回后，Client会将messages中的内容注入到与LLM的对话上下文中。LLM看到这些消息后，会理解它需要执行的任务，并可能调用Server暴露的Tools来获取必要的数据。这种Prompt + Tool的组合形成了一个强大的动态工作流。
4.3.6 Server开发最佳实践与常见陷阱
在实际开发MCP Server的过程中，有一些最佳实践和常见陷阱需要特别注意。
stdout与stderr的严格分离。这是stdio模式下最常见的错误来源。MCP协议要求stdout必须只包含有效的JSON-RPC消息。任何额外的输出——console.log的调试信息、第三方库的warning、未捕获的Promise异常输出——都会破坏JSON-RPC消息流，导致Client解析失败40^。正确的做法是：所有日志和调试信息使用console.error()（写入stderr）；在正式发布前移除所有console.log调用；如果依赖的第三方库会打印到stdout，在初始化阶段将其重定向到stderr或完全禁用。
Tool描述的编写技巧。Tool的description直接影响LLM调用工具的准确性。一个好的描述应该回答三个问题：这个工具做什么（功能描述）、什么时候使用它（触发条件）、需要什么参数（输入说明）。避免模糊描述如”处理文件”，应该使用”Extract ASCII and Unicode strings from a binary file. Use this to find URLs, API names, embedded passwords, and debug information in executable files.”
Schema约束的重要性。Zod Schema不仅提供类型安全，还提供了运行时验证。为每个参数添加.min()、.max()、.email()等约束，可以在无效输入到达你的业务逻辑之前就拒绝它。这减少了错误处理的复杂度，也为LLM提供了参数取值范围的明确提示38^。
错误处理的黄金法则。Tool处理函数中的错误不应该抛出未捕获异常——未捕获异常会导致整个Server进程崩溃（在stdio模式下尤其致命）。正确的模式是try-catch所有业务逻辑，在catch块中返回isError: true的响应，包含人类可读的错误信息和修复建议。如果错误是暂时性的（如网络超时），可以在错误消息中建议LLM重试。
Server命名规范。MCP社区遵循{功能域}-{实现语言}-server或{功能域}-mcp-server的命名约定。例如filesystem-node-server、github-mcp-server。遵循命名规范有助于用户发现和理解你的Server。Server的name字段应该是小写字母和连字符的组合，不包含空格或特殊字符。
能力声明的精确性。在McpServer构造函数或初始化阶段，只声明你实际支持的能力。如果你只暴露了Tools而没有Resources或Prompts，确保能力声明中只包含tools: {}。不精确的能力声明会导致Client尝试调用不支持的方法，产生不必要的错误往返。
4.4 二进制分析MCP Server实战
经过前三节的学习，你已经掌握了MCP协议的核心概念和Server开发的基础技能。现在，让我们将这些知识应用到实际场景中：构建一个完整的二进制分析MCP Server。这个Server将暴露三个核心分析工具——反汇编、字符串提取和文件信息分析——并通过Node.js MCP Server桥接底层的Python分析脚本。
为什么使用Python做底层分析而不用纯TypeScript实现？原因在于二进制分析生态：Capstone反汇编引擎、pefile PE文件解析器、LIEF二进制分析库等核心工具都是Python生态的成熟项目46^。通过Node.js MCP Server作为前端协议层、Python脚本作为后端分析引擎的架构，我们既能享受TypeScript SDK的标准化协议支持，又能复用Python生态丰富的二进制分析工具。
4.4.1 分析能力封装：将Python二进制分析脚本封装为MCP工具
首先，我们需要编写Python分析脚本。这些脚本将作为独立的命令行工具运行，通过JSON输入输出与Node.js层通信。
代码示例9：Python二进制分析脚本
#!/usr/bin/env python3
"""
binary_analyzer.py —— 二进制分析Python脚本
通过stdin接收JSON请求，stdout输出JSON结果
支持：反汇编、字符串提取、文件信息分析
"""
import sys
import json
import struct
import re
import math
from pathlib import Path
from typing import List, Dict, Any, Optional, Tuple
def analyze_file_info(file_path: str) -> Dict[str, Any]:
"""分析二进制文件的基础信息"""
path = Path(file_path)
if not path.exists():
return {"error": f"File not found: {file_path}"}
data = path.read_bytes()
file_size = len(data)
# 文件类型检测
file_type, arch, bitness = detect_file_type(data)
# 计算文件哈希
import hashlib
sha256 = hashlib.sha256(data).hexdigest()
md5 = hashlib.md5(data).hexdigest()
# 计算整体熵值
entropy = shannon_entropy(data)
# 提取文件头信息
header_info = extract_header_info(data, file_type)
return {
"file_path": str(path.resolve()),
"file_name": path.name,
"file_size": file_size,
"file_type": file_type,
"architecture": arch,
"bitness": bitness,
"sha256": sha256,
"md5": md5,
"entropy": round(entropy, 4),
"header": header_info,
"suspicious": {
"high_entropy": entropy > 7.0, # 可能加壳或加密
"very_high_entropy": entropy > 7.5,
}
}
def detect_file_type(data: bytes) -> Tuple[str, str, int]:
"""检测文件类型和架构"""
if len(data) < 4:
return ("Unknown", "unknown", 0)
# PE文件检测
if data[:2] == b'MZ':
try:
pe_offset = struct.unpack('<I', data[0x3C:0x40])[0]
if pe_offset + 24 <= len(data) and data[pe_offset:pe_offset+4] == b'PE\x00\x00':
machine = struct.unpack('<H', data[pe_offset+4:pe_offset+6])[0]
arch_map = {0x14c: ("x86", 32), 0x8664: ("x64", 64),
0x1c0: ("ARM", 32), 0xaa64: ("ARM64", 64)}
arch, bits = arch_map.get(machine, ("unknown", 0))
return ("PE", arch, bits)
except (struct.error, IndexError):
pass
return ("PE", "unknown", 0)
# ELF文件检测
if data[:4] == b'\x7fELF':
ei_class = data[4]
if ei_class == 1:
bits = 32
elif ei_class == 2:
bits = 64
else:
bits = 0
if len(data) >= 20:
ei_machine = struct.unpack('<H', data[18:20])[0]
arch_map = {0x03: "x86", 0x3E: "x64", 0x28: "ARM",
0xB7: "ARM64", 0x08: "MIPS", 0x14: "PowerPC"}
arch = arch_map.get(ei_machine, "unknown")
return ("ELF", arch, bits)
return ("ELF", "unknown", bits)
# Mach-O文件检测
magic = struct.unpack('<I', data[:4])[0]
if magic in (0xfeedface, 0xcefaedfe, 0xfeedfacf, 0xcffaedfe):
is_64 = magic in (0xfeedfacf, 0xcffaedfe)
if len(data) >= 8:
cputype = struct.unpack('<i', data[4:8])[0]
arch_map = {0x07: "x86", 0x01000007: "x64",
0x0C: "ARM", 0x0100000C: "ARM64"}
arch = arch_map.get(cputype, "unknown")
return ("Mach-O", arch, 64 if is_64 else 32)
return ("Mach-O", "unknown", 64 if is_64 else 32)
return ("Raw", "unknown", 0)
def extract_header_info(data: bytes, file_type: str) -> Dict[str, Any]:
"""提取文件头详细信息"""
header = {}
if file_type == "PE" and len(data) >= 64:
try:
pe_offset = struct.unpack('<I', data[0x3C:0x40])[0]
header["pe_header_offset"] = pe_offset
header["timestamp"] = struct.unpack('<I', data[pe_offset+8:pe_offset+12])[0]
header["number_of_sections"] = struct.unpack('<H', data[pe_offset+6:pe_offset+8])[0]
# 可选头大小
header["optional_header_size"] = struct.unpack('<H', data[pe_offset+20:pe_offset+22])[0]
# 特征标志
header["characteristics"] = hex(struct.unpack('<H', data[pe_offset+22:pe_offset+24])[0])
except (struct.error, IndexError):
header["parse_error"] = "Failed to parse PE header"
elif file_type == "ELF" and len(data) >= 52:
try:
header["entry_point"] = hex(struct.unpack('<I', data[24:28])[0])
header["program_header_offset"] = struct.unpack('<I', data[28:32])[0]
header["section_header_offset"] = struct.unpack('<I', data[32:36])[0]
header["flags"] = hex(struct.unpack('<I', data[36:40])[0])
header["program_header_entry_size"] = struct.unpack('<H', data[42:44])[0]
header["number_of_program_headers"] = struct.unpack('<H', data[44:46])[0]
except struct.error:
header["parse_error"] = "Failed to parse ELF header"
return header
def disassemble_file(file_path: str, max_instructions: int = 500) -> Dict[str, Any]:
"""反汇编二进制文件"""
path = Path(file_path)
if not path.exists():
return {"error": f"File not found: {file_path}"}
data = path.read_bytes()
file_type, arch, _ = detect_file_type(data)
try:
import capstone
arch_map = {
("PE", "x86"): (capstone.CS_ARCH_X86, capstone.CS_MODE_32),
("PE", "x64"): (capstone.CS_ARCH_X86, capstone.CS_MODE_64),
("ELF", "x86"): (capstone.CS_ARCH_X86, capstone.CS_MODE_32),
("ELF", "x64"): (capstone.CS_ARCH_X86, capstone.CS_MODE_64),
("ELF", "ARM"): (capstone.CS_ARCH_ARM, capstone.CS_MODE_ARM),
}
cs_arch, cs_mode = arch_map.get((file_type, arch),
(capstone.CS_ARCH_X86, capstone.CS_MODE_64))
md = capstone.Cs(cs_arch, cs_mode)
md.detail = True
# 确定代码段范围（简化：从entry point开始）
code_offset = 0
base_address = 0x1000
# 对于PE文件，尝试找到代码段入口
if file_type == "PE":
try:
pe_offset = struct.unpack('<I', data[0x3C:0x40])[0]
code_offset = struct.unpack('<I', data[pe_offset+40:pe_offset+44])[0] - base_address
code_offset = max(0, code_offset)
except (struct.error, IndexError):
code_offset = 0
# 限制分析范围（避免分析整个文件）
code_data = data[code_offset:code_offset + 65536]
instructions = []
for i, insn in enumerate(md.disasm(code_data, base_address + code_offset)):
if i >= max_instructions:
break
instructions.append({
"address": f"0x{insn.address:08x}",
"bytes": insn.bytes.hex(),
"mnemonic": insn.mnemonic,
"operands": insn.op_str,
"size": insn.size
})
return {
"file": str(path.name),
"architecture": arch,
"file_type": file_type,
"instructions_analyzed": len(instructions),
"max_instructions": max_instructions,
"instructions": instructions
}
except ImportError:
return {
"error": "Capstone engine not installed. Install with: pip install capstone",
"fallback": "Basic byte analysis only",
"architecture": arch,
"file_type": file_type,
"first_64_bytes": data[:64].hex()
}
def extract_strings(file_path: str, min_length: int = 4,
encoding: str = "all") -> Dict[str, Any]:
"""从二进制文件中提取字符串"""
path = Path(file_path)
if not path.exists():
return {"error": f"File not found: {file_path}"}
data = path.read_bytes()
strings_result = []
# ASCII字符串提取
if encoding in ("all", "ascii"):
ascii_pattern = rb'[\x20-\x7e]{' + str(min_length).encode() + rb',}'
for match in re.finditer(ascii_pattern, data):
strings_result.append({
"address": f"0x{match.start():08x}",
"string": match.group().decode('ascii', errors='ignore'),
"type": "ascii",
"length": len(match.group())
})
# Unicode (UTF-16LE)字符串提取
if encoding in ("all", "unicode"):
unicode_pattern = rb'(?:[\x20-\x7e]\x00){' + str(min_length).encode() + rb',}'
for match in re.finditer(unicode_pattern, data):
try:
decoded = match.group().decode('utf-16le', errors='ignore')
if len(decoded) >= min_length:
strings_result.append({
"address": f"0x{match.start():08x}",
"string": decoded,
"type": "unicode",
"length": len(decoded)
})
except (UnicodeDecodeError, UnicodeError):
pass
# 按地址排序
strings_result.sort(key=lambda x: int(x["address"], 16))
# 分类统计
suspicious = filter_suspicious_strings(strings_result)
return {
"file": str(path.name),
"total_strings": len(strings_result),
"ascii_count": sum(1 for s in strings_result if s["type"] == "ascii"),
"unicode_count": sum(1 for s in strings_result if s["type"] == "unicode"),
"suspicious_count": len(suspicious),
"suspicious_strings": suspicious[:50], # 限制返回数量
"strings": strings_result[:200] # 限制返回数量
}
def filter_suspicious_strings(strings: List[Dict]) -> List[Dict]:
"""过滤可疑字符串"""
patterns = [
r'https?://[^\s]+', # URLs
r'[\w.-]+@[\w.-]+\.\w+', # Email addresses
r'HKEY_[\w\\]+', # Registry keys
r'(?:cmd\.exe|powershell|bash|sh\s)',
r'(?:CreateRemoteThread|VirtualAlloc|WriteProcessMemory)',
r'(?:password|passwd|secret|key|token|credential)',
r'(?:WSAStartup|socket|connect|send|recv)',
r'(?:Crypt|BCrypt|NCrypt)',
]
compiled = [re.compile(p, re.IGNORECASE) for p in patterns]
suspicious = []
for s in strings:
text = s["string"]
for pattern in compiled:
if pattern.search(text):
s["match_reason"] = pattern.pattern[:50]
suspicious.append(s)
break
return suspicious
def shannon_entropy(data: bytes) -> float:
"""计算香农熵"""
if not data:
return 0.0
entropy = 0.0
for x in range(256):
p_x = data.count(x) / len(data)
if p_x > 0:
entropy += -p_x * math.log2(p_x)
return entropy
def main():
"""主入口：通过stdin接收JSON请求，stdout输出JSON结果"""
try:
request = json.load(sys.stdin)
command = request.get("command")
params = request.get("params", {})
handlers = {
"file_info": analyze_file_info,
"disassemble": lambda p: disassemble_file(
p["file_path"], p.get("max_instructions", 500)
),
"extract_strings": lambda p: extract_strings(
p["file_path"], p.get("min_length", 4), p.get("encoding", "all")
),
}
handler = handlers.get(command)
if not handler:
result = {"error": f"Unknown command: {command}",
"available_commands": list(handlers.keys())}
else:
result = handler(params)
# 输出JSON结果到stdout
print(json.dumps(result, indent=2))
sys.stdout.flush()
except json.JSONDecodeError as e:
print(json.dumps({"error": f"Invalid JSON input: {str(e)}"}, indent=2))
sys.stdout.flush()
except Exception as e:
print(json.dumps({"error": f"Internal error: {str(e)}"}, indent=2))
sys.stdout.flush()
if __name__ == "__main__":
main()
在深入Python代码之前，理解二进制文件格式的基本结构有助于理解分析逻辑。PE（Portable Executable）是Windows平台的可执行文件格式，其文件头以MZ魔数开头，在偏移量0x3C处有一个指针指向PE头（以PE\x00\x00标识）。PE头中包含机器类型（x86为0x14c，x64为0x8664）、时间戳、代码入口点地址等关键信息。ELF（Executable and Linkable Format）是Linux/Unix平台的可执行文件格式，以\x7fELF魔数开头，第5字节（EI_CLASS）指示32位（值为1）还是64位（值为2），第18-19字节指示目标架构（x86为0x03，x64为0x3E）。Mach-O是macOS/iOS平台的可执行文件格式，以小端或大端编码的0xfeedface（32位）或0xfeedfacf（64位）魔数开头。
熵值（Entropy）是信息论中的概念，用于衡量数据的随机性。对于二进制文件，熵值的范围是0到8（每个字节8位时的最大熵值）。正常的编译代码熵值通常在5-7之间，而加壳或加密的代码段熵值接近8（因为加密后的数据看起来像是完全随机的）。因此，熵值分析是检测加壳二进制文件的有效手段——如果一个PE文件的某个段熵值超过7.0，就值得怀疑它是否经过了加壳或加密处理。
这个Python脚本的设计遵循了几个关键原则：
JSON-RPC over stdio：脚本通过sys.stdin读取JSON请求，解析command和params字段，调用对应的处理函数，然后将结果以JSON格式写入sys.stdout。这种设计使得脚本可以被任何支持JSON交互的父进程调用，不限于Node.js。实际上，你可以手动测试这个脚本：echo '{"command":"file_info","params":{"file_path":"/bin/ls"}}' | python3 binary_analyzer.py，它会直接在终端输出JSON结果。
模块化命令处理：handlers字典将命令名称映射到处理函数。添加新的分析能力只需要新增处理函数并在字典中注册即可。比如，如果你想添加PE导入表分析功能，只需要新增一个analyze_imports函数并在handlers中添加"analyze_imports": analyze_imports即可。
防御性编程：每个函数在开始分析前检查文件是否存在，try-except块捕获所有潜在异常并以结构化JSON错误响应，避免脚本崩溃导致父进程挂起。在子进程通信模型中，子进程的崩溃（未捕获异常导致进程退出）会被父进程检测为code !== 0，触发Python process exited with code错误，这比静默失败要好得多。
结果限制：反汇编指令数、字符串数量都有限制（max_instructions默认500，strings默认返回200条），防止分析大文件时输出过大导致内存问题。这个限制同时保护了Python端（不消耗过多内存序列化结果）和Node.js端（不接收过大的JSON响应），以及LLM端（上下文窗口不会被单个工具调用撑满）。
4.4.2 文件资源暴露：将上传的二进制文件作为MCP Resource暴露
接下来，我们需要在MCP Server中暴露二进制文件资源。当用户上传一个二进制文件后，Server将其注册为可通过URI访问的Resource。资源注册表的设计需要考虑几个问题：内存中的Map适合开发和小规模使用，但在生产环境中可能需要持久化存储（如Redis或SQLite）来避免Server重启后丢失已注册的文件信息。此外，对于安全分析场景，还需要考虑文件的访问权限——哪些用户可以访问哪些文件，以及文件在系统中的保存策略（临时文件vs持久存储）。本章的实现使用内存Map作为简化方案，但在实际部署时应该根据安全要求选择适当的持久化和权限方案。
代码示例10：二进制文件资源的MCP暴露
// src/resources/binary-files.ts —— 二进制文件资源注册
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { readFileSync } from "fs";
// 文件注册表：file_id -> 文件信息
interface FileRecord {
path: string;
name: string;
size: number;
fileType: string;
architecture: string;
uploadedAt: Date;
}
const fileRegistry = new Map<string, FileRecord>();
export function registerFileResources(server: McpServer) {
// Resource: 二进制文件原始内容（base64编码）
server.resource(
"binary-raw-content",
"binary://{file_id}/raw",
{
description: "Raw binary content of an uploaded file, base64-encoded. " +
"Access this resource to get the actual bytes of the binary for deep analysis.",
mimeType: "application/json"
},
async (uri, { file_id }) => {
const file = fileRegistry.get(file_id);
if (!file) {
return {
contents: [{
uri: uri.href,
text: JSON.stringify({ error: `File not found: ${file_id}` }),
mimeType: "application/json"
}]
};
}
try {
const buffer = readFileSync(file.path);
// 限制返回大小：如果文件大于100KB，只返回前100KB
const maxSize = 100 * 1024;
const truncated = buffer.length > maxSize;
const data = truncated ? buffer.subarray(0, maxSize) : buffer;
return {
contents: [{
uri: uri.href,
text: JSON.stringify({
file_id,
name: file.name,
size: file.size,
returned_size: data.length,
truncated,
base64: data.toString("base64")
}, null, 2),
mimeType: "application/json"
}]
};
} catch (error) {
return {
contents: [{
uri: uri.href,
text: JSON.stringify({
error: "Failed to read file",
details: error instanceof Error ? error.message : "Unknown"
}),
mimeType: "application/json"
}]
};
}
}
);
// Resource: 文件分析索引
server.resource(
"binary-file-index",
"binary://files",
{
description: "Index of all registered binary files available for analysis. " +
"Use this to discover what files are currently available.",
mimeType: "application/json"
},
async (uri) => {
const files = Array.from(fileRegistry.entries()).map(([id, info]) => ({
file_id: id,
name: info.name,
size: info.size,
file_type: info.fileType,
architecture: info.architecture,
uploaded_at: info.uploadedAt.toISOString()
}));
return {
contents: [{
uri: uri.href,
text: JSON.stringify({
count: files.length,
files: files.sort((a, b) =>
new Date(b.uploaded_at).getTime() - new Date(a.uploaded_at).getTime()
)
}, null, 2),
mimeType: "application/json"
}]
};
}
);
}
// 导出注册函数供Tool调用
export function registerBinaryFile(
fileId: string,
filePath: string,
name: string,
size: number,
fileType: string,
architecture: string
): void {
fileRegistry.set(fileId, {
path: filePath,
name,
size,
fileType,
architecture,
uploadedAt: new Date()
});
}
export function getFileRecord(fileId: string): FileRecord | undefined {
return fileRegistry.get(fileId);
}
export function listRegisteredFiles(): Array<{ file_id: string } & Omit<FileRecord, "uploadedAt" | "path"> & { uploaded_at: string }> {
return Array.from(fileRegistry.entries()).map(([id, info]) => ({
file_id: id,
name: info.name,
size: info.size,
file_type: info.fileType,
architecture: info.architecture,
uploaded_at: info.uploadedAt.toISOString()
}));
}
4.4.3 分析结果格式化：统一分析结果的JSON Schema定义
为了让LLM能够一致地解析和处理分析结果，我们需要定义统一的JSON Schema。无论底层Python脚本使用哪个库（Capstone、pefile、LIEF），返回给LLM的数据结构都应该遵循相同的格式。
代码示例11：统一分析结果格式化和Python桥接层
// src/bridge/python-runner.ts —— Python桥接层：子进程管理
import { spawn } from "child_process";
import { resolve } from "path";
import { fileURLToPath } from "url";
const __dirname = fileURLToPath(new URL(".", import.meta.url));
// Python脚本路径（相对于项目根目录）
const PYTHON_SCRIPT = resolve(__dirname, "../../scripts/binary_analyzer.py");
interface PythonBridgeOptions {
timeout?: number; // 超时时间（毫秒），默认30000
pythonPath?: string; // Python解释器路径，默认"python3"
}
interface PythonBridgeResult<T = unknown> {
success: boolean;
data?: T;
error?: string;
exitCode: number;
executionTime: number; // 执行时间（毫秒）
}
/**
* 调用Python分析脚本
*
* 设计说明：
* - 每次调用启动一个新的Python子进程，确保状态隔离
* - 通过stdin发送JSON请求，stdout接收JSON响应
* - stderr用于Python日志输出
* - 支持超时控制，防止分析大文件时无限等待
*/
export async function callPythonAnalyzer<T = unknown>(
command: string,
params: Record<string, unknown>,
options: PythonBridgeOptions = {}
): Promise<PythonBridgeResult<T>> {
const { timeout = 30000, pythonPath = "python3" } = options;
const startTime = Date.now();
return new Promise((resolve) => {
// 启动Python子进程
const child = spawn(pythonPath, [PYTHON_SCRIPT], {
stdio: ["pipe", "pipe", "pipe"], // stdin, stdout, stderr均为管道
env: {
...process.env,
PYTHONUNBUFFERED: "1", // 禁用Python输出缓冲
}
});
let stdout = "";
let stderr = "";
let timeoutId: NodeJS.Timeout;
let finished = false;
// 收集stdout数据
child.stdout.on("data", (data: Buffer) => {
stdout += data.toString("utf-8");
});
// 收集stderr数据（Python日志）
child.stderr.on("data", (data: Buffer) => {
stderr += data.toString("utf-8");
});
// 设置超时
timeoutId = setTimeout(() => {
if (!finished) {
finished = true;
child.kill("SIGTERM"); // 优雅终止
// 如果5秒后仍未退出，强制kill
setTimeout(() => {
if (!child.killed) {
child.kill("SIGKILL");
}
}, 5000);
resolve({
success: false,
error: `Analysis timed out after ${timeout}ms. The file may be too large or the analysis too complex.`,
exitCode: -1,
executionTime: Date.now() - startTime
});
}
}, timeout);
// 子进程退出处理
child.on("close", (code) => {
if (finished) return;
finished = true;
clearTimeout(timeoutId);
const executionTime = Date.now() - startTime;
// 非零退出码表示Python脚本出错
if (code !== 0) {
resolve({
success: false,
error: `Python process exited with code ${code}. stderr: ${stderr.trim() || "N/A"}`,
exitCode: code ?? -1,
executionTime
});
return;
}
// 解析JSON输出
try {
const trimmed = stdout.trim();
if (!trimmed) {
resolve({
success: false,
error: "Python script produced no output",
exitCode: 0,
executionTime
});
return;
}
const result = JSON.parse(trimmed) as T;
// 检查Python脚本内部错误
if (result && typeof result === "object" && "error" in result) {
resolve({
success: false,
error: String((result as Record<string, unknown>).error),
data: result,
exitCode: 0,
executionTime
});
return;
}
resolve({
success: true,
data: result,
exitCode: 0,
executionTime
});
} catch (parseError) {
resolve({
success: false,
error: `Failed to parse Python output as JSON: ${parseError instanceof Error ? parseError.message : "Unknown"}. Raw output: ${stdout.substring(0, 500)}`,
exitCode: 0,
executionTime
});
}
});
child.on("error", (err) => {
if (!finished) {
finished = true;
clearTimeout(timeoutId);
resolve({
success: false,
error: `Failed to start Python process: ${err.message}. Ensure Python 3 and capstone are installed.`,
exitCode: -1,
executionTime: Date.now() - startTime
});
}
});
// 发送请求到Python stdin
const request = JSON.stringify({ command, params });
child.stdin.write(request + "\n");
child.stdin.end();
});
}
// ====== 类型化的分析结果接口 ======
export interface FileInfoResult {
file_path: string;
file_name: string;
file_size: number;
file_type: string;
architecture: string;
bitness: number;
sha256: string;
md5: string;
entropy: number;
header: Record<string, unknown>;
suspicious: {
high_entropy: boolean;
very_high_entropy: boolean;
};
}
export interface DisassemblyResult {
file: string;
architecture: string;
file_type: string;
instructions_analyzed: number;
max_instructions: number;
instructions: Array<{
address: string;
bytes: string;
mnemonic: string;
operands: string;
size: number;
}>;
error?: string;
fallback?: string;
}
export interface StringsResult {
file: string;
total_strings: number;
ascii_count: number;
unicode_count: number;
suspicious_count: number;
suspicious_strings: Array<{
address: string;
string: string;
type: string;
length: number;
match_reason?: string;
}>;
strings: Array<{
address: string;
string: string;
type: string;
length: number;
}>;
}
这个桥接层的设计考虑了生产环境的多个关键需求：
进程隔离：每次工具调用启动新的Python子进程，一个调用的失败不会影响其他调用。分析大文件的内存泄漏不会拖垮MCP Server本身。在子进程模型中，Python端的崩溃被限制在单个调用范围内——Node.js检测到非零退出码后，将该错误返回给LLM，然后继续处理后续请求。相比之下，如果分析逻辑直接内联在Node.js进程中，一个未处理的异常可能导致整个Server崩溃。
超时控制：timeout参数防止分析任务无限执行。如果Python进程在指定时间内没有退出，先发送SIGTERM优雅终止，5秒后若仍在运行则发送SIGKILL强制终止47^。这种两步终止策略既给了Python脚本清理资源的机会（关闭打开的文件句柄、释放内存），又确保了僵尸进程不会长期占用系统资源。在二进制分析场景中，某些特殊构造的文件（如极大但有效的PE文件、递归压缩的打包文件）可能导致分析算法的时间复杂度急剧上升，超时控制是防止这种情况影响系统稳定性的关键防线。
错误分级处理：区分四种错误类型——进程启动失败（Python未安装或路径错误）、进程执行错误（非零退出码，通常是Python脚本未捕获的异常）、JSON解析错误（Python输出了非JSON内容，可能是print语句或库日志）、Python内部错误（脚本正确运行但返回了包含error字段的结果）。每种错误都有清晰的错误信息，帮助LLM和开发者快速定位问题根因。
结果验证：解析JSON后检查error字段，Python脚本内部的错误（如文件不存在、Capstone未安装）被正确传递而不是被忽略。这个双向错误通道非常重要——Node.js端无法预知Python脚本内部的业务逻辑错误，只有Python自己知道”文件格式不支持”或”Capstone引擎未安装”这样的具体问题。
执行时间追踪：每个调用都记录executionTime，用于性能监控和后续优化。当某个工具的平均执行时间突然增加时，可能意味着Python脚本遇到了性能退化，或输入数据的复杂度增加了。这些数据对于生产环境的容量规划和SLA监控至关重要。
4.4.4 Server启动与测试：stdio模式启动、Inspector调试工具使用
现在我们将所有组件组合起来，创建完整的Server入口文件，并学习如何使用MCP Inspector进行测试。在深入代码之前，先理解Server入口文件的设计思路：它负责注册所有Tools和Resources，初始化Python桥接，配置错误处理，然后启动stdio Transport进入事件循环。这个文件是MCP Server的”指挥中心”，所有组件在此汇合。
// src/index.ts —— 二进制分析MCP Server主入口
#!/usr/bin/env node
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { callPythonAnalyzer } from "./bridge/python-runner.js";
import { registerFileResources } from "./resources/binary-files.js";
import { registerBinaryFile } from "./resources/binary-files.js";
const server = new McpServer({
name: "binary-analysis-mcp-server",
version: "1.0.0"
});
// ====== Tool 1: 获取文件信息 ======
server.tool(
"get_file_info",
{
description: "Analyze a binary file and return its metadata including " +
"file type (PE/ELF/Mach-O/Raw), architecture, size, entropy, and header info. " +
"Use this as the first step when examining any unknown binary file.",
inputSchema: z.object({
file_path: z.string()
.describe("Absolute path to the binary file to analyze")
.min(1, "File path cannot be empty")
}),
annotations: {
readOnlyHint: true,
destructiveHint: false,
idempotentHint: true,
openWorldHint: false,
}
},
async ({ file_path }) => {
const result = await callPythonAnalyzer("file_info", { file_path });
if (!result.success) {
return {
content: [{ type: "text", text: JSON.stringify({ error: result.error }, null, 2) }],
isError: true
};
}
// 注册文件到Resource系统
const data = result.data as { file_name?: string; file_size?: number; file_type?: string; architecture?: string };
const fileId = Buffer.from(file_path).toString("base64url");
registerBinaryFile(
fileId, file_path,
data.file_name || "unknown",
data.file_size || 0,
data.file_type || "Unknown",
data.architecture || "unknown"
);
return {
content: [{ type: "text", text: JSON.stringify(result.data, null, 2) }]
};
}
);
// ====== Tool 2: 反汇编分析 ======
server.tool(
"disassemble",
{
description: "Disassemble a binary file and return instructions with addresses, " +
"mnemonics, and operands. Use this to examine the actual machine code of a binary. " +
"Best for understanding program logic, identifying functions, and finding vulnerabilities.",
inputSchema: z.object({
file_path: z.string()
.describe("Absolute path to the binary file to disassemble"),
max_instructions: z.number().int().min(1).max(2000).optional()
.describe("Maximum number of instructions to return (1-2000, default: 500)")
}),
annotations: {
readOnlyHint: true,
idempotentHint: true,
openWorldHint: false,
}
},
async ({ file_path, max_instructions = 500 }) => {
const result = await callPythonAnalyzer("disassemble", {
file_path,
max_instructions
}, { timeout: 60000 }); // 反汇编可能需要更长时间
if (!result.success) {
return {
content: [{ type: "text", text: JSON.stringify({ error: result.error }, null, 2) }],
isError: true
};
}
return {
content: [{ type: "text", text: JSON.stringify(result.data, null, 2) }]
};
}
);
// ====== Tool 3: 字符串提取 ======
server.tool(
"extract_strings",
{
description: "Extract ASCII and Unicode strings from a binary file. " +
"Useful for finding URLs, file paths, registry keys, error messages, " +
"debug information, and other human-readable content embedded in binaries. " +
"Automatically flags suspicious strings (URLs, API calls, passwords, etc.).",
inputSchema: z.object({
file_path: z.string()
.describe("Absolute path to the binary file"),
min_length: z.number().int().min(1).max(50).optional()
.describe("Minimum string length to include (default: 4)"),
encoding: z.enum(["all", "ascii", "unicode"]).optional()
.describe("String encoding to extract (default: all)")
}),
annotations: {
readOnlyHint: true,
idempotentHint: true,
openWorldHint: false,
}
},
async ({ file_path, min_length = 4, encoding = "all" }) => {
const result = await callPythonAnalyzer("extract_strings", {
file_path,
min_length,
encoding
}, { timeout: 30000 });
if (!result.success) {
return {
content: [{ type: "text", text: JSON.stringify({ error: result.error }, null, 2) }],
isError: true
};
}
return {
content: [{ type: "text", text: JSON.stringify(result.data, null, 2) }]
};
}
);
// 注册Resources
registerFileResources(server);
// ====== 启动Server ======
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("Binary Analysis MCP Server running on stdio");
console.error("Tools: get_file_info, disassemble, extract_strings");
}
main().catch((error) => {
console.error("Fatal error:", error);
process.exit(1);
});
使用MCP Inspector测试：
MCP Inspector是Anthropic官方提供的调试工具，让你可以交互式地测试MCP Server的工具和资源48^。
安装和启动Inspector：
# 无需单独安装，通过npx直接运行
npx @anthropics/mcp-inspector node dist/index.js
# Inspector会启动一个Web界面，通常在 http://localhost:5173
# 打开浏览器即可交互式测试
Inspector提供以下测试能力：
- Tools面板：查看所有注册的工具及其Schema描述，手动调用工具并查看返回结果。
- Resources面板：浏览所有可访问的资源URI，读取资源内容。
- Prompts面板：查看所有提示模板，填充参数并预览生成的消息序列。
- Server Info面板：查看Server元数据、协议版本、能力声明。
测试get_file_info工具的流程：准备一个测试用的二进制文件（如/bin/ls），在Inspector的Tools面板中选择get_file_info，输入{"file_path": "/bin/ls"}，点击执行。你应该看到返回的JSON包含文件类型、架构、大小、熵值等元数据。
测试注意事项：
- 确保Python 3已安装：
python3 --version - 确保Capstone已安装：
pip3 install capstone（可选，但推荐用于完整反汇编功能） - 测试文件路径使用绝对路径
- 如果Python脚本不在预期位置，检查
PYTHON_SCRIPT路径配置
常见故障排除：
如果在Inspector中看到"error": "Python process exited with code..."错误，首先检查stderr输出。常见原因包括：Python脚本路径不正确（检查PYTHON_SCRIPT常量是否指向正确位置）、缺少Python依赖（运行pip3 install capstone安装Capstone反汇编引擎）、或文件权限问题（确保Node.js进程有权限读取目标文件和Python脚本）。
如果看到"error": "Analysis timeout after 60000ms"超时错误，通常是因为分析的文件过大或Capstone在处理复杂二进制时耗时较长。可以尝试减小max_instructions参数（比如从500减到100），或增加timeout配置。如果超时频繁发生，建议在生产环境中使用进程池方案来控制并发。
如果反汇编结果为空或只有很少指令，可能是因为Capstone未能正确识别文件架构。可以先用get_file_info工具确认文件的architecture和file_type字段，如果显示为"unknown"，说明自动检测失败了。此时Python脚本会回退到默认的x64架构反汇编，可能导致大量无效指令。对于自定义或加壳的二进制文件，这种情况是正常的46^。
4.5 项目里程碑：二进制分析MCP Server
本节将本章的所有知识整合为一个完整的项目交付物。你将获得完整的Server实现代码、测试验证方案和Python桥接方案的最佳实践。
4.5.1 完整MCP Server实现：包含反汇编、字符串提取、文件信息三个工具
以下是整合所有组件的完整MCP Server实现，包含了三个核心分析工具、文件资源暴露、错误处理和Python桥接的完整代码。
代码示例12：完整MCP Server实现与Python桥接
// src/index.ts —— 完整二进制分析MCP Server
#!/usr/bin/env node
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";
import { spawn } from "child_process";
import { resolve } from "path";
import { fileURLToPath } from "url";
import { readFileSync } from "fs";
const __dirname = fileURLToPath(new URL(".", import.meta.url));
const PYTHON_SCRIPT = resolve(__dirname, "../scripts/binary_analyzer.py");
// ====== 类型定义 ======
interface AnalysisResult<T = unknown> {
success: boolean;
data?: T;
error?: string;
exitCode: number;
executionTime: number;
}
interface FileRecord {
path: string;
name: string;
size: number;
fileType: string;
architecture: string;
uploadedAt: Date;
}
// ====== 文件注册表 ======
const fileRegistry = new Map<string, FileRecord>();
// ====== Python桥接层 ======
async function callPython<T>(
command: string,
params: Record<string, unknown>,
options: { timeout?: number; pythonPath?: string } = {}
): Promise<AnalysisResult<T>> {
const { timeout = 30000, pythonPath = "python3" } = options;
const startTime = Date.now();
return new Promise((resolve) => {
const child = spawn(pythonPath, [PYTHON_SCRIPT], {
stdio: ["pipe", "pipe", "pipe"],
env: { ...process.env, PYTHONUNBUFFERED: "1" }
});
let stdout = "";
let stderr = "";
let finished = false;
child.stdout.on("data", (d: Buffer) => { stdout += d.toString(); });
child.stderr.on("data", (d: Buffer) => { stderr += d.toString(); });
const timeoutId = setTimeout(() => {
if (finished) return;
finished = true;
child.kill("SIGTERM");
setTimeout(() => { if (!child.killed) child.kill("SIGKILL"); }, 5000);
resolve({ success: false, error: `Timeout after ${timeout}ms`, exitCode: -1, executionTime: Date.now() - startTime });
}, timeout);
child.on("close", (code) => {
if (finished) return;
finished = true;
clearTimeout(timeoutId);
const elapsed = Date.now() - startTime;
if (code !== 0) {
resolve({ success: false, error: `Process exited with code ${code}. stderr: ${stderr.trim() || "N/A"}`, exitCode: code ?? -1, executionTime: elapsed });
return;
}
try {
const data = JSON.parse(stdout.trim()) as T;
if (data && typeof data === "object" && "error" in (data as Record<string, unknown>)) {
resolve({ success: false, error: String((data as Record<string, unknown>).error), data, exitCode: 0, executionTime: elapsed });
return;
}
resolve({ success: true, data, exitCode: 0, executionTime: elapsed });
} catch {
resolve({ success: false, error: `JSON parse error. Output: ${stdout.substring(0, 300)}`, exitCode: 0, executionTime: elapsed });
}
});
child.on("error", (err) => {
if (finished) return;
finished = true;
clearTimeout(timeoutId);
resolve({ success: false, error: `Spawn error: ${err.message}`, exitCode: -1, executionTime: Date.now() - startTime });
});
child.stdin.write(JSON.stringify({ command, params }) + "\n");
child.stdin.end();
});
}
// ====== 创建Server ======
const server = new McpServer({
name: "binary-analysis-mcp-server",
version: "1.0.0"
});
// ====== Tool 1: get_file_info ======
server.tool(
"get_file_info",
{
description: "Analyze a binary file and return its metadata: file type " +
"(PE/ELF/Mach-O/Raw), architecture (x86/x64/ARM), size, SHA256 hash, " +
"entropy (packing/encryption detection), and header details. " +
"Use this as the FIRST step when examining any binary.",
inputSchema: z.object({
file_path: z.string().min(1).describe("Absolute path to the binary file")
}),
annotations: { readOnlyHint: true, idempotentHint: true }
},
async ({ file_path }) => {
const result = await callPython("file_info", { file_path });
if (!result.success) {
return { content: [{ type: "text", text: JSON.stringify({ error: result.error }, null, 2) }], isError: true };
}
// 注册到资源系统
const d = result.data as Record<string, unknown>;
fileRegistry.set(Buffer.from(file_path).toString("base64url"), {
path: file_path, name: String(d.file_name || "unknown"),
size: Number(d.file_size || 0), fileType: String(d.file_type || "Unknown"),
architecture: String(d.architecture || "unknown"), uploadedAt: new Date()
});
return { content: [{ type: "text", text: JSON.stringify(result.data, null, 2) }] };
}
);
// ====== Tool 2: disassemble ======
server.tool(
"disassemble",
{
description: "Disassemble a binary file into human-readable assembly instructions. " +
"Returns addresses, bytes, mnemonics, and operands. " +
"Use for understanding program logic, identifying functions, and finding vulnerabilities. " +
"Requires capstone engine (pip install capstone).",
inputSchema: z.object({
file_path: z.string().min(1).describe("Absolute path to the binary file"),
max_instructions: z.number().int().min(1).max(2000).optional()
.describe("Max instructions to return (1-2000, default: 500)")
}),
annotations: { readOnlyHint: true, idempotentHint: true }
},
async ({ file_path, max_instructions = 500 }) => {
const result = await callPython("disassemble", { file_path, max_instructions }, { timeout: 60000 });
if (!result.success) {
return { content: [{ type: "text", text: JSON.stringify({ error: result.error }, null, 2) }], isError: true };
}
return { content: [{ type: "text", text: JSON.stringify(result.data, null, 2) }] };
}
);
// ====== Tool 3: extract_strings ======
server.tool(
"extract_strings",
{
description: "Extract ASCII and Unicode strings from a binary file. " +
"Useful for finding URLs, API names, registry keys, passwords, " +
"error messages, and debug info. Automatically flags suspicious strings.",
inputSchema: z.object({
file_path: z.string().min(1).describe("Absolute path to the binary file"),
min_length: z.number().int().min(1).max(50).optional()
.describe("Minimum string length (default: 4)"),
encoding: z.enum(["all", "ascii", "unicode"]).optional()
.describe("Encoding filter (default: all)")
}),
annotations: { readOnlyHint: true, idempotentHint: true }
},
async ({ file_path, min_length = 4, encoding = "all" }) => {
const result = await callPython("extract_strings", { file_path, min_length, encoding });
if (!result.success) {
return { content: [{ type: "text", text: JSON.stringify({ error: result.error }, null, 2) }], isError: true };
}
return { content: [{ type: "text", text: JSON.stringify(result.data, null, 2) }] };
}
);
// ====== Resources ======
server.resource(
"binary-raw-content",
"binary://{file_id}/raw",
{ description: "Base64-encoded raw binary content", mimeType: "application/json" },
async (uri, { file_id }) => {
const file = fileRegistry.get(file_id);
if (!file) return { contents: [{ uri: uri.href, text: JSON.stringify({ error: `Not found: ${file_id}` }), mimeType: "application/json" }] };
const buf = readFileSync(file.path);
const max = 100 * 1024;
return { contents: [{ uri: uri.href, text: JSON.stringify({ file_id, size: file.size, truncated: buf.length > max, base64: (buf.length > max ? buf.subarray(0, max) : buf).toString("base64") }), mimeType: "application/json" }] };
}
);
server.resource(
"binary-file-index",
"binary://files",
{ description: "List of all registered binary files", mimeType: "application/json" },
async (uri) => {
const files = Array.from(fileRegistry.entries()).map(([id, f]) => ({
file_id: id, name: f.name, size: f.size,
file_type: f.fileType, architecture: f.architecture,
uploaded_at: f.uploadedAt.toISOString()
}));
return { contents: [{ uri: uri.href, text: JSON.stringify({ count: files.length, files }, null, 2), mimeType: "application/json" }] };
}
);
// ====== Prompt: 安全分析模板 ======
server.prompt(
"security-analysis",
{
description: "Comprehensive security analysis prompt for binary files",
argsSchema: {
file_path: z.string().describe("Path to the binary file"),
file_type: z.enum(["PE", "ELF", "Mach-O", "Raw"]).optional()
}
},
async ({ file_path, file_type = "Raw" }) => ({
messages: [{
role: "user",
content: {
type: "text",
text: `Analyze ${file_type} binary at ${file_path} for security vulnerabilities. ` +
`Use get_file_info first, then disassemble and extract_strings. ` +
`Focus on: buffer overflows, dangerous APIs, suspicious strings, ` +
`crypto weaknesses. Output findings as JSON with severity, confidence, addresses.`
}
}]
})
);
// ====== 启动 ======
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error("Binary Analysis MCP Server v1.0.0 running on stdio");
console.error("Available tools: get_file_info, disassemble, extract_strings");
}
main().catch(err => { console.error("Fatal:", err); process.exit(1); });
这个完整实现遵循了MCP Server开发的最佳实践：
工具描述质量。每个工具的description都包含三个要素：功能描述（做什么）、使用场景（什么时候用）、依赖说明（需要什么）。这使得LLM能够准确地在合适的时机调用正确的工具。
参数验证。Zod Schema不仅定义了类型，还定义了约束：file_path必须非空，max_instructions在1-2000之间，min_length在1-50之间。这些约束在运行时自动验证。
错误分级。桥接层区分了进程级错误（Python未安装、超时）和业务级错误（文件不存在、Capstone未安装），每种错误都有清晰的信息和建议。
资源自动注册。当调用get_file_info时，文件自动注册到资源系统，后续可以通过binary://{file_id}/raw URI访问原始内容。这种自动注册机制的设计考虑是：用户在分析一个文件时，通常的第一步就是获取文件信息。在获取文件信息的同时自动注册到资源系统，避免了用户需要执行额外的注册步骤。这种隐式注册简化了交互流程，让用户体验更加流畅——用户只需要提供一个文件路径，后续所有分析工具和资源访问都可以围绕这个文件展开。
完整的Server代码还包含了一个安全分析Prompt模板（security-analysis），这个Prompt模板是整个Server中最体现MCP设计哲学的部分。它不仅告诉LLM要做什么（分析二进制文件的安全性），还告诉LLM怎么做（先用get_file_info获取基础信息，再用disassemble和extract_strings获取详细数据），以及如何组织输出（JSON格式，包含severity、confidence、address等字段）。这种Prompt + Tool的组合模式是MCP区别于传统API的关键——Server不仅是能力的提供者，更是工作流的设计者44^。
4.5.2 测试验证：用MCP Inspector测试各工具的正确性
编译Server并运行测试：
# 1. 编译TypeScript
npx tsc
# 2. 确保Python脚本在正确位置
mkdir -p scripts
cp binary_analyzer.py scripts/
# 3. 使用MCP Inspector启动Server
npx @anthropics/mcp-inspector node dist/index.js
Inspector启动后，打开浏览器访问http://localhost:5173。按以下顺序测试：
测试1：get_file_info
选择一个系统二进制文件进行测试（如/bin/ls或C:\Windows\System32\notepad.exe）：
{
"file_path": "/bin/ls"
}
期望返回包含file_type、architecture、file_size、entropy、sha256的JSON对象。如果文件是ELF格式，还应该包含header字段（entry point、program header offset等）。
测试2：extract_strings
{
"file_path": "/bin/ls",
"min_length": 5,
"encoding": "all"
}
期望返回包含total_strings、ascii_count、unicode_count、suspicious_count的JSON对象。strings数组应该包含地址、字符串内容、类型和长度的条目。suspicious_strings数组应该包含被标记为可疑的字符串及其匹配原因。
测试3：disassemble
{
"file_path": "/bin/ls",
"max_instructions": 100
}
如果Capstone已安装，期望返回包含instructions数组的JSON对象，每个指令包含address、bytes、mnemonic、operands和size。如果Capstone未安装，应该返回友好的错误消息，提示安装pip install capstone。
测试4：错误处理
测试文件不存在的场景：
{
"file_path": "/nonexistent/file.bin"
}
期望返回isError: true的响应，content中包含清晰的错误信息。验证错误信息是否包含”File not found”而不是晦涩的Python异常栈跟踪。
测试5：大文件超时
选择一个大于100MB的文件测试超时行为：
{
"file_path": "/path/to/large/file.dmg",
"max_instructions": 2000
}
如果分析时间超过配置的超时值（默认30秒），应该返回"error": "Analysis timeout after 30000ms"而不是进程挂起或崩溃。
测试6：Resource访问
在Inspector的Resources面板中，先调用get_file_info注册一个文件，然后尝试访问binary://files查看索引，以及binary://{file_id}/raw获取原始内容（file_id是文件路径的base64url编码）。
测试7：Prompt模板验证
在Inspector的Prompts面板中选择security-analysis，填入参数{"file_path": "/bin/ls", "file_type": "ELF"}，验证生成的消息序列是否包含完整的安全分析指令和输出格式要求。质量好的Prompt模板应该让LLM在没有额外指导的情况下就能执行标准化的分析流程。
4.5.3 Python桥接方案：Node.js调用Python脚本的子进程管理方案
Node.js与Python之间的桥接是本项目的核心架构决策。本节详细说明桥接方案的设计考虑、实现细节和生产环境最佳实践。
架构选型分析：
为什么不用纯TypeScript实现二进制分析？主要有三个原因。第一，二进制分析生态以Python为主导——Capstone、pefile、LIEF、pwntools等核心库都是Python原生实现49^，用TypeScript重新实现这些功能不仅工作量大，而且难以保持与上游库的同步更新。第二，Python在科学计算和数据处理方面的生态远优于JavaScript/TypeScript，NumPy、Pandas等库可以大大简化分析数据的处理。第三，Python脚本是独立的进程，分析任务崩溃不会影响MCP Server的稳定性。
为什么不用HTTP API而直接用stdio？对于本地部署场景，stdio比HTTP更简单——不需要端口管理、不需要HTTP客户端、没有网络开销。每次工具调用就是一次子进程启动和stdin/stdout通信，语义清晰且调试方便50^。
子进程管理的完整最佳实践：
// src/bridge/python-runner-robust.ts —— 生产级Python桥接
import { spawn, ChildProcess } from "child_process";
import { resolve } from "path";
// 进程池配置
interface ProcessPoolOptions {
maxConcurrent: number; // 最大并发进程数
maxQueued: number; // 最大队列长度
idleTimeout: number; // 空闲进程超时（毫秒）
pythonPath: string;
scriptPath: string;
}
interface QueuedTask<T> {
command: string;
params: Record<string, unknown>;
timeout: number;
resolve: (value: AnalysisResult<T>) => void;
startTime: number;
}
/**
* 生产级Python进程池
*
* 特性：
* - 限制并发进程数，防止系统资源耗尽
* - 请求队列管理，超出容量时优雅拒绝
* - 进程超时控制，防止僵尸进程
* - 错误重试机制（对临时性错误）
*/
class PythonProcessPool {
private running = 0;
private queue: Array<QueuedTask<unknown>> = [];
private options: ProcessPoolOptions;
constructor(options: Partial<ProcessPoolOptions> = {}) {
this.options = {
maxConcurrent: options.maxConcurrent ?? 3,
maxQueued: options.maxQueued ?? 10,
idleTimeout: options.idleTimeout ?? 30000,
pythonPath: options.pythonPath ?? "python3",
scriptPath: options.scriptPath ?? resolve(process.cwd(), "scripts/binary_analyzer.py")
};
}
async execute<T>(
command: string,
params: Record<string, unknown>,
timeout = 30000
): Promise<AnalysisResult<T>> {
// 队列满时直接拒绝
if (this.queue.length >= this.options.maxQueued) {
return {
success: false,
error: `Analysis queue full (${this.options.maxQueued} pending). Try again later.`,
exitCode: -1,
executionTime: 0
};
}
// 如果未达到并发上限，直接执行
if (this.running < this.options.maxConcurrent) {
return this.runTask<T>(command, params, timeout);
}
// 否则加入队列等待
return new Promise((resolve) => {
this.queue.push({ command, params, timeout, resolve: resolve as (value: AnalysisResult<unknown>) => void, startTime: Date.now() });
}) as Promise<AnalysisResult<T>>;
}
private async runTask<T>(
command: string,
params: Record<string, unknown>,
timeout: number
): Promise<AnalysisResult<T>> {
this.running++;
const startTime = Date.now();
try {
const result = await this.spawnAndCollect<T>(command, params, timeout);
// 如果队列中有等待的任务，继续执行
const next = this.queue.shift();
if (next) {
this.runTask(next.command, next.params, next.timeout)
.then(next.resolve);
}
return result;
} finally {
this.running--;
}
}
private spawnAndCollect<T>(
command: string,
params: Record<string, unknown>,
timeout: number
): Promise<AnalysisResult<T>> {
return new Promise((resolve) => {
let child: ChildProcess;
try {
child = spawn(this.options.pythonPath, [this.options.scriptPath], {
stdio: ["pipe", "pipe", "pipe"],
env: { ...process.env, PYTHONUNBUFFERED: "1" }
});
} catch (err) {
resolve({
success: false,
error: `Failed to spawn Python: ${err instanceof Error ? err.message : "Unknown"}`,
exitCode: -1,
executionTime: Date.now() - startTime
});
return;
}
let stdout = "";
let stderr = "";
let finished = false;
const taskStart = Date.now();
// 数据收集
child.stdout!.on("data", (d: Buffer) => { stdout += d.toString(); });
child.stderr!.on("data", (d: Buffer) => { stderr += d.toString(); });
// 超时处理
const timeoutId = setTimeout(() => {
if (finished) return;
finished = true;
// 优雅终止序列
child.kill("SIGTERM");
const forceKillId = setTimeout(() => {
if (!child.killed) child.kill("SIGKILL");
}, 5000);
child.on("close", () => clearTimeout(forceKillId));
resolve({
success: false,
error: `Analysis timeout after ${timeout}ms`,
exitCode: -1,
executionTime: Date.now() - taskStart
});
}, timeout);
// 进程退出
child.on("close", (code) => {
if (finished) return;
finished = true;
clearTimeout(timeoutId);
const elapsed = Date.now() - taskStart;
if (code !== 0) {
resolve({
success: false,
error: `Exit code ${code}. stderr: ${stderr.trim() || "N/A"}`,
exitCode: code ?? -1,
executionTime: elapsed
});
return;
}
try {
const trimmed = stdout.trim();
if (!trimmed) throw new Error("Empty output");
const data = JSON.parse(trimmed) as T;
if (data && typeof data === "object" && "error" in (data as Record<string, unknown>)) {
resolve({ success: false, error: String((data as Record<string, unknown>).error), data, exitCode: 0, executionTime: elapsed });
return;
}
resolve({ success: true, data, exitCode: 0, executionTime: elapsed });
} catch {
resolve({ success: false, error: `Parse error: ${stdout.substring(0, 200)}`, exitCode: 0, executionTime: elapsed });
}
});
// 发送请求
child.stdin!.write(JSON.stringify({ command, params }) + "\n");
child.stdin!.end();
});
}
}
// 导出单例
export const pythonPool = new PythonProcessPool();
// 类型定义
export interface AnalysisResult<T> {
success: boolean;
data?: T;
error?: string;
exitCode: number;
executionTime: number;
}
生产级桥接方案在基础版本上增加了以下关键能力：
并发控制。maxConcurrent限制同时运行的Python进程数（默认3个），防止大文件分析同时提交时耗尽系统资源。maxQueued限制队列长度，超出时新请求立即被拒绝而不是无限等待。
优雅终止。超时后先发送SIGTERM信号，给予Python进程5秒时间清理资源（关闭文件句柄、释放内存）。如果进程仍然存活，再发送SIGKILL强制终止。这种两步终止策略避免了数据损坏和资源泄漏51^。
队列管理。当并发数达到上限时，新请求进入FIFO队列。任务完成后自动从队列中取出下一个执行，而不是让每个调用者独立等待。
进程池单例。整个MCP Server共享一个PythonProcessPool实例，确保全局层面的并发控制，而不是每个工具调用各自管理进程。
部署检查清单：
将二进制分析MCP Server部署到生产环境前，确认以下事项：
| 检查项 | 说明 | 命令/方法 |
| — | — | — |
| Node.js版本 | 需要18+ | node --version |
| Python 3可用 | Python 3.8+ | python3 --version |
| Capstone引擎 | 反汇编必需 | pip3 install capstone |
| stdio管道权限 | stdout/stderr未被重定向 | 启动后观察日志输出 |
| 文件系统权限 | 能读取目标二进制文件 | ls -la <target_file> |
| 超时配置 | 大文件分析需要更长超时 | 调整timeout参数 |
| 日志监控 | stderr日志可查看 | 配置日志收集 |
| 进程限制 | 系统ulimit允许足够进程 | ulimit -u |
| Python脚本路径 | NODE_PATH环境变量或直接指定 | 检查脚本可执行权限 |
| 内存限制 | Python分析大文件可能消耗大量内存 | 监控系统内存使用 |
安全配置注意事项：
二进制分析MCP Server处理的是用户提交的文件，安全是需要特别关注的方面。首先，确保分析的文件被保存在受限的临时目录中，分析完成后及时清理，避免恶意文件长期驻留在系统中。其次，对输入的文件路径进行验证，防止路径遍历攻击（如../../../etc/passwd）。Python脚本中的路径验证应该检查目标路径是否在允许的目录范围内。第三，限制单个文件的大小（如最大100MB），防止资源耗尽攻击。最后，如果Server运行在多用户环境中，考虑为每个用户创建独立的分析沙箱，防止一个用户的分析任务访问另一个用户的文件51^。
性能优化建议：
对于频繁使用的分析服务，有几个性能优化方向值得考虑。其一，可以实现分析结果缓存——相同的文件（通过SHA256识别）不需要重复分析，直接从缓存返回结果。这在分析团队反复查看同一个二进制文件时非常有效。其二，对于Capstone反汇编，可以预加载引擎实例而不是每次调用都重新初始化，这能将反汇编延迟降低50%以上。其三，Python端可以使用multiprocessing.Pool来并行处理多个分析请求，而不是完全依赖Node.js端的串行控制。这些优化应该在你验证了基础功能的正确性之后再实施——过早优化是万恶之源47^。
扩展能力路线图：
本章实现的二进制分析MCP Server已经具备了核心分析能力，但在实际的安全研究场景中，你可能需要进一步扩展它的功能。以下是按优先级排序的扩展建议：
| 扩展功能 | 说明 | Python依赖 |
| — | — | — |
| PE/ELF导入表分析 | 列出DLL/符号导入，识别危险API | pefile, pyelftools |
| 熵值分段分析 | 检测加壳/加密区域 | 标准库 |
| YARA规则扫描 | 匹配恶意软件签名 | yara-python |
| 控制流图生成 | 可视化函数调用关系 | networkx, capstone |
| 反编译 | 汇编到伪C代码 | angr 或 Ghidra API |
| 漏洞签名匹配 | 基于已知漏洞模式检测 | 自定义规则库 |
与现有工具链的集成：
这个MCP Server的设计理念是互补而非替代。它不是要取代IDA Pro、Ghidra或Binary Ninja这样的专业逆向工程工具，而是为这些工具的能力提供一个标准化的AI可访问接口。安全分析师仍然使用IDA Pro进行深度逆向分析，但当需要快速批量分析、自动化检测或AI辅助审查时，MCP Server提供了轻量级的替代方案。
在实际工作流中，一个典型的使用场景是：安全分析师将可疑的二进制文件拖入Claude Desktop（已配置本MCP Server），通过对话方式让AI执行初步分析——获取文件信息、提取可疑字符串、反汇编关键函数。AI会调用相应的MCP工具获取数据，然后基于分析结果给出安全评估。如果AI发现了可疑指标，分析师再决定是否需要使用Ghidra进行深度逆向。这种人机协作模式大大提高了初步分析的效率49^。
从stdio到Streamable HTTP的迁移路径：
当你的分析服务需要从本地单机扩展到团队共享时，stdio模式的局限性就显现出来了——它只支持本地一对一连接。迁移到Streamable HTTP的过程相对简单：
- 将
StdioServerTransport替换为StreamableHTTPServerTransport - 添加Express或Fastify路由处理HTTP请求
- 配置认证（Bearer Token或OAuth 2.1）
- Python桥接部分完全不需要修改——它仍然通过子进程stdio与Node.js通信
这种分层架构的优势在于，传输层的变更不会影响业务逻辑层。无论你是本地stdio还是远程HTTP，分析工具的注册、Python桥接的调用、结果格式化的逻辑都完全一致。
完成本章的学习后，你已经拥有了一个功能完整的二进制分析MCP Server。它能够：
- 通过stdio模式与任何MCP兼容客户端通信
- 暴露三个核心分析工具（文件信息、反汇编、字符串提取）
- 通过Resource暴露二进制文件内容
- 提供安全分析的标准化Prompt模板
- 通过Python桥接复用Capstone等成熟的二进制分析库
- 具备超时控制、错误处理和并发管理的生产级稳定性
在第五章中，你将学习如何编写MCP Client来连接和使用这个Server，以及如何在一个Agent中同时协调多个MCP Server的能力。你还将学习如何将MCP工具集成到Agent的工具调用循环中，让Agent能够动态发现并调用你刚刚创建的Binary Analysis Server——这标志着你的Agent从内置工具阶段迈入了开放的工具生态系统阶段。
MCP生态系统的参与方式：
当你开发了一个有用的MCP Server后，可以通过多种方式分享给社区。首先，将你的Server发布到npm（Node.js Server）或PyPI（Python Server），让其他开发者可以通过包管理器安装。其次，在MCP社区注册表（如Glama、Smithery等第三方索引平台）中提交你的Server，提高可见性。提交时需要提供Server名称、描述、安装命令、配置示例和权限说明。最后，在项目的README中详细说明Server的功能、配置选项和安全注意事项——清晰的文档是Server被广泛采用的关键因素50^。
对于企业内部的MCP Server，建议建立私有注册中心或使用内部npm/PyPI仓库。通过内部文档和培训推广MCP的使用，让更多团队了解如何通过标准化协议共享工具能力。一个成功的内部MCP生态系统通常从几个高价值的通用工具（如数据库查询、文件操作、日志分析）开始，逐步扩展到更多业务领域。
免责声明：
本文所载程序、技术方法仅面向合法合规的安全研究与教学场景，旨在提升网络安全防护能力，具有明确的技术研究属性。
任何单位或个人未经授权，将本文内容用于攻击、破坏等非法用途的，由此引发的全部法律责任、民事赔偿及连带责任，均由行为人独立承担，本站不承担任何连带责任。
本站内容均为技术交流与知识分享目的发布，若存在版权侵权或其他异议，请通过邮件联系处理，具体联系方式可点击页面上方的联系我。
本文转载自：SPEEDCoding 李北辰
李北辰《4. MCP协议基础与Server开发》