自媒體創作場景實踐｜通義千問3+MCP=一切皆有可能

2025-06-05 03:20 阿里雲開發者

前言
本方案接入了最新開源的Qwen3-235B-A22B模型，在以往的MCP文章中，我們介紹了MCP的概念原理，今天這篇文章將結合實際場景，從使用者角度出發，思考MCP在未來AI場景中更深度的用途和作用。去探索MCP的能做什麼，以及MCP的能力範圍是什麼。
本文中，將使用通義千問大語言模型透過MCP的多個Servers和部署在阿里雲自研GPU叢集上的ComfyUI Server進行互動，實現打破AI雲服務的資料孤島的訴求。
最終能夠做到透過大模型讓ComfyUI服務獲取本地、網路、資料庫等資料來源的資料，讓ComfyUI生產的影像、影片傳輸到本地、雲上儲存、資料庫，甚至是自動釋出到微博、抖音、小紅書、微信公眾號等媒體社交平臺，方案架構如下：

透過上面的方案架構圖可以看到，透過MCP可以很便捷的接入多個MCP Server讓大模型排程、協作，相比多Agent的方案，可擴充套件性、易用性都很高。
實踐方案
場景介紹
從本地儲存prompt的檔案中讀取內容，然後將prompt內容轉換成圖片，再將生成的圖片下載到本地的images資料夾中。最後再將這些圖片配合描述生成小紅書文案，釋出到小紅書中。
  • MCP Server
    • FileSystem(現有可用)
    • 小紅書(個人開發部署)
    • ComfyUI(個人開發部署)
  • 大模型
    • Qwen3-235B-A22B
  • prompt內容
一隻小狗在草地上奔跑一群小朋友在歡樂的玩耍寫實風格的山水畫
效果展示
最終效果
筆記最終成功釋出到小紅書中

環境準備
Python
Python 3.10版本及以上
MCP
MCP環境準備參考官方文件https://modelcontextprotocol.io/quickstart/server#set-up-your-environment
ComfyUI
服務可以部署在本地，也可以是雲端。具體安裝部署請參考官方文件https://github.com/comfyanonymous/ComfyUI
實踐部署
通義Qwen3接入

前往百鍊申請AccessKey [1]

VSCode + CLine
1. VSCode安裝CLine外掛
2. CLine接入Qwen3介面
  • 選擇“OpenAI Compatible”
  • Base URL新增：https://dashscope.aliyuncs.com/compatible-mode/v1
  • Model ID:qwen3-235b-a22b

MCP Server

FileSytem

參考百鍊MCP Filesystem Server[2]

Filesystem MCP Server 功能概述

  • 安全的終端命令執行：允許在受控環境中執行命令而不影響系統的安全性。
  • 目錄導航：支援對目錄結構的查詢和瀏覽。
  • 檔案系統操作：包括建立、讀取、更新和刪除檔案等基本檔案操作功能。
ComfyUI MCP Server

程式碼

