Local LLM(Llama)을 활용한 MCP Server 구축

 

현재 로컬 LLM은 서버에 설치되어 있으며,

MCP는 각자의 노트북에서도 실행이 가능해야 하기 때문에 로컬 환경에서 동작 가능한 구조를 갖춰야 한다.
이를 위해 로컬 LLM을 API 형태로 배포하고, 로컬에서 해당 LLM을 호출해 MCP 서버를 실행하는 방식을 구성했다.

 

 

📌 전체 Flow

1. 서버: LLM이 설치되어 있음 → Docker로 띄워 외부에서 접근 가능하도록 설정
2. 로컬: MCP 서버 설치 → Docker로 배포된 LLM API를 호출하여 사용

 

🖥 서버 환경 설정

1️⃣ 가상환경 생성

 

버전 충돌을 방지하기 위해 conda 환경에서 별도의 가상환경을 만든다.

conda create -n 가상환경명
conda activate 가상환경명
conda deactivate
conda remove -n 가상환경명 -all

 

mcp-use는 Python 3.11 혹은 3.12 버전에서만 설치 가능하기 때문에 다음과 같이 생성한다.

conda create -n mcp_server python=3.12

 

2️⃣ LLM API화

 

main.py에 다음과 같이 API endpoint를 정의해둔다.

@app.post("/chat")

 

여기에 assistant라는 파라미터를 추가하여 용도별로 다른 함수를 호출하도록 구성했다.
예를 들어, "llm_f_mcp"라는 assistant를 등록하여 MCP용 LLM 엔드포인트가 연결되도록 설정했다.

 

 

 

💻 로컬 환경 설정

1️⃣ 프로젝트 구조

mcp/
├── config/
│   ├── config.json
├── hwp_mcp/                 
├── llm/
│   └── docker_llm_wrapper.py
├── mcp_tools/
│   └── tool_loader.py
├── ui/
│   └── chat_ui_streamlit.py

 

 

2️⃣ MCP 환경 구축

 

로컬에서도 가상환경을 만들어 MCP를 실행한다. (Windows 기준)

python -m venv mcp-env
Set-ExecutionPolicy -Scope Process -ExecutionPolicy Bypass
.\mcp-env\Scripts\Activate.ps1

 

 

3️⃣ MCP-Use 설치

 

mcp-use는 MCP 서버와 커스텀 에이전트를 간단히 연결할 수 있는 도구이다.

pip install mcp-use

 

GitHub 소스에서 직접 설치하려면:

git clone https://github.com/pietrozullo/mcp-use.git
cd mcp-use
pip install -e .

 

 

 

 

 

🦙 Docker LLM Wrapper 구성

로컬에서는 docker_llm_wrapper.py를 통해 서버에 설치된 LLM을 호출한다.

from mcp_use import MCPClient
import requests
from typing import Sequence, Optional
from pydantic import Field
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import HumanMessage, AIMessage
from langchain_core.outputs import ChatResult, ChatGeneration
from langchain_core.tools import BaseTool

class DockerRemoteLLMWrapper(BaseChatModel):
    api_url: str = Field(...)
    assistant: str = Field(default="llm_f_mcp")
    _bound_tools: Optional[Sequence[BaseTool]] = None
    
    def _generate(self, messages, stop=None, run_manager=None) -> ChatResult:
        prompt = "\n".join([msg.content for msg in messages if isinstance(msg, HumanMessage)])
        payload = {
            "query": prompt,
            "assistant": self.assistant,
            "temperature": 0,
            "top_p": 0.9,
            "max_seq_len": 2048,
            "max_batch_size": 16,
            "max_gen_len": 256
        }
        response = requests.post(self.api_url, data=payload)
        result_text = response.json()["response"]
        return ChatResult(generations=[ChatGeneration(message=AIMessage(content=result_text))])
    
    def bind_tools(self, tools: Sequence[BaseTool]) -> BaseChatModel:
        self._bound_tools = tools
        return self
    
    @property
    def _llm_type(self) -> str:
        return "custom_remote_llm"

 

mcp-use는 ChatResult를 AIMessage 형태로 받기 때문에 반드시 일치시켜야 한다.
또한 ChatGPT나 Claude와 달리 로컬 LLM에는 tool-calling 기능이 내장되어 있지 않다.
따라서 MCP 서버를 정확히 호출하기 위해서는 별도의 tool-calling 로직이 필요하다.

 

 

📌  For Better Tool-Calling

1️⃣ Tool Loader 작성 (tool_loader.py)

 

config 파일로부터 MCP 서버 설정을 읽고, transport 방식에 따라 다르게 처리한다.

import json
import os

def load_mcp_config():
    with open("config 경로", 'r', encoding='utf-8') as f:
        return json.load(f)

def create_server_config():
    config = load_mcp_config()
    server_config = {}
    for name, data in config["mcpServers"].items():
        if "command" in data:
            server_config[name] = {"command": data["command"], "args": data.get("args", []), "transport": "stdio"}
        elif "url" in data:
            server_config[name] = {"url": data["url"], "transport": "sse"}
    return server_config

 

 

 

 

