自媒体创作场景实践｜通义千问3 + MCP=一切皆有可能

import urllibimport uuidfrom typing import Dict, 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()来异步地生成需要的图片.")def queue_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}")async def download_image(prompt_id: str) -> bytes:    uri = f"ws://{COMFY_SERVER}/ws?clientId={CLIENT_ID}"    logger.info(f"Connecting to websocket at {uri}")    async with websockets.connect(uri) as websocket:        while True:            try:                message = await websocket.recv()                if isinstance(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 is None:                                    logger.info("Generation complete signal received")                                    break                    except:                        pass                else:                    logger.info(f"Received binary message of length: {len(message)}")                    if len(message) > 8:  # Check if we have actual image data                        return message[8:]  # Remove binary header                    else:                        logger.warning(f"Received short binary message: {message}")            except websockets.exceptions.ConnectionClosed as e:                logger.error(f"WebSocket connection closed: {e}")                break            except Exception as e:                logger.error(f"Error processing message: {e}")                continuedef get_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()def get_history(prompt_id):    with urllib.request.urlopen("http://{}/history/{}".format(COMFY_SERVER, prompt_id)) as response:        return json.loads(response.read())def get_image_and_download(prompt_id, path):    output_images = []    while True:        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_images        else:            logger.info(f"promot {prompt_id} unfinished meta: {history}")            sleep(1)            continue@mcp.prompt()def generate_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()def generate_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()def get_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)

import jsonimport logging
import 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服务def sign(uri, data=None, a1="", web_session=""):    for _ in range(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}")            pass    raise 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="这个服务是用来发布管理和查看小红书笔记的")def create_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()def publish_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:        return f"error {e}"if __name__ == "__main__":    mcp.run(transport="sse",host="127.0.0.1", port=9001)

商业化场景的思

不足之处

当前MCP Server的开发技巧和建议

截至目前，Qwen3 已斩获LiveBench、LiveCodeBench、SuperClue、Artificial Analysis等榜单的全球开源冠军、国产模型冠军。在阿里云，您可以快速使用 Qwen3，最快 10 分钟，最低 0 元。

点击阅读原文查看详情。