1. 本server實現程式碼中，使用了固定的workflow，可以按需替換成自己需要的workflow；
2. 只針對文生圖場景的實現，其他場景可以按需自己新增mcp server tool；
import urllibimport uuidfrom typing importDict, Any, Listimport httpximport loggingimport websocketsfrom PIL import Imageimport iofrom fastmcp import FastMCPfrom fastmcp.prompts import UserMessageimport jsonfrom time import sleep# Configure logginglogging.basicConfig(level=logging.INFO)logger = logging.getLogger("comfy-image-mcp-server")COMFY_SERVER = "8.147.113.150:8000"CLIENT_ID = str(uuid.uuid4())# Create a basic server instancemcp = FastMCP(name="CompyImageServer")# You can also add instructions for how to interact with the servermcp_with_instructions = FastMCP(    name="HelpfulAssistant",    instructions="這個服務是用來透過comfyui生成圖片的.""呼叫generate_image_async()來非同步地生成需要的圖片.")defqueue_prompt(prompt: Dict[str, Any] = None) -> Dict[str, Any]:    url = f"http://{COMFY_SERVER}/api/prompt"    json = {"prompt": prompt,"client_id": CLIENT_ID    }try:        response = httpx.post(url=url, json=json, verify=False, trust_env=False, timeout=60.0)if response.status_code != 200:raise RuntimeError(f"Failed to queue prompt: {response.status_code} - {response.text}")return response.json()except httpx.RequestError as e:# logger.info(traceback.format_exc())raise RuntimeError(f"HTTP request failed: {e}")asyncdefdownload_image(prompt_id: str) -> bytes:    uri = f"ws://{COMFY_SERVER}/ws?clientId={CLIENT_ID}"    logger.info(f"Connecting to websocket at {uri}")asyncwith websockets.connect(uri) as websocket:whileTrue:try:                message = await websocket.recv()ifisinstance(message, str):try:                        data = json.loads(message)                        logger.info(f"Received text message: {data}")if data.get("type") == "executing":                            exec_data = data.get("data", {})if exec_data.get("prompt_id") == prompt_id:                                node = exec_data.get("node")                                logger.info(f"Processing node: {node}")if node isNone:                                    logger.info("Generation complete signal received")breakexcept:passelse:                    logger.info(f"Received binary message of length: {len(message)}")iflen(message) > 8:  # Check if we have actual image datareturn message[8:]  # Remove binary headerelse:                        logger.warning(f"Received short binary message: {message}")except websockets.exceptions.ConnectionClosed as e:                logger.error(f"WebSocket connection closed: {e}")breakexcept Exception as e:                logger.error(f"Error processing message: {e}")continuedefget_image(filename, subfolder, folder_type):    data = {"filename": filename, "subfolder": subfolder, "type": folder_type}    url_values = urllib.parse.urlencode(data)with urllib.request.urlopen("http://{}/view?{}".format(COMFY_SERVER, url_values)) as response:return response.read()defget_history(prompt_id):with urllib.request.urlopen("http://{}/history/{}".format(COMFY_SERVER, prompt_id)) as response:return json.loads(response.read())defget_image_and_download(prompt_id, path):    output_images = []whileTrue:        history = get_history(prompt_id)[prompt_id]if history['status']['status_str'] == 'success':for node_id in history['outputs']:                node_output = history['outputs'][node_id]                images_output = []if'images'in node_output:for image in node_output['images']:                        image_data = get_image(image['filename'], image['subfolder'], image['type'])                        image_bytes = Image.open(io.BytesIO(image_data))                        file_path = ""if path.endswith("/"):                            file_path = path + image['filename']else:                            file_path = path + "/" + image['filename']                        image_bytes.save(file_path)                        output_images.append(file_path)#         images_output.append(image_data)# output_images[node_id] = images_output# output_images_names[node_id] =return output_imageselse:            logger.info(f"promot {prompt_id} unfinished meta: {history}")            sleep(1)continue@mcp.prompt()defgenerate_image_request(prompt: str, style: str = "動漫風格") -> UserMessage:"""Generates a user message requesting image generation"""    content = f"生成一個comfyui的英文prompt，要求包含下面的內容: {prompt} 並且要求生成的圖片需要具有很高的質量，風格是{style}"return UserMessage(content=content)@mcp.tool()defgenerate_image_async(prompt: str = "a cat with yellow hat", width=512, height=512,seed=4787458) -> Dict[str, Any]:    workflow = {"6": {"inputs": {"text": prompt,"clip": ["30",1                ]            },"class_type": "CLIPTextEncode","_meta": {"title": "CLIP Text Encode (Positive Prompt)"            }        },"8": {"inputs": {"samples": ["31",0                ],"vae": ["30",2                ]            },"class_type": "VAEDecode","_meta": {"title": "VAE解碼"            }        },"9": {"inputs": {"filename_prefix": "ComfyUI","images": ["8",0                ]            },"class_type": "SaveImage","_meta": {"title": "儲存影像"            }        },"27": {"inputs": {"width": width,"height": height,"batch_size": 1            },"class_type": "EmptySD3LatentImage","_meta": {"title": "空Latent影像（SD3）"            }        },"30": {"inputs": {"ckpt_name": "flux1-dev-fp8.safetensors"            },"class_type": "CheckpointLoaderSimple","_meta": {"title": "Checkpoint載入器（簡易）"            }        },"31": {"inputs": {"seed": seed,"steps": 20,"cfg": 1,"sampler_name": "euler","scheduler": "simple","denoise": 1,"model": ["30",0                ],"positive": ["35",0                ],"negative": ["33",0                ],"latent_image": ["27",0                ]            },"class_type": "KSampler","_meta": {"title": "K取樣器"            }        },"33": {"inputs": {"text": "","clip": ["30",1                ]            },"class_type": "CLIPTextEncode","_meta": {"title": "CLIP Text Encode (Negative Prompt)"            }        },"35": {"inputs": {"guidance": 3.5,"conditioning": ["6",0                ]            },"class_type": "FluxGuidance","_meta": {"title": "Flux引導"            }        }    }return queue_prompt(workflow)@mcp.tool()defget_image_status_and_download_to_local(prompt_id:str, absolute_path: str = "/Users/wangrupeng/Documents/work/files/images") -> List[str]:"""get image generate status and download to local when it's generating success"""    images = get_image_and_download(prompt_id, absolute_path)return imagesif __name__ == "__main__":    mcp.run(transport="sse",host="127.0.0.1", port=9000)# create_simple_note()# print(generate_image_async("a cute girl with red hat standing on the green land"))# images = download_image_to_local("03e6da53-9779-4af8-b7a9-564f90eeea36")# print(images)