2️⃣ Subprocess 관련 오류 해결 (chat_ui_streamlit.py)

 

Windows 환경에서는 stdio transport 방식이 subprocess 루프에서 shutdown 오류를 유발할 수 있다.
다음 코드로 asyncio 정책을 재정의하면 문제를 방지할 수 있다.

import asyncio
import asyncio.base_subprocess as _bsp
import asyncio.proactor_events as _pro
import warnings
import sys

if sys.platform.startswith("win"):
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())

def _safe_repr(self):
    try:
        return object.__repr__(self)
    except Exception:
        return "<closed transport>"

_bsp.BaseSubprocessTransport.__repr__ = _safe_repr
_pro._ProactorBasePipeTransport.__repr__ = _safe_repr
_bsp.BaseSubprocessTransport.__del__ = lambda self: None
_pro._ProactorBasePipeTransport.__del__ = lambda self: None

warnings.filterwarnings("ignore", category=ResourceWarning)

 

 

 

 

3️⃣ System Message와 Chat History 활용

 

LLM이 맥락을 기억하도록 system message와 대화 이력을 함께 전달한다.

SYSTEM_MESSAGE = """
You are an intelligent AI assistant that can communicate with various tools and systems.
You can use the tools to perform tasks and provide information to the user.

Rules:
- Only access directories that are explicitly configured in the MCP server.
- All the directories listed in the MCP server are accessible.
- Your response should be in Korean.
"""

 

 

 

 

4️⃣ Tool 사용 메시지 변환

 

로컬 LLM에서는 user는 HumanMessage, assistant는 AIMessage 타입을 사용해야 한다.

def convert_to_langchain_messages(chat_history):
    result = [SYSTEM_MESSAGE]
    for m in chat_history:
        if m["role"] == "user":
            result.append(HumanMessage(content=m["content"]))
        elif m["role"] == "assistant":
            result.append(AIMessage(content=m["content"]))
    return result

 

 

 

 

5️⃣ MCP Agent 구성

def get_agent():
    api_url = "<api 주소>"
    model = DockerRemoteLLMWrapper(api_url=api_url, assistant="llm_f_mcp")
    server_config = create_server_config()
    return MultiServerMCPClient(server_config), model

 

Streamlit 기반 UI에서는 Claude처럼 LLM이 사용하는 Tool 정보를 사이드바에 시각화했다.

 

async def run_agent(chat_history : list[dict]) -> str:
    client, model = get_agent()
    async with client:
        agent = create_react_agent(model, client.get_tools())
        #input_message = HumanMessage(content=query)
        messages = convert_to_langchain_messages(chat_history)
        response = await agent.ainvoke({"messages": messages})

        tool_outputs = []
        tool_names = []
        final_answer = None
        all_ai_messages = []
        # 전체 message 리스트 가져오기
        messages = response.get("messages", [])
        print(f"[messages]: {messages}")

        for msg in messages:
            if isinstance(msg, ToolMessage):
                tool_outputs.append(msg.content)
                tool_names.append(msg.name if hasattr(msg, 'name') else 'unknown tool')
                #print("[ToolMessage]:", msg.content)

            elif isinstance(msg, AIMessage):
                if msg.content.strip():
                    all_ai_messages.append(msg.content)
                
                # additional_kwargs에서 tool_calls 정보 확인
                if hasattr(msg, 'additional_kwargs') and 'tool_calls' in msg.additional_kwargs:
                    for tool_call in msg.additional_kwargs['tool_calls']:
                        if 'function' in tool_call and 'name' in tool_call['function']:
                            # 이미 수집된 툴 이름과 출력이 대응되도록 순서 유지
                            if len(tool_names) > len(tool_outputs):
                                continue
                            tool_names.append(tool_call['function']['name'])

        if tool_outputs:
            with st.sidebar:
                st.markdown("### 🧪 중간 Tool 출력")
                for i, (name, out) in enumerate(zip(tool_names, tool_outputs)):
                    st.markdown(f"**Step {i+1}: {name}**")  # 툴 이름 표시
                    st.code(out, language="text")
        
        if all_ai_messages:
            final_answer = all_ai_messages[-1]
        else:
            final_answer = "[No response]"
        
        print("[최종 AI 응답]:", final_answer)
                
        return final_answer

 

 

---

🪄 For My Understanding

이 코드에서는 Llama 모델을 LangChain Custom Wrapper로 감싸 LangChain에서 사용할 수 있게 했다.
@property 데코레이터는 객체의 메서드를 속성처럼 사용할 수 있도록 해준다.

 

class Rectangle:
    def __init__(self, width, height):
        self.width = width
        self.height = height
    
    @property
    def area(self):
        return self.width * self.height

r = Rectangle(10, 5)
print(r.area)  # -> 50

 

r.area()가 아니라 r.area로 접근해도 내부적으로는 함수가 호출된다.
이 덕분에 Llama Wrapper 내부에서 _llm_type을 변수처럼 접근해도 실제로는 메서드가 실행되는 구조가 된다.