AI Agent实战：通过MCP协议调用PaddleOCR-VL实现PDF与图片精准识别
1. 前言：从被动响应到主动感知的AI Agent进化
在2025年，AI Agent已不再是简单的问答机器人。我们期望它能像一位真正的数字员工，具备自主判断、调用工具、执行任务的能力。而要实现这一目标，关键在于“能力可插拔”和“协议标准化”。
MCP（Model Calling Protocol）正是为AI Agent时代设计的一种轻量级服务调用协议。它允许Agent动态发现并调用外部能力，无需硬编码逻辑。本文将带你完成一个真实生产环境中的集成案例——如何将百度开源的PaddleOCR-VL模型封装为符合MCP规范的服务，并通过Flask构建HTTP客户端，无缝接入Dify 1.10的Agent工作流。
当用户上传一张保单截图或合同PDF时，Agent能够自动识别需要OCR处理，调度本地OCR引擎完成解析，并将结构化文本融入后续推理流程。这不仅是技术整合，更是迈向“感知-决策-执行”闭环的关键一步。
2. 为什么选择MCP？重新定义AI Agent的工具调用方式
2.1 传统OCR集成方案的三大痛点
在过去，将OCR能力集成进低代码平台如Dify，通常有以下几种做法：
- 后端硬编码：检测到图片上传就直接调用OCR接口。这种方式耦合度高，无法复用于其他场景。
- Function Calling注册：利用LLM的函数调用能力定义
ocr_extract函数。但存在明显缺陷：- 函数需写死在配置中，缺乏动态发现机制；
- 多个Agent重复注册相同功能；
- 无法跨语言、跨网络调用；
- 模型升级或更换引擎时必须修改Agent逻辑。
这些方法本质上仍是“以模型为中心”的旧思路，而非“以能力为中心”的原生Agent设计理念。
2.2 MCP协议带来的四大变革
MCP是一种基于JSON-RPC风格的远程过程调用协议，专为AI Agent设计。其核心价值体现在：
- 彻底解耦：Agent与工具完全分离，工具可独立开发、部署、升级；
- 自动发现：通过
/manifest接口获取服务支持的能力列表、参数说明及调用示例； - 标准输入输出：所有调用遵循统一格式，便于日志记录、监控和重试；
- 安全隔离：可通过网关控制权限，确保敏感数据不出内网。
在金融、保险等对数据安全要求极高的行业中，这种架构尤为重要。PaddleOCR-VL作为处理客户证件、保单的核心组件，必须运行在私有环境中。MCP提供了安全、规范、可审计的调用通道。
2.3 为何采用HTTP + Flask作为MCP Client？
社区常见的MCP Client多为SDK形式，需嵌入主程序。但在Dify这类平台中，开发者无法直接修改源码。我们的解决方案是：构建一个独立的HTTP服务作为中转层。
具体流程如下：
- Dify中的Agent配置自定义工具，指向Flask服务（如
http://mcp-client:5000/call）； - 当Agent决定调用OCR时，向该URL发送标准MCP请求；
- Flask服务接收请求，转发至目标MCP Server（如
http://paddleocr-mcp:8080）； - 获取结果后按Dify要求格式返回。
这种设计的优势显而易见：
- 无需改动Dify源码
- 支持多MCP Server路由扩展
- 便于调试与日志追踪
- 符合微服务架构理念
某头部保险公司知识库系统落地此方案后，客服Agent自动处理用户上传的保单、身份证照片等材料，OCR准确率超92%，人工干预下降70%。
2.4 为什么是PaddleOCR-VL？
在众多OCR方案中，PaddleOCR-VL脱颖而出的原因包括：
- 多模态理解能力强：不仅能识字，还能理解版面结构（标题、段落、表格）、图文关系；
- 中文场景深度优化：针对发票、合同、证件等复杂文档训练，效果远超通用OCR；
- 开源免费+私有部署：无调用费用，数据不出内网，满足金融合规要求；
- 支持ONNX/TensorRT加速：推理速度快，适合高并发场景。
测试表明，对于模糊手机拍摄的保单照片，PaddleOCR-VL能准确提取“被保险人”、“保单号”、“生效日期”等字段并保留表格结构，而其他工具常出现乱码或漏识。
3. 环境准备与系统架构设计
3.1 整体技术栈与依赖
| 组件 | 技术选型 |
|---|---|
| OCR引擎 | PaddleOCR-VL-WEB（百度开源） |
| MCP Server | Python 3.13 + FastMCP |
| MCP Client | Flask + uvicorn |
| Agent平台 | Dify 1.10 |
| 文件服务 | Nginx静态资源服务器 |
3.2 关键环境搭建步骤
部署PaddleOCR-VL-WEB镜像
# 使用4090D单卡部署 conda activate paddleocrvl cd /root ./1键启动.sh # 启动6006端口Web服务访问实例列表，点击网页推理即可验证服务是否正常。
构建MCP Server与Client环境
# 创建Python 3.13虚拟环境 conda create -n py13 python=3.13 -y conda activate py13 # 安装uv包管理器 powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex" # 初始化项目 uv init quickmcp cd quickmcp修改.python-version和.project.toml中的版本号为3.13，然后创建虚拟环境：
uv venv --python="D:\utility\miniconda3\envs\py13\python.exe" .venv激活环境并安装必要依赖：
.\.venv\Scripts\activate uv add mcp-server mcp mcp[cli] requests flask flask-cors anthropic python-dotenv npm install @modelcontextprotocol/inspector@0.8.04. MCP Server实现：封装PaddleOCR-VL为标准服务能力
4.1 核心代码解析（BatchOcr.py）
import json import logging from typing import List, Dict from pydantic import BaseModel, Field from mcp.server.fastmcp import FastMCP from starlette.applications import Starlette import uvicorn # 日志初始化 log_dir = os.path.join(os.path.dirname(__file__), "logs") os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, f"BatchOcr_{datetime.now().strftime('%Y%m%d')}.log") file_handler = RotatingFileHandler(log_file, maxBytes=50*1024*1024, backupCount=30) file_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')) logging.basicConfig(level=logging.INFO, handlers=[file_handler, logging.StreamHandler()]) logger = logging.getLogger("BatchOcr")工具输入模型定义
class FileData(BaseModel): file: str = Field(..., description="文件URL地址") fileType: int = Field(..., description="文件类型: 0=PDF, 1=图片") class OcrFilesInput(BaseModel): files: List[FileData] = Field(..., description="要处理的文件列表")OCR调用核心逻辑
@mcp.tool() async def ocr_files(files: List[FileData]) -> str: OCR_SERVICE_URL = "http://localhost:8080/layout-parsing" all_text_results = [] for idx, file_data in enumerate(files): try: ocr_payload = {"file": file_data.file, "fileType": file_data.fileType} async with httpx.AsyncClient(timeout=60.0) as client: response = await client.post(OCR_SERVICE_URL, json=ocr_payload) if response.status_code != 200: all_text_results.append(f"错误: OCR服务返回{response.status_code}") continue ocr_response = response.json() text_blocks = [] if "result" in ocr_response and "layoutParsingResults" in ocr_response["result"]: for layout in ocr_response["result"]["layoutParsingResults"]: if "prunedResult" in layout and "parsing_res_list" in layout["prunedResult"]: blocks = layout["prunedResult"]["parsing_res_list"] for block in blocks: content = block.get("block_content", "") if content: text_blocks.append(content) file_result = "\n".join(text_blocks) all_text_results.append(file_result) except Exception as e: all_text_results.append(f"错误: {str(e)}") final_result = "\n".join(all_text_results) return json.dumps({"result": final_result}, ensure_ascii=False)该服务暴露/sse接口，支持SSE长连接通信，确保大文件处理过程中状态实时同步。
5. MCP Client实现：构建Flask网关对接Dify
5.1 核心代码结构（QuickMcpClient.py）
app = Flask(__name__) CORS(app) class MCPClient: def __init__(self): self.session = None self._streams_context = None self._session_context = None self._loop = None self._loop_thread = None self.anthropic = Anthropic() async def connect_to_sse_server(self, base_url: str): self._streams_context = sse_client(url=base_url) streams = await self._streams_context.__aenter__() self._session_context = ClientSession(*streams) self.session = await self._session_context.__aenter__() await self.session.initialize() return True5.2 提供三个关键API端点
健康检查/health
@app.route('/health', methods=['GET']) def health_check(): return jsonify({ "status": "ok", "connected": mcp_client.session is not None }), 200工具发现/listTools
@app.route('/listTools', methods=['POST']) def list_tools(): data = request.get_json() or {} base_url = data.get('base_url') if base_url and not mcp_client.session: success = mcp_client.run_async(mcp_client.connect_to_sse_server(base_url)) if not success: return jsonify({"status": "error"}), 500 tools_data = mcp_client.run_async(mcp_client.get_tools_list()) return jsonify({"status": "success", "data": tools_data}), 200工具调用/callTool
@app.route('/callTool', methods=['POST']) def call_tool(): data = request.get_json() tool_name = data.get('tool_name') tool_args = data.get('tool_args', {}) result = mcp_client.run_async(mcp_client.call_tool(tool_name, tool_args)) # 解析MCP返回结果 result_data = {} if hasattr(result, 'content') and isinstance(result.content, list): first_content = result.content[0] if hasattr(first_content, 'text'): try: result_data = json.loads(first_content.text) except: result_data = {"text": first_content.text} return jsonify({"status": "success", "data": result_data}), 2006. 完整工作流与Dify集成实践
6.1 启动服务
# 启动MCP Server python BatchOcr.py --host 127.0.0.1 --port 8090 # 启动MCP Client python QuickMcpClient.py # 监听8500端口6.2 Dify中的Agentic Flow设计
判断是否需要工具
- 系统提示：根据用户输入判断是否需要调用工具
- 输出格式：
{"needCallTool": true/false}
查询可用工具集
- 调用
/listTools获取当前支持的工具元数据 - 返回示例包含
ocr_files及其参数结构
- 调用
工具适配与参数生成
- 若工具存在，则将用户提问转化为标准调用参数
- 示例输出：
{ "tool_name": "ocr_files", "tool_args": { "files": [ {"file": "http://localhost/mkcdn/ocrsample/test-1.pdf", "fileType": 0}, {"file": "http://localhost/mkcdn/ocrsample/test-1.png", "fileType": 1} ] } }
执行调用并返回结果
- 发起
/callTool请求 - 接收OCR解析后的结构化文本
- 直接回复用户
- 发起
6.3 实际运行效果
用户输入：
请解析http://localhost/mkcdn/ocrsample/下的test-1.png和test-1.pdf系统在2.1秒内完成两份文件的OCR处理，合并输出清晰的结构化内容，涵盖PDF中的《朝花夕拾》文本与PNG图片中的PaddleOCR-VL简介信息。
7. 总结：迈向能力编织的AI未来
将PaddleOCR-VL封装为MCP服务并接入Dify，看似只是一个技术集成动作，实则代表了一种思维方式的根本转变——从“功能堆砌”走向“能力编织”。
未来的AI Agent将拥有无数“感官”：
- OCR是眼睛
- TTS是嘴巴
- RPA是双手
- 知识图谱是记忆
而MCP，就是连接这一切的神经网络。它让每一种能力都能被自由组合、动态调用，真正实现“按需使用、即插即用”。
更令人兴奋的是，这种架构具备天然的扩展性。只需在MCP Server中新增一个deepseek_ocr工具，无需改动任何前端逻辑，Agent就能理解“用DeepSeek OCR解析文件”这样的指令。这正是热插拔能力的魅力所在。
愿我们不仅是这些系统的使用者，更是建设者。
获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。
