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

前言
本方案接入了最新開源的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 importDictAnyListimport 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[strAny] = None) -> Dict[strAny]:    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[strAny]:    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 元。
點選閱讀原文檢視詳情。

相關文章