啟動

此處記住埠
fastmcp run server.py:mcp --transport sse --host 127.0.0.1 --port 9000

第三方媒體Server

1. 參考Github專案 social-auto-upload：https://github.com/dreammis/social-auto-upload，支援抖音、B站、小紅書等社交媒體；
2. 自己實現的server功能比較簡單，僅為Demo展示使用；

程式碼

import jsonimport loggingimport httpxfrom fastmcp import FastMCPfrom time import sleepfrom playwright.sync_api import sync_playwrightfrom xhs import XhsClientcookie = "a1=xxxx;"XIAOHONGSHU_SERVER = "127.0.0.1:5005"# social-auto-load服務defsign(uri, data=None, a1="", web_session=""):for _ inrange(10):try:with sync_playwright() as playwright:                stealth_js_path = "/Users/wangrupeng/Documents/dev/github/stealth_min/stealth.min.js"                chromium = playwright.chromium# 如果一直失敗可嘗試設定成 False 讓其開啟瀏覽器，適當新增 sleep 可檢視瀏覽器狀態                browser = chromium.launch(headless=True)                browser_context = browser.new_context()                browser_context.add_init_script(path=stealth_js_path)                context_page = browser_context.new_page()                context_page.goto("https://www.xiaohongshu.com")                browser_context.add_cookies([                    {'name': 'a1', 'value': a1, 'domain': ".xiaohongshu.com", 'path': "/"}]                )                context_page.reload()# 這個地方設定完瀏覽器 cookie 之後，如果這兒不 sleep 一下簽名獲取就失敗了，如果經常失敗請設定長一點試試                sleep(1)                encrypt_params = context_page.evaluate("([url, data]) => window._webmsxyw(url, data)", [uri, data])return {"x-s": encrypt_params["X-s"],"x-t": str(encrypt_params["X-t"])                }except Exception as e:# 這兒有時會出現 window._webmsxyw is not a function 或未知跳轉錯誤，因此加一個失敗重試趴            logger.warning(f"failed : {e}")passraise Exception("重試了這麼多次還是無法簽名成功")xhs_client = XhsClient(cookie, sign=sign)# Configure logginglogging.basicConfig(level=logging.INFO)logger = logging.getLogger("comfy-image-mcp-server")# Create a basic server instancemcp = FastMCP(name="XiaoHongShuServer")# You can also add instructions for how to interact with the servermcp_with_instructions = FastMCP(    name="HelpfulAssistant",    instructions="這個服務是用來發布管理和檢視小紅書筆記的")defcreate_simple_note(title: str, desc: str, images: []) -> str:    note = xhs_client.create_image_note(title, desc, images, is_private=False)return json.dumps(note, ensure_ascii=False, indent=2)@mcp.tool()defpublish_xiaohongshu_note(title: str = "", desc = "a cat", images: [] = ["/path/to/local/demo.jpg",    ]) -> str:"""publish a xiaohongshu note"""try:        url = f"http://{XIAOHONGSHU_SERVER}/create"        json = {"title": title,"desc": desc,"images": images}        response = httpx.post(url=url, json=json, verify=False, trust_env=False, timeout=60.0)return response.json()except Exception as e:returnf"error {e}"if __name__ == "__main__":    mcp.run(transport="sse",host="127.0.0.1", port=9001)
CLINE配置MCP Server
CLINE中右上角點選MCP Servers按鈕

更新cline_mcp_settings.json配置檔案
{"mcpServers":{"filesystem":{"autoApprove":["read_file","list_allowed_directories","read_multiple_files","create_directory","list_directory","directory_tree","search_files","get_file_info"],"disabled":false,"timeout":60,"command":"npx","args":["-y","@modelcontextprotocol/server-filesystem","/Users/wangrupeng/Documents/Cline/MCP/filesystem-server"],"transportType":"stdio"},"comfyui":{"autoApprove":["generate_image_async","get_image_status","get_image_status_and_download_to_local"],"disabled":false,"timeout":60,"url":"http://127.0.0.1:9001/sse","transportType":"sse"},"xiaohongshu":{"disabled":false,"timeout":60,"url":"http://127.0.0.1:9002/sse","transportType":"sse","autoApprove":["publish_xiaohongshu_note"]}}}
感悟和思考
不管是大資料還是AI，核心其實都是資料，任何只要有IO的系統(不管是生理系統，還是計算機系統)，IO過程都可以抽象為下面的模型：

