Cursor + Flux 构建高质量本地运行的文生图 MCP 服务

curl -LsSf https://astral.sh/uv/install.sh | sh

# 创建并进入项目目录
mkdir flux-image-service && cd flux-image-service

# 初始化Python项目
uv init
uv venv
source .venv/bin/activate

# 安装核心依赖
uv add "mcp[cli]" dashscope python-dotenv requests

from mcp.server.fastmcp import FastMCP
from http import HTTPStatus
from urllib.parse import urlparse, unquote
from pathlib import Path
import requests
from dashscope import ImageSynthesis
from dotenv import load_dotenv
import os
import json
from pathlib import PurePosixPath

# 加载环境变量
load_dotenv()

# 创建MCP服务器实例
mcp = FastMCP("flux-schnell-server")

# 配置输出目录
OUTPUT_DIR = Path("output")
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)

# 获取当前工作目录的绝对路径
CURRENT_DIR = Path.cwd().resolve()
print(f"MCP服务器工作目录：{CURRENT_DIR}")


# 定义文生图工具
@mcp.tool()
def generate_image(prompt: str) -> dict:
    """Generate an image from text prompt using flux-schnell/flux-dev model

    Args:
        prompt (str): 图片生成提示词

    Returns:
        dict: 包含生成状态、图片路径等信息的字典
    """
    model = "flux-dev"  # 也可选择"flux-schnell"用于更快速度
    api_key = os.getenv("DASH_API_KEY")
    if not api_key:
        error_msg = {
            "status": "error",
            "message": "DASH_API_KEY not found in config or environment variables",
        }
        print(f"错误：{json.dumps(error_msg, ensure_ascii=False)}")
        return error_msg

    try:
        print(f"开始生成图片，提示词：{prompt}")
        rsp = ImageSynthesis.call(
            model=model, prompt=prompt, size="1024*1024", api_key=api_key
        )
        print(f"API响应状态码：{rsp.status_code}")
        print(f"API响应内容：{rsp}")

        if rsp.status_code == HTTPStatus.OK:
            # 保存文件到output目录
            image_paths = []
            for result in rsp.output.results:
                file_name = PurePosixPath(unquote(urlparse(result.url).path)).parts[-1]
                # 使用相对路径存储
                relative_path = f"output/{file_name}"
                output_path = OUTPUT_DIR / file_name

                print(f"正在保存图片：{relative_path}")
                with open(output_path, "wb+") as f:
                    f.write(requests.get(result.url).content)
                # 转换为绝对路径
                absolute_path = str((CURRENT_DIR / relative_path).resolve())
                image_paths.append(absolute_path)

            result = {
                "status": "success",
                "image_paths": image_paths,
            }
            print(f"图片生成成功：{json.dumps(result, ensure_ascii=False)}")
            return result
        else:
            error_msg = {
                "status": "error",
                "status_code": rsp.status_code,
                "code": rsp.code,
                "message": rsp.message,
            }
            print(f"生成失败：{json.dumps(error_msg, ensure_ascii=False)}")
            return error_msg
    except Exception as e:
        error_msg = {"status": "error", "message": f"发生异常：{str(e)}"}
        print(f"发生异常：{error_msg}")
        return error_msg


# 运行服务器
if __name__ == "__main__":
    mcp.run()

    # 调试用代码（正式使用时注释掉）
    # prompt = "一只可爱的橘猫在阳光下打盹"
    # result = generate_image(prompt)

DASH_API_KEY=your_api_key_here

{
  "mcpServers": {
    "flux-schnell-server": {
      "command": "/full/path/to/uv",
      "args": [
        "--directory",
        "/full/path/to/project",
        "run",
        "server.py"
      ]
    }
  }
}