手把手教你将PaddleOCR-VL集成到Dify的MCP服务中-育师

手把手教你将PaddleOCR-VL集成到Dify的MCP服务中

1. 前言：AI Agent时代的工具调用新范式

在当前AI技术快速演进的背景下，AI Agent已从概念验证阶段进入工程化落地的关键时期。我们不再满足于大模型被动响应查询，而是期望其具备主动感知环境、调用外部能力、执行复杂任务的能力——如同一个真正的数字员工。

实现这一目标的核心在于“能力可插拔”与“协议标准化”。MCP（Model Calling Protocol）正是为此而生的一种轻量级、开放且面向AI Agent的服务调用协议。它允许Agent动态发现并安全调用部署在本地或内网中的各类工具服务，而无需硬编码逻辑。

本文将基于百度开源的PaddleOCR-VL-WEB镜像——一款专为文档解析设计的SOTA视觉-语言大模型，详细介绍如何将其封装为符合MCP规范的服务端，并通过Flask构建HTTP MCP Client，最终无缝集成至Dify 1.10的Agent工作流中。

当用户上传包含关键信息的PDF或截图时，Agent能够自动判断需要OCR处理，并通过MCP协议调度本地OCR引擎完成结构化解析，再将结果融入后续推理流程。这不仅是技术整合，更是迈向“感知-决策-执行”闭环的重要一步。

2. 技术架构设计与核心组件说明

2.1 整体架构概览

本方案采用分层解耦设计，确保各模块职责清晰、易于维护和扩展：

[用户输入] ↓ [Dify Agent] ↓ (HTTP POST /callTool) [Flask MCP Client] ←→ [MCP Server (BatchOcr.py)] ↓ [PaddleOCR-VL Local Web Service]

Dify Agent：负责对话理解、意图识别及流程控制。
Flask MCP Client：作为中间代理层，接收Dify请求并转发给MCP Server。
MCP Server：实现标准MCP接口，封装对PaddleOCR-VL的实际调用。
PaddleOCR-VL Web服务：提供原始OCR能力，支持多语言、复杂版式、表格公式等高精度识别。

该架构实现了完全的协议兼容性与系统隔离性，适用于金融、保险等对数据安全要求极高的场景。

2.2 核心优势分析

组件	优势
PaddleOCR-VL	支持109种语言，中文优化好；能识别文本、表格、公式、图表；资源消耗低，适合私有部署
MCP协议	解耦Agent与工具，支持动态发现、权限控制、跨平台调用
Flask Client	无需修改Dify源码即可接入；便于日志追踪、限流缓存、多服务路由

这种组合既保证了功能完整性，又兼顾了企业级系统的安全性与可运维性。

3. 环境准备与依赖配置

3.1 基础环境搭建

（1）Nginx静态资源服务

首先需配置Nginx，将本地目录暴露为HTTP访问路径，用于存放待OCR的文件：

server { listen 80; server_name localhost; location /mkcdn/ { alias /path/to/your/ocr_files/; autoindex on; } }

重启后可通过http://localhost/mkcdn/访问上传的图片/PDF文件。

（2）PaddleOCR-VL本地Web服务部署

使用提供的镜像启动服务（以4090D单卡为例）：

# 启动容器 docker run -it --gpus all \ -p 8080:8080 \ -v /local/ocr_data:/root/ocr_data \ paddleocrvl-web:latest # 进入Jupyter环境，激活conda环境并启动服务 conda activate paddleocrvl cd /root ./1键启动.sh # 监听6006端口，反向代理至8080

验证服务是否正常运行：

curl -X POST http://localhost:8080/layout-parsing \ -H "Content-Type: application/json" \ -d '{"file": "http://localhost/mkcdn/test.pdf", "fileType": 0}'

预期返回JSON格式的版面解析结果。

3.2 MCP服务端与客户端环境初始化

创建独立Python虚拟环境（推荐Python 3.13），使用uv进行包管理：

# 创建虚拟环境 conda create -n py13 python=3.13 -y conda activate py13 # 安装uv（Rust-based Python installer） powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex" # 初始化项目 uv init quickmcp cd quickmcp # 修改.project.toml 和 .python-version 为3.13 uv venv --python="your_conda_path/envs/py13/bin/python" .venv source .venv/Scripts/activate

安装所需依赖：

uv add mcp-server mcp mcp[cli] requests uv add mcp anthropic python-dotenv uv add flask flask-cors npm install @modelcontextprotocol/inspector@0.8.0

至此，MCP Server与Client的开发环境已准备就绪。

4. 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 mcp.server import Server import uvicorn from starlette.applications import Starlette from starlette.routing import Route import httpx # 日志配置 log_dir = "./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"错误: HTTP {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"]: for block in layout["prunedResult"]["parsing_res_list"]: 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服务启动

def create_starlette_app(mcp_server: Server, *, debug: bool = False) -> Starlette: sse = SseServerTransport("/messages/") async def handle_sse(request: Request): async with sse.connect_sse(request.scope, request.receive, request._send) as (read_stream, write_stream): await mcp_server.run(read_stream, write_stream, mcp_server.create_initialization_options()) return Starlette(debug=debug, routes=[ Route("/sse", endpoint=handle_sse), Mount("/messages/", app=sse.handle_post_message), ]) def run_server(): parser = argparse.ArgumentParser() parser.add_argument('--host', default='127.0.0.1') parser.add_argument('--port', type=int, default=8090) args = parser.parse_args() mcp_server = mcp._mcp_server starlette_app = create_starlette_app(mcp_server, debug=True) uvicorn.run(starlette_app, host=args.host, port=args.port) if __name__ == "__main__": run_server()

