【LLM】如何在Cursor中调用Dify工作流

• 需要准备文本生成模型、向量模型、Rerank 模型（可选），这些都可以在 阿里云百炼平台 申请免费使用额度。

• 准备知识库文档，我这里使用 Dify 维护知识库API 文档进行展示。

• 已熟悉使用 Cursor 。

# 下载代码
git clone https://github.com/langgenius/dify.git
# 进入docker目录
cd docker
# 使用 docker compose 部署
cp .env.example .env
docker compose up -d

1. 向量检索：基于向量相似度检索方式。

2. 全文索引：基于关键词倒排索引，类似 ES。

3. 混合索引：同时执行全文检索和向量检索，最后通过 Rerank 模型对结果进行重排序。Top K 表示返回的结果数量（相似度得分最高的 K 个值）。

{
  "mcpServers": {
    "OnlineDifyServer": {
      "url": "https://api.dify.ai/mcp/server/QfH9DT07Mf33kxCX/mcp"
    }
}

import os
import uuid
from typing import Dict, Any, Optional

import httpx
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

# 定义一个 FastMCp 实例
mcp = FastMCP("DIFY_MCP_SERVER")

# DIFY API配置信息
load_dotenv()
BASE_URL = "https://api.dify.ai/v1/chat-messages"
API_KEY = os.getenv("API_KEY")

async def query_knowledge(query: str) -> Optional[Dict[str, Any]]:
    """查询知识库信息"""

    if not API_KEY:
        return {"error": "API_KEY 未配置或为空"}

    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
    }

    data = {
        "query": query,
        "user": str(uuid.uuid4()),
        "inputs": {},
        "response_mode": "blocking",
        "conversation_id": "",
    }

    try:
        async with httpx.AsyncClient(timeout=120.0) as client:
            response = await client.post(BASE_URL, headers=headers, json=data)
            response.raise_for_status()
            print(f"Response status: {response.status_code}")
            print(f"Response text: {response.text}")
            return response.json()
    except httpx.HTTPError as e:
        return {"error": f"查询知识库请求失败：{e}"}
    except Exception as e:
        return {"error": f"未知错误：{e}"}


def format_response_data(response_data: Dict[str, Any]) -> str:
    """格式化 Dify 返回数据"""

    event = response_data.get("event", "")
    if "error" == event:
        return str(response_data["message"])

    return response_data.get("answer", "")


@mcp.tool()
async def query_dify_interface_info(query: str) -> str:
    """
    用于查询 Dify 维护知识库接口文档

    参数:
    - query: 查询内容,str格式
    - 请求示例:
        1、通过文本创建文档
        2、通过文件创建文档
        3、创建空知识库
        4、知识库列表
        ...
    """
    response_data = await query_knowledge(query)
    return format_response_data(response_data)


if __name__ == "__main__":
    # 运行 MCP 服务器，默认传输协议 stdio
    mcp.run(transport="stdio")

{
  "mcpServers": {
    "LocalDifyServer": {
      "command": "uv",
      "args": [
        "--directory",
        "D:\\workspace\\project\\python\\2025\\good-agent-fastapi\\mcps",
        "run",
        "dify_mcp_server.py"
      ],
      "env": {
        "API_KEY": "填写服务密钥"
      }
    }
  }
}