接上一篇 MCP 入门，这次来点硬核的——从零搭建一个支持多模型切换的视觉理解 MCP Server。代码全量可跑，踩过的坑也都标注了。—## 一句话概括通过 MCP 协议封装 OpenAI GPT-4o、Gemini、GLM-4V、Qwen-VL 等多模态大模型，让 Claude Desktop 或 Cursor 能直接「看懂」图片。核心思路：模型适配器模式 + 任务感知路由 + 场景化 Tool 拆分。—## 1. 问题拆解让 AI 理解一张图片，技术链路分四步：1. 图像采集：从摄像头、截图或文件获取原始字节流2. 编码传输：把图像编码成模型能接受的格式（不同模型要求不同）3. 模型推理：调用多模态 API，获取理解结果4. 结果返回：通过 MCP 协议将分析结果传回 AI Host每一步都有坑：- 格式碎片化：OpenAI 要 base64 data URL，Gemini 能直接吃 bytes，GLM-4V 和 Qwen-VL 各有各的参数格式- 大图烧 token：4K 原图 base64 后几百 KB，一次调用吃掉上千 token- 实时场景延迟敏感：摄像头实时分析场景下，端到端延迟体验至关重要—## 2. 架构总览┌──────────────┐ MCP Protocol ┌──────────────────┐│ AI Host │ ◄──────────────────► │ Vision MCP ││ (Claude / │ (JSON-RPC over │ Server ││ Cursor) │ stdio/HTTP+SSE) │ │└──────────────┘ │ ┌─────────────┐ │ │ │ Image │ │ │ │ Preprocessor │ │ │ └──────┬──────┘ │ │ │ │ │ ┌──────▼──────┐ │ │ │ Model Router │ │ │ │ (task-aware) │ │ │ └──────┬──────┘ │ │ │ │ ┌─────────────────────────┼─────────┼─────────┤ │ │ │ │ ┌──────▼──────┐ ┌──────▼──────┐ ┌──────▼──────┐ ... │ OpenAI │ │ Gemini │ │ GLM-4V │ │ (GPT-4o) │ │ (2.0 Flash) │ │ │ └─────────────┘ └─────────────┘ └─────────────┘架构设计的关键决策：Tool 粒度要按场景拆分，而不是做一个万能的 analyze_image。原因后面详述。—## 3. 核心实现### 3.1 目录结构mcp-vision-server/├── server.py # MCP Server 主入口├── models/│ ├── __init__.py│ ├── base.py # 抽象基类│ ├── openai.py # GPT-4o 适配器│ ├── gemini.py # Gemini 适配器│ ├── glm4v.py # GLM-4V 适配器│ └── qwen_vl.py # Qwen-VL 适配器├── tools/│ ├── capture.py # 图像预处理│ └── analyzers.py # 场景化分析工具├── model_router.py # 模型路由器└── requirements.txt### 3.2 模型适配器：统一的抽象基类python# models/base.pyfrom abc import ABC, abstractmethodfrom dataclasses import dataclassfrom typing import Optionalimport base64@dataclassclass VisionRequest: """统一的视觉分析请求""" image_data: bytes prompt: str system_prompt: Optional[str] = None max_tokens: int = 1024 temperature: float = 0.7@dataclassclass VisionResponse: """统一的分析响应""" content: str model: str tokens_used: int latency_ms: floatclass BaseVisionModel(ABC): """所有多模态模型的抽象基类""" @abstractmethod async def analyze(self, request: VisionRequest) -> VisionResponse: """对图像执行视觉分析""" pass @abstractmethod def supports(self, mime_type: str) -> bool: """检查是否支持该图像格式""" pass @staticmethod def encode_image(image_data: bytes, mime_type: str = "image/jpeg") -> str: """编码为 base64 data URL（大多数模型通用）""" b64 = base64.b64encode(image_data).decode("utf-8") return f"data:{mime_type};base64,{b64}"### 3.3 OpenAI 适配器实现python# models/openai.pyfrom openai import AsyncOpenAIfrom .base import BaseVisionModel, VisionRequest, VisionResponseclass OpenAIVision(BaseVisionModel): def __init__(self, api_key: str, model: str = "gpt-4o"): self.client = AsyncOpenAI(api_key=api_key) self.model = model async def analyze(self, request: VisionRequest) -> VisionResponse: import time start = time.time() image_url = self.encode_image(request.image_data) messages = [] if request.system_prompt: messages.append({"role": "system", "content": request.system_prompt}) messages.append({ "role": "user", "content": [ {"type": "text", "text": request.prompt}, {"type": "image_url", "image_url": {"url": image_url}} ] }) response = await self.client.chat.completions.create( model=self.model, messages=messages, max_tokens=request.max_tokens, temperature=request.temperature, ) return VisionResponse( content=response.choices[0].message.content, model=self.model, tokens_used=response.usage.total_tokens, latency_ms=(time.time() - start) * 1000, ) def supports(self, mime_type: str) -> bool: return mime_type in ("image/jpeg", "image/png", "image/webp", "image/gif")### 3.4 Gemini 适配器（直接传 bytes，免 base64）python# models/gemini.pyimport google.generativeai as genaifrom .base import BaseVisionModel, VisionRequest, VisionResponseclass GeminiVision(BaseVisionModel): def __init__(self, api_key: str, model: str = "gemini-2.0-flash"): genai.configure(api_key=api_key) self.model = genai.GenerativeModel(model) async def analyze(self, request: VisionRequest) -> VisionResponse: import time start = time.time() # Gemini 直接支持 bytes，免去 base64 编解码开销 response = await self.model.generate_content_async([ request.prompt, {"mime_type": "image/jpeg", "data": request.image_data} ]) return VisionResponse( content=response.text, model=self.model.model_name, tokens_used=response.usage_metadata.total_token_count, latency_ms=(time.time() - start) * 1000, ) def supports(self, mime_type: str) -> bool: return mime_type in ("image/jpeg", "image/png", "image/webp")### 3.5 模型路由器：按任务智能调度python# model_router.pyfrom enum import Enumfrom typing import Dictclass TaskType(Enum): GENERAL = "general" # 通用场景理解 FOOD = "food" # 饮食识别 PLANT = "plant" # 植物识别 MATH = "math" # 数学/逻辑题目 OUTFIT = "outfit" # 穿搭分析 SPATIAL = "spatial" # 空间布局 SOCIAL = "social" # 社交媒体文案class ModelRouter: def __init__(self): self.models: Dict[str, BaseVisionModel] = {} # 每个任务类型的模型偏好列表（按优先级排） self.task_preferences: Dict[TaskType, list[str]] = { TaskType.GENERAL: ["gpt-4o", "gemini-2.0-flash"], TaskType.FOOD: ["gemini-2.0-flash", "gpt-4o"], TaskType.PLANT: ["gpt-4o", "qwen-vl-max"], TaskType.MATH: ["gpt-4o", "glm-4v"], TaskType.OUTFIT: ["gpt-4o", "glm-4v"], TaskType.SPATIAL: ["gemini-2.0-flash", "gpt-4o"], TaskType.SOCIAL: ["qwen-vl-max", "glm-4v"], } def register(self, name: str, model: BaseVisionModel): self.models[name] = model async def route( self, task: TaskType, request: VisionRequest ) -> VisionResponse: """按任务类型选择最优模型，失败自动 fallback""" candidates = self.task_preferences.get(task, ["gpt-4o"]) last_error = None for model_name in candidates: if model_name not in self.models: continue try: model = self.models[model_name] if not model.supports("image/jpeg"): continue return await model.analyze(request) except Exception as e: last_error = e continue raise RuntimeError( f"All models failed for task {task.value}. " f"Last error: {last_error}" )路由器设计的要点：- 任务感知路由：GPT-4o 解数学题最强，但 Gemini 食物识别准确率高且便宜- 自动 fallback：首选模型异常时自动切换备选- 可扩展：新增模型只需实现 BaseVisionModel 并注册到路由器### 3.6 图像预处理：省钱省时的关键一步python# tools/capture.pyfrom PIL import Imageimport ioclass ImagePreprocessor: MAX_DIMENSION = 2048 # 最大边长 JPEG_QUALITY = 85 @classmethod def process(cls, image_data: bytes) -> tuple[bytes, str]: """预处理：缩放 + 格式统一""" img = Image.open(io.BytesIO(image_data)) # RGBA → RGB if img.mode in ("RGBA", "P"): img = img.convert("RGB") # 等比缩放 w, h = img.size scale = cls.MAX_DIMENSION / max(w, h) if scale < 1.0: img = img.resize( (int(w * scale), int(h * scale)), Image.LANCZOS ) # 输出 JPEG buf = io.BytesIO() img.save(buf, format="JPEG", quality=cls.JPEG_QUALITY) return buf.getvalue(), "image/jpeg"实测数据（4032×3024 原图）：| 处理方式 | 文件大小 | GPT-4o Token 消耗 | 延迟 ||----------|---------|------------------|------|| 原始 base64 | ~2.8 MB | ~1100 tokens | ~3.2s || 压缩到 2048px | ~180 KB | ~280 tokens | ~1.4s || 压缩到 1024px | ~60 KB | ~85 tokens | ~0.9s |2048px 是性价比最优解：延迟降 56%，token 省 75%，细节损失不明显。### 3.7 MCP Server：组装所有模块python# server.pyfrom fastmcp import FastMCPfrom tools.capture import ImagePreprocessorfrom model_router import ModelRouter, TaskTypefrom models.openai import OpenAIVisionfrom models.gemini import GeminiVisionfrom models.glm4v import GLM4Visionfrom models.qwen_vl import QwenVLVisionfrom models.base import VisionRequestimport base64mcp = FastMCP("视觉理解服务")# 初始化路由器router = ModelRouter()router.register("gpt-4o", OpenAIVision(api_key=***"OPENAI_API_KEY"))router.register("gemini-2.0-flash", GeminiVision(api_key=***"GEMINI_API_KEY"))router.register("glm-4v", GLM4Vision(api_key=***"ZHIPU_API_KEY"))router.register("qwen-vl-max", QwenVLVision(api_key=***"DASHSCOPE_API_KEY"))@mcp.tool()async def analyze_image(image_base64: str, question: str = "请详细描述图片内容") -> str: """通用图像分析""" raw = base64.b64decode(image_base64) processed, _ = ImagePreprocessor.process(raw) resp = await router.route(TaskType.GENERAL, VisionRequest( image_data=processed, prompt=question, system_prompt="你是专业的视觉分析助手，请仔细观察并给出准确描述。", )) return ( f"📷 分析结果（模型：{resp.model}）:\n\n{resp.content}\n\n" f"---\n⏱ {resp.latency_ms:.0f}ms | 📊 {resp.tokens_used} tokens" )@mcp.tool()async def analyze_food(image_base64: str) -> str: """饮食分析：识别食物，返回营养信息和热量估算""" raw = base64.b64decode(image_base64) processed, _ = ImagePreprocessor.process(raw) resp = await router.route(TaskType.FOOD, VisionRequest( image_data=processed, prompt="识别所有食物，列出名称、主要食材、预估热量范围，给出营养均衡评价。用表格输出。", system_prompt="你是专业营养分析助手，客观准确分析食物。", )) return resp.content@mcp.tool()async def solve_problem(image_base64: str) -> str: """解题助手：识别数学/逻辑题，给出详细解题过程""" raw = base64.b64decode(image_base64) processed, _ = ImagePreprocessor.process(raw) resp = await router.route(TaskType.MATH, VisionRequest( image_data=processed, prompt="请阅读题目，给出完整解题步骤和最终答案。选择题先排除错误选项。", system_prompt="你是耐心严谨的数学辅导老师，逐步推导，每一步清晰可循。", temperature=0.3, # 数学题降低温度 )) return resp.contentif __name__ == "__main__": mcp.run(transport="stdio")—## 4. 几个关键设计决策### 4.1 为什么按场景拆 Tool 而不是一个万能 Tool？一开始我做了一个 analyze_image(prompt)，让用户自由写 prompt。但实际效果很差：1. AI Host 不擅长写视觉 prompt。Claude 自己临时想的 prompt 质量飘忽，经常漏掉关键信息。2. 场景化拆分降低使用门槛。用户说「帮我看看吃了什么」，AI 自动路由到 analyze_food，不需要手动写 prompt。3. 安全边界好控制。不同场景的 system prompt 可以有不同的安全约束。Tool 设计原则：让 AI 容易用对，比给它无限灵活性更重要。### 4.2 MCP 传输图片的注意事项MCP 的 Tool 参数支持 string 类型，所以 base64 字符串能直接传。但注意：- 不要用 MCP Resources 传二进制图像。当前 Anthropic SDK 实现中 Resource 的二进制内容可能被截断- Tool 参数走 base64 string 是最稳妥的方式### 4.3 生产环境的熔断保护pythonclass ProductionRouter(ModelRouter): def __init__(self): super().__init__() self.circuit_breakers: Dict[str, int] = {} self.failure_threshold = 5 async def route(self, task, request): for name in self.task_preferences.get(task, []): if self.circuit_breakers.get(name, 0) >= self.failure_threshold: continue try: result = await self.models[name].analyze(request) self.circuit_breakers[name] = 0 # 成功则重置熔断计数 return result except Exception: self.circuit_breakers[name] = ( self.circuit_breakers.get(name, 0) + 1 ) continue—## 5. 部署：从本地到云端### 本地运行bashpip install fastmcp openai google-generativeai pillowexport OPENAI_API_KEY="sk-***"python server.pyClaude Desktop 配置（claude_desktop_config.json）：json{ "mcpServers": { "vision": { "command": "python", "args": ["/path/to/server.py"] } }}### 云端部署本地跑通后上线给别人用，核心挑战是 HTTPS 证书和 7×24 进程守护。我用的 imcp.pro，代码推送自动构建部署，无需手动管理服务器：bashgit push origin main # 自动部署拿到 MCP Server URL 后，配置改成远程模式：json{ "mcpServers": { "vision": { "url": "https://your-server.imcp.pro/sse" } }}—## 6. 总结这篇文章走完了从架构到代码的全流程：模型适配器统一多模型接口 → 任务感知路由智能调度 → 场景化 Tool 降低使用门槛 → 图像预处理优化成本。核心收获：- 用抽象基类统一多模型接入，扩展新模型只需实现接口- 任务感知路由比随机轮询效果好得多，不同模型有各自擅长的领域- 图像预处理不是可选项而是必选项，直接传原图成本高几倍- MCP Tool 设计要站在 AI 使用者的角度，而不是开发者的角度代码完整可跑，有问题欢迎在评论区交流。—*上一篇入门：《最近在折腾 MCP，顺便聊聊这玩意到底是什么》**MCP 官方文档：modelcontextprotocol.io**FastMCP：github.com/jlowin/fastmcp*MCP 部署平台：imcp.pro