4.2 启动MCP Server

python BatchOcr.py --host 127.0.0.1 --port 8090

服务启动后监听/sse路径，支持SSE长连接通信。

5. MCP Client实现：构建Flask代理服务

5.1 核心代码（QuickMcpClient.py）

import asyncio import threading from flask import Flask, request, jsonify from mcp.client.sse import sse_client from mcp import ClientSession from contextlib import AsyncExitStack app = Flask(__name__) CORS(app) class MCPClient: def __init__(self): self.session = None self.exit_stack = AsyncExitStack() self._loop = None self._loop_thread = None self._streams_context = None self._session_context = None async def connect_to_sse_server(self, base_url: str): try: 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 True except Exception as e: logger.error(f"连接失败: {e}") return False

提供RESTful接口

@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", "message": "连接失败"}), 500 if not mcp_client.session: return jsonify({"status": "error", "message": "未连接"}), 400 tools_data = mcp_client.run_async(mcp_client.get_tools_list()) return jsonify({"status": "success", "data": tools_data}), 200 @app.route('/callTool', methods=['POST']) def call_tool(): data = request.get_json() if not data: return jsonify({"status": "error", "message": "空请求"}), 400 base_url = data.get('base_url', 'http://127.0.0.1:8090/sse') tool_name = data.get('tool_name') tool_args = data.get('tool_args', {}) if not tool_name: return jsonify({"status": "error", "message": "缺少tool_name"}), 400 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", "message": "连接失败"}), 500 result = mcp_client.run_async(mcp_client.call_tool(tool_name, tool_args)) # 解析MCP返回内容 result_data = {} if hasattr(result, 'content') and result.content: first = result.content[0] if hasattr(first, 'text'): try: result_data = json.loads(first.text) except: result_data = {"text": first.text} return jsonify({"status": "success", "data": result_data}), 200 @app.route('/health', methods=['GET']) def health_check(): return jsonify({"status": "ok", "connected": mcp_client.session is not None}), 200

5.2 启动MCP Client

python QuickMcpClient.py

服务默认监听0.0.0.0:8500，提供以下接口：

GET /health：健康检查
POST /listTools：获取可用工具列表
POST /callTool：调用指定工具

6. Dify集成与Agentic Flow设计

6.1 在Dify中配置自定义工具

登录Dify → 应用编辑 → 添加“自定义工具”
名称：OCR Parser
请求方式：POST
URL：http://<client-host>:8500/callTool

参数映射：

{ "tool_name": "ocr_files", "tool_args": { "files": [ { "file": "{{file_url}}", "fileType": "{{file_type}}" } ] } }

6.2 构建完整Agentic Flow

步骤1：判断是否需要调用工具

系统提示词（猫娘-system）：

#任务设定 1.你基于用户当前的输入看一下是否需要调用工具来辅助你完成。 2.你的返回是以这样的JSON Schema来返回： { "needCallTool": true/false }

若返回true，进入下一步。

步骤2：查询可用工具集

调用/listTools接口获取当前支持的工具元数据。

步骤3：判断工具是否匹配需求

使用LLM判断用户请求是否可由现有工具满足：

#系统可提供的工具 {{tools_list}} #用户当前的提问 {{query}} #输出 {"toolExisted": true/false}

若为false，回复礼貌拒绝；若为true，继续下一步。

步骤4：生成调用参数

将用户自然语言转换为结构化调用参数：

根据工具metadata，把用户提问转化为tool_args

输出示例：

{ "tool_name": "ocr_files", "tool_args": { "files": [ {"file": "http://localhost/mkcdn/test-1.png", "fileType": 1}, {"file": "http://localhost/mkcdn/test-1.pdf", "fileType": 0} ] } }

步骤5：调用MCP Client并返回结果

通过HTTP节点调用/callTool，并将返回的结构化文本直接输出给用户。

7. 实际运行效果与性能表现

用户输入：

请解析 http://localhost/mkcdn/ocrsample/ 下的 test-1.png 和 test-1.pdf 文件

系统行为：

自动识别需调用OCR工具
查询工具列表确认支持ocr_files
提取URL并构造调用参数
发起MCP调用，等待约2.1秒
返回合并后的结构化文本内容

优势体现：

全程无需人工干预
支持批量文件处理
输出保留原始段落结构
错误自动捕获与日志记录

8. 总结

本文详细介绍了如何将百度开源的PaddleOCR-VL模型通过MCP协议集成至Dify平台，构建一个具备自主感知与工具调用能力的AI Agent。核心价值体现在：

协议标准化：MCP使工具调用变得统一、可发现、可审计；
系统解耦：Agent与OCR服务完全分离，便于独立升级与维护；
安全可控：敏感数据处理可在内网完成，避免外泄风险；
热插拔设计：只需替换MCP Server即可切换不同OCR引擎（如DeepSeek OCR），不影响上层逻辑。

未来，随着更多能力被封装为MCP服务（TTS、RPA、数据库查询等），我们将真正构建起一个由“神经网络+协议+服务”组成的智能体生态。

获取更多AI镜像
想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。

手把手教你将PaddleOCR-VL集成到Dify的MCP服务中