我們日常生活中實際上也是在各種輸入和輸出的場景中切換，MCP實際上是藉助大模型打通了人在各種場景的上下文，打通了大模型和物理世界的互動聯絡。

商業化場景的思
1. 售賣MCP Server API
  • 對一些比較重的複雜的操作，比如商業非開源線上的產品，供應商可以封裝內部API，對客戶通MCP Server的呼叫；
2. 訂單自動化分析管理
  • 電商等平臺訂單系統API接入MCP Server，可以做到大模型自動幫客戶分析訂單，製作報表或者商家商品，靈活調整售價等等；
3. 腦機、殘疾人機械臂接入大模型
  • 有了MCP之後，可以為殘障任務定製化場景，讓他們能夠透過MCP + 大模型完成任務，遊戲、工作等；
4. 具身智慧機器人場景
  • 大模型驅動的移動機器人覆蓋路徑規劃 ；
  • 大模型賦能的手術機器人能夠理解複雜的手術環境和任務要求 ；
  • 智慧家居機器人能夠理解並執行各種家庭任務；
總之只要有API介面，MCP理論上都能夠接入，從這裡看能限制MCP應用的只有想象力了。不過這裡說的MCP能做，離做好還是有差距的，請看下面的不足分析。
不足之處
1. 大模型還不夠智慧
  • demo製作過程中嘗試過多款模型，最好的大模型依然會有失誤的情況，而且一步錯可能步步錯；
  • 精細化的任務很難處理，比如給大模型下一個透過blender建模的工作，生成的blender指令經常是錯誤的，這個也和大模型掌握的blender版本知識庫過舊有關係；
2. MCP Server有一定開發難度
  • 基於FastMCP2.0開發的程式碼量不大，但是中間有很多網路、依賴衝突等方面的坑，目前很多都需要自己解決，網上資料很少；
3. 安全風險
  • 身份驗證機制還不完善；
  • 可在本地執行有風險的操作，比如Filesystem server可能會誤修改本地檔案；
  • 經過大模型“翻譯”之後，下達給MCP的指令可能是錯的，甚至是有風險的；
  • MCP的操作很多不是“原子性”的，比如預定機票，萬一定錯了可能無法退訂，會有資金損失的風險；
  • 本地Secret資料可能被第三方mcp server伺服器收集，存在洩露風險。
當前MCP Server的開發技巧和建議
建議使用FastMCP框架:https://github.com/jlowin/fastmcp，其在MCP官方SDK的基礎上做了更好的封裝，開發更簡潔，而且FastMCP已經被官方採納，官網Demo就是用FastMCP實現的。
1. tool定義的函式和引數命名儘可能的明確 ；
2. 函式內部加tool功能的prompt註解和註釋 ；
3. 引數儘量附帶預設值，防止呼叫的時候漏掉引數等 ；
這樣mcp list tool和選擇tool的時候會對每個tool的功能有更明確的理解。
未來展望
1. 讓大模型更智慧，通義千問3開源之後我們又往前更近了一步；
2. 應用場景上會有越來越多的MCP應用場景出現，也會有越來越多的server出現；
3. 亟待解決安全類問題，不然商業場景上會存在很大不可靠性。
參考連結：
[1]https://bailian.console.aliyun.com/
[2]https://bailian.console.aliyun.com/?spm=5176.21213303.J_v8LsmxMG6alneH-O7TCPa.3.3e532f3dk98C98&scm=20140722.S_card%40%40%E4%BA%A7%E5%93%81%40%402983180._.ID_card%40%40%E4%BA%A7%E5%93%81%40%402983180-RL_%E7%99%BE%E7%82%BC-LOC_2024SPSearchCard-OR_ser-PAR1_213e364317425218525852943e3017-V_4-RE_new6-P0_0-P1_0&tab=mcp#/mcp-market/detail/Filesystem
輕鬆上手 Qwen3：最新全球開源冠軍
截至目前，Qwen3 已斬獲LiveBench、LiveCodeBench、SuperClue、Artificial Analysis等榜單的全球開源冠軍、國產模型冠軍。在阿里雲，您可以快速使用 Qwen3，最快 10 分鐘，最低 0 元。
點選閱讀原文檢視詳情。

相關文章