Model Context Protocol(MCP)는 대형 언어 모델(LLM)의 기능을 외부 시스템과 연결하여 확장할 수 있는 표준화된 인터페이스입니다. 이를 통해 사용자 정의 비즈니스 로직이나 특정 작업을 LLM이 직접 수행할 수 있도록 만들 수 있습니다. 본 가이드에서는 Python과 FastMCP 라이브러리를 사용하여 로컬 또는 클라우드에 배포된 Dify 워크플로우를 제어하는 MCP 서버를 구축하는 방법을 설명합니다.

사전 요구 사항

개발 환경을 구축하기 위해 다음 조건이 필요합니다:

• Python 3.10 이상이 설치된 환경
• 패키지 관리 도구 uv
• 작동 중인 Dify 인스턴스 (로컬 또는 클라우드) 및 API 접근 권한

프로젝트 환경 설정

먼저 uv를 사용하여 프로젝트를 초기화하고 가상 환경을 생성합니다. 터미널에서 다음 명령어를 실행하여 프로젝트 폴더를 만들고 진입합니다.

uv init dify-mcp-bridge
cd dify-mcp-bridge

Python 3.10 이상을 사용하여 가상 환경을 활성화합니다.

uv venv --python 3.11
.venv\Scripts\activate

필요한 의존성을 설치합니다. 비동기 HTTP 요청 처리를 위해 httpx를, MCP 서버 구현을 위해 mcp 패키지를 추가합니다.

uv add mcp httpx python-dotenv

MCP 서버 코드 작성

Dify API와 통신하여 워크플로우를 실행하는 비동기 함수를 구현합니다. FastMCP를 사용하여 서버 인스턴스를 생성하고, 도구(Tool)로 등록할 함수를 정의합니다. Dify API의 타임아웃 이슈를 방지하기 위해 스트리밍 응답을 수집한 뒤 통합하여 반환하는 로직을 적용합니다.

import httpx
import json
import os
from typing import Any, Dict, Optional
from mcp.server.fastmcp import FastMCP

# MCP 서버 인스턴스 생성
dify_server = FastMCP("DifyWorkflowController")

@dify_server.tool()
async def trigger_dify_workflow(
    prompt: str,
    app_token: Optional[str] = None,
    base_url: Optional[str] = None,
    user_identifier: str = "mcp-client-user",
    response_mode: str = "streaming",
    timeout_sec: int = 60
) -> Dict[str, Any]:
    """
    Dify 워크플로우 API를 호출하여 작업을 실행하고 결과를 반환합니다.

    Args:
        prompt: LLM 또는 사용자로부터 전달받은 입력 질의
        app_token: Dify 앱의 API Key (기본값: 환경 변수 또는 로컬 기본값)
        base_url: Dify API 엔드포인트 (기본값: 로컬 호스트)
        user_identifier: 요청을 보내는 사용자의 고유 ID
        response_mode: 응답 모드 ('streaming' 권장)
        timeout_sec: 요청 제한 시간(초)

    Returns:
        워크플로우 실행 결과, 성공 여부, 에러 메시지, 토큰 사용량 등을 포함한 딕셔너리
    """
    
    # 기본 설정 값 로드 (보안상 실제 사용 시에는 환경 변수 권장)
    if app_token is None:
        app_token = os.getenv("DIFY_API_KEY", "app-xxxxxxxxxxxx")
    
    if base_url is None:
        base_url = os.getenv("DIFY_API_URL", "http://127.0.0.1/v1/chat-messages")

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

    payload = {
        "inputs": {},
        "query": prompt,
        "response_mode": response_mode,
        "user": user_identifier,
        "files": []
    }

    response_data = {
        "status": "failed",
        "output": "",
        "metadata": None,
        "error_message": None
    }

    try:
        async with httpx.AsyncClient(timeout=timeout_sec) as client:
            async with client.stream("POST", base_url, headers=headers, json=payload) as response:
                if response.status_code != 200:
                    error_text = response.text
                    try:
                        error_json = response.json()
                        error_text = error_json.get("message", error_text)
                    except:
                        pass
                    response_data["error_message"] = f"HTTP Error {response.status_code}: {error_text}"
                    return response_data

                # 스트리밍 데이터 처리
                full_content = ""
                workflow_meta = {}
                
                async for line in response.aiter_lines():
                    if not line.strip():
                        continue
                    
                    if line.startswith("data: "):
                        json_str = line[6:] # "data: " 접두사 제거
                        try:
                            event_packet = json.loads(json_str)
                            event = event_packet.get("event")
                            
                            if event == "message":
                                full_content += event_packet.get("answer", "")
                            elif event == "message_end":
                                if "metadata" in event_packet:
                                    response_data["metadata"] = event_packet["metadata"].get("usage")
                            elif event == "workflow_finished":
                                workflow_data = event_packet.get("data", {})
                                workflow_meta["status"] = workflow_data.get("status")
                                if workflow_data.get("status") == "failed":
                                    err = workflow_data.get("error") or "Workflow execution failed"
                                    response_data["error_message"] = err
                            elif event == "error":
                                response_data["error_message"] = event_packet.get("message", "Unknown API error")
                                break
                                
                        except json.JSONDecodeError:
                            continue # 파싱 오류 시 라인 건너뜀

                # 결과 집계
                if not response_data["error_message"]:
                    response_data["status"] = "success"
                    response_data["output"] = full_content

    except httpx.TimeoutException:
        response_data["error_message"] = "Request timed out during workflow execution."
    except httpx.ConnectError:
        response_data["error_message"] = "Connection failed. Check if Dify server is running."
    except Exception as e:
        response_data["error_message"] = f"Unexpected error: {str(e)}"

    return response_data

if __name__ == "__main__":
    dify_server.run(transport='stdio')

Claude Desktop 통합 설정

작성한 MCP 서버를 Claude Desktop 애플리케이션에서 사용하려면 설정 파일을 수정해야 합니다. 설정 파일의 경로는 운영체제에 따라 다르지만, Windows의 경우 %APPDATA%\Claude\claude_desktop_config.json입니다.

이 파일을 열어 mcpServers 섹션에 다음과 같이 서버 정의를 추가합니다. directory 경로는 실제 프로젝트가 위치한 폴더로 수정해야 합니다.

{
  "mcpServers": {
    "dify-bridge": {
      "command": "uv",
      "args": [
        "--directory",
        "C:\\Projects\\dify-mcp-bridge",
        "run",
        "main.py"
      ]
    }
  }
}

설정을 저장한 후 Claude Desktop을 재시작하면, LLM은 Dify에 구성된 검색 및 데이터 처리 워크플로우를 자체적인 도구로 사용할 수 있게 됩니다. 예를 들어, 사용자가 "최신 기술 뉴스를 검색해 줘"라고 요청하면 Claude는 MCP 툴을 호출하여 Dify의 SearXNG 연동 워크플로우를 실행하고, 그 결과를 바탕으로 답변을 생성합니다.