728x90
반응형
로컬 Ollama에 MCP 붙이기
Claude, VSCode, Cursor 등은 자체적으로 MCP Client를 제공해주기 때문에 정말 간단하게 원하는 MCP를 붙여 사용할 수 있지만, 우리의 로컬 llama는 안타깝게도 MCP Client가 없다. 때문에 직접 만들어주거나 남들이 만들어둔 MCP Client를 사용하는 방법이 있다.
그런데 장고로 개발하는 김에 한번 MCP Client가 어떻게 굴러가는지를 알아보고자 제 2의 두뇌 커서와 함께 한번 만들어보기로 했다. (그리고 후회했다.)
MCP Client
MCP가 USB 포트와 같은 역할을 한다는 설명은 많이 들었지만, 그래서 어떻게 동작하는지에 대해서는 잘 와닿지가 않았다. 그래서 코파일럿에게 설명을 요구했다.
- 요청 수신 (from Host or Model)
- AI 모델 또는 Host가 어떤 작업(예: "날씨 알려줘")을 요청합니다.
- 이 요청은 일반적으로 자연어 또는 내부 명령 형식입니다.
- 요청 파싱 및 매핑
- MCP Client는 이 요청을 MCP에서 정의한 표준 포맷(예: JSON-RPC 기반 메시지)으로 변환합니다.
- 이때 필요한 정보:
- 어떤 MCP Server에 보낼지 (도구 이름, 기능 ID 등)
- 어떤 메서드를 호출할지 (예: getWeather)
- 어떤 파라미터를 전달할지 (예: { "location": "Seoul" })
- MCP Server로 요청 전송
- 매핑된 메시지를 해당 MCP Server로 전송합니다.
- 이때는 비동기 방식으로 처리되며, 응답을 기다리는 동안 다른 작업도 가능합니다.
- 응답 수신 및 역매핑
- MCP Server가 외부 도구를 호출하고 결과를 반환하면,
- MCP Client는 이 응답을 다시 AI 모델이 이해할 수 있는 형식으로 역매핑합니다.
- 응답 전달 (to Host or Model)
- 최종적으로 AI 모델 또는 Host에 결과를 전달하여 사용자 응답을 생성합니다.
이런식으로 굴러간다고 한다. 결국 우리의 llama 모델은 내가 요청한 내용을 포맷에 맞춰 잘 변형시키는 역할을 주된 역할로 수행하는 것이다. 어떤 API를 사용해야 하는지 AI가 결정한다는 점에서 참신했다.
MCP를 사용할 때 조금 더 정확하게 출력되게 하고자 llama3 모델에서 llama3.1:8b로 교체했다.
Python Code
#!/usr/bin/env python3
"""
Ollama MCP Client - Notion MCP Server와 Ollama를 연결하는 클라이언트
"""
import json
import asyncio
import subprocess
import requests
import os
import re
from typing import Dict, List, Any, Optional
from dotenv import load_dotenv
import logging
# 로깅 설정
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MCPClient:
def __init__(self, config_file: str = "mcp_config.json"):
"""MCP 클라이언트 초기화"""
load_dotenv()
self.config = self._load_config(config_file)
self.servers = {}
self.tools = {}
def _load_config(self, config_file: str) -> Dict:
"""MCP 설정 파일 로드"""
try:
with open(config_file, 'r') as f:
config = json.load(f)
# 환경변수로 토큰 설정
notion_token = os.getenv("NOTION_TOKEN")
if notion_token:
config["mcpServers"]["notionApi"]["env"]["OPENAPI_MCP_HEADERS"] = \
config["mcpServers"]["notionApi"]["env"]["OPENAPI_MCP_HEADERS"].replace(
"NOTION_TOKEN_PLACEHOLDER", notion_token
)
return config
except FileNotFoundError:
logger.error(f"설정 파일을 찾을 수 없습니다: {config_file}")
return {}
except json.JSONDecodeError:
logger.error(f"설정 파일 형식이 잘못되었습니다: {config_file}")
return {}
async def start_servers(self):
"""MCP 서버들을 시작"""
for server_name, server_config in self.config.get("mcpServers", {}).items():
try:
logger.info(f"서버 시작 중: {server_name}")
# npx의 전체 경로 찾기
npx_path = "/opt/homebrew/bin/npx"
if not os.path.exists(npx_path):
# 대안 경로들 시도
possible_paths = [
"/usr/local/bin/npx",
"/opt/homebrew/bin/npx",
"npx" # PATH에서 찾기
]
npx_path = None
for path in possible_paths:
try:
result = subprocess.run([path, "--version"],
capture_output=True, text=True, timeout=5)
if result.returncode == 0:
npx_path = path
break
except:
continue
if not npx_path:
logger.error("npx를 찾을 수 없습니다. Node.js가 설치되어 있는지 확인하세요.")
continue
# 환경변수 설정
env = os.environ.copy()
env.update(server_config.get("env", {}))
env["PATH"] = "/opt/homebrew/bin:/usr/local/bin:" + env.get("PATH", "")
process = await asyncio.create_subprocess_exec(
npx_path,
*server_config["args"],
env=env,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
self.servers[server_name] = process
logger.info(f"서버 {server_name} 시작됨")
except Exception as e:
logger.error(f"서버 {server_name} 시작 실패: {e}")
async def initialize_server(self, server_name: str):
"""서버 초기화 및 도구 목록 가져오기"""
if server_name not in self.servers:
logger.error(f"서버 {server_name}가 시작되지 않았습니다")
return
process = self.servers[server_name]
# 초기화 요청
init_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
},
"clientInfo": {
"name": "ollama-mcp-client",
"version": "1.0.0"
}
}
}
try:
# 초기화 요청 전송
process.stdin.write((json.dumps(init_request) + "\n").encode())
await process.stdin.drain()
# 응답 읽기
response = await process.stdout.readline()
init_response = json.loads(response.decode().strip())
if "result" in init_response:
logger.info(f"서버 {server_name} 초기화 성공")
# 도구 목록 요청
tools_request = {
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list"
}
process.stdin.write((json.dumps(tools_request) + "\n").encode())
await process.stdin.drain()
tools_response = await process.stdout.readline()
tools_data = json.loads(tools_response.decode().strip())
if "result" in tools_data:
self.tools[server_name] = tools_data["result"]["tools"]
logger.info(f"서버 {server_name}에서 {len(self.tools[server_name])}개 도구 로드됨")
except Exception as e:
logger.error(f"서버 {server_name} 초기화 실패: {e}")
async def call_tool(self, server_name: str, tool_name: str, arguments: Dict) -> Dict:
"""도구 호출"""
if server_name not in self.servers:
return {"error": f"서버 {server_name}가 시작되지 않았습니다"}
process = self.servers[server_name]
call_request = {
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
try:
process.stdin.write((json.dumps(call_request) + "\n").encode())
await process.stdin.drain()
response = await process.stdout.readline()
result = json.loads(response.decode().strip())
return result
except Exception as e:
logger.error(f"도구 호출 실패: {e}")
return {"error": str(e)}
class OllamaMCPBridge:
def __init__(self):
"""Ollama와 MCP를 연결하는 브리지"""
self.mcp_client = MCPClient()
self.ollama_url = "http://localhost:11434/api/generate"
async def setup(self):
"""MCP 서버들 시작 및 초기화"""
await self.mcp_client.start_servers()
for server_name in self.mcp_client.servers:
await self.mcp_client.initialize_server(server_name)
def ask_ollama(self, prompt: str, tools_info: str = "") -> str:
"""Ollama에 질문"""
full_prompt = f"""
사용자 요청: {prompt}
사용 가능한 Notion 도구들:
{tools_info}
위 도구들 중에서 사용자 요청을 처리하기 위해 적절한 도구를 선택하고, 필요한 인자들을 JSON 형태로 제공해주세요.
사용자가 제목이나 내용을 명시했다면, 그 내용을 정확히 반영해주세요.
응답 형식:
도구이름: [선택한 도구 이름]
제목: [사용자가 요청한 제목 또는 추천 제목]
내용: [사용자가 요청한 내용 또는 생성할 내용]
인자: {{"parent": {{"page_id": "PARENT_PAGE_ID"}}, "properties": {{"title": [{{"text": {{"content": "제목"}}}}]}}, "children": [{{"object": "block", "type": "paragraph", "paragraph": {{"rich_text": [{{"type": "text", "text": {{"content": "내용"}}}}]}}}}]}}
예시:
사용자가 "2027년 7월 6일을 제목으로 MCP 관련 내용을 넣어달라"고 요청했다면:
도구이름: API-post-page
제목: 2027년 7월 6일
내용: MCP (Model Context Protocol)는 AI 모델과 외부 도구 간의 표준화된 통신 프로토콜입니다...
인자: {{"parent": {{"page_id": "PARENT_PAGE_ID"}}, "properties": {{"title": [{{"text": {{"content": "2027년 7월 6일"}}}}]}}, "children": [{{"object": "block", "type": "paragraph", "paragraph": {{"rich_text": [{{"type": "text", "text": {{"content": "MCP (Model Context Protocol)는 AI 모델과 외부 도구 간의 표준화된 통신 프로토콜입니다..."}}}}]}}}}]}}
"""
data = {
"model": "llama3.1:8b",
"prompt": full_prompt,
"stream": False
}
try:
response = requests.post(self.ollama_url, json=data)
response.raise_for_status()
return response.json()["response"]
except Exception as e:
logger.error(f"Ollama 요청 실패: {e}")
return f"Ollama 요청 실패: {e}"
def parse_ollama_response(self, ollama_response: str) -> Dict:
"""Ollama 응답을 파싱하여 도구와 인자를 추출"""
try:
# 도구이름 추출 - 더 정확한 패턴 매칭
tool_patterns = [
r'도구이름:\s*(API-\w+(?:-\w+)*)',
r'도구이름:\s*(\w+)',
r'API-\w+(?:-\w+)*'
]
tool_name = "API-post-page" # 기본값
for pattern in tool_patterns:
tool_match = re.search(pattern, ollama_response)
if tool_match:
extracted = tool_match.group(1) if tool_match.groups() else tool_match.group(0)
if extracted.startswith('API-'):
tool_name = extracted
break
# 제목 추출
title_match = re.search(r'제목:\s*(.+)', ollama_response)
title = title_match.group(1).strip() if title_match else "Ollama가 생성한 페이지"
# 내용 추출
content_match = re.search(r'내용:\s*(.+)', ollama_response)
content = content_match.group(1).strip() if content_match else ollama_response[:2000]
# 인자 추출 - 더 유연한 JSON 파싱
args_match = re.search(r'인자:\s*(\{.*?\})', ollama_response, re.DOTALL)
if args_match:
try:
arguments = json.loads(args_match.group(1))
# 추출한 제목과 내용으로 인자 업데이트
arguments = self._update_arguments_with_content(arguments, title, content)
except json.JSONDecodeError:
arguments = self._get_custom_page_arguments(title, content)
else:
arguments = self._get_custom_page_arguments(title, content)
# 설명 추출
desc_match = re.search(r'설명:\s*(.+)', ollama_response)
description = desc_match.group(1) if desc_match else f"제목: {title}, 내용: {content[:50]}..."
return {
"tool_name": tool_name,
"arguments": arguments,
"description": description,
"action_type": self._get_action_type(tool_name),
"title": title,
"content": content
}
except Exception as e:
logger.error(f"Ollama 응답 파싱 실패: {e}")
# 기본값 반환
return {
"tool_name": "API-post-page",
"arguments": self._get_default_page_arguments(ollama_response),
"description": "파싱 실패로 기본 페이지 생성 도구 사용",
"action_type": "페이지 생성",
"title": "Ollama가 생성한 페이지",
"content": ollama_response[:2000]
}
def _update_arguments_with_content(self, arguments: Dict, title: str, content: str) -> Dict:
"""인자에 제목과 내용을 업데이트"""
parent_page_id = os.getenv("NOTION_PARENT_PAGE_ID")
# parent 업데이트
if "parent" not in arguments or not arguments["parent"]:
arguments["parent"] = {"page_id": parent_page_id}
# properties 업데이트
if "properties" not in arguments:
arguments["properties"] = {}
arguments["properties"]["title"] = [
{
"text": {
"content": title
}
}
]
# children 업데이트
if "children" not in arguments:
arguments["children"] = []
# 내용을 2000자로 제한
content = content[:2000] if len(content) > 2000 else content
arguments["children"].append({
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [
{
"type": "text",
"text": {"content": content}
}
]
}
})
return arguments
def _get_custom_page_arguments(self, title: str, content: str) -> Dict:
"""사용자 지정 제목과 내용으로 페이지 생성 인자"""
parent_page_id = os.getenv("NOTION_PARENT_PAGE_ID")
if not parent_page_id:
return {"error": "NOTION_PARENT_PAGE_ID가 설정되지 않았습니다"}
# 내용을 2000자로 제한
content = content[:2000] if len(content) > 2000 else content
return {
"parent": {"page_id": parent_page_id},
"properties": {
"title": [
{
"text": {
"content": title
}
}
]
},
"children": [
{
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [
{
"type": "text",
"text": {"content": content}
}
]
}
}
]
}
def _get_action_type(self, tool_name: str) -> str:
"""도구 이름으로부터 액션 타입 결정"""
if "post-page" in tool_name:
return "페이지 생성"
elif "retrieve-a-page" in tool_name:
return "페이지 조회"
elif "patch-page" in tool_name:
return "페이지 수정"
elif "create-a-database" in tool_name:
return "데이터베이스 생성"
elif "create-a-comment" in tool_name:
return "댓글 생성"
elif "post-search" in tool_name:
return "검색"
else:
return "기타"
def get_tools_info(self) -> str:
"""사용 가능한 도구 정보를 문자열로 반환"""
tools_info = []
for server_name, tools in self.mcp_client.tools.items():
tools_info.append(f"\n{server_name} 서버:")
for tool in tools:
tools_info.append(f" - {tool['name']}: {tool.get('description', '설명 없음')}")
return "\n".join(tools_info)
def _get_default_page_arguments(self, content: str) -> Dict:
"""기본 페이지 생성 인자"""
parent_page_id = os.getenv("NOTION_PARENT_PAGE_ID")
if not parent_page_id:
return {"error": "NOTION_PARENT_PAGE_ID가 설정되지 않았습니다"}
# 내용을 2000자로 제한
content = content[:2000] if len(content) > 2000 else content
return {
"parent": {"page_id": parent_page_id},
"properties": {
"title": [
{
"text": {
"content": "Ollama가 생성한 페이지"
}
}
]
},
"children": [
{
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [
{
"type": "text",
"text": {"content": content}
}
]
}
}
]
}
async def process_user_request(self, user_input: str):
"""사용자 요청 처리"""
# 도구 정보 가져오기
tools_info = self.get_tools_info()
# Ollama에 질문 (LLM이 도구 선택)
ollama_response = self.ask_ollama(user_input, tools_info)
print(f"🤖 Ollama 응답: {ollama_response}")
# Ollama 응답 파싱 (LLM이 선택한 도구와 인자 추출)
action = self.parse_ollama_response(ollama_response)
print(f"🔧 LLM이 선택한 도구: {action['tool_name']}")
print(f"📋 제목: {action['title']}")
print(f"📝 내용: {action['content'][:100]}...")
print(f"🎯 액션 타입: {action['action_type']}")
# MCP 도구 호출
if "error" not in action["arguments"]:
result = await self.mcp_client.call_tool(
"notionApi",
action["tool_name"],
action["arguments"]
)
print(f"📄 Notion 결과: {result}")
if "result" in result:
print(f"✅ {action['action_type']} 성공!")
print(f"📖 생성된 페이지 제목: {action['title']}")
else:
print(f"❌ {action['action_type']} 실패!")
else:
print(f"❌ 오류: {action['arguments']['error']}")
async def main():
"""메인 함수"""
bridge = OllamaMCPBridge()
print("🚀 Ollama MCP 브리지 시작...")
await bridge.setup()
print("✅ 설정 완료! 사용자 입력을 받습니다.")
print("📝 Notion에 생성할 내용을 입력하세요 (종료: 'quit'):")
print("💡 예시 명령어:")
print(" - '페이지 생성해줘'")
print(" - '데이터베이스 만들어줘'")
print(" - '댓글 달아줘'")
print(" - '검색해줘'")
while True:
try:
user_input = input("> ")
if user_input.lower() == 'quit':
break
await bridge.process_user_request(user_input)
except KeyboardInterrupt:
print("\n👋 종료합니다.")
break
except Exception as e:
print(f"❌ 오류 발생: {e}")
if __name__ == "__main__":
asyncio.run(main())
어마무시하게 길지만, 요약하면 ask_ollama에서 원하는 포맷을 예시로 주며 해당 포맷 형태로 return하면, 그에 맞춰 MCP Server로 각각의 API 요청을 보내는 것이다.
https://github.com/Lajancia/mcp-ollama
GitHub - Lajancia/mcp-ollama
Contribute to Lajancia/mcp-ollama development by creating an account on GitHub.
github.com
간단한 python 코드는 여기에 올려두었다.
하지만 문제가 많다...
제대로 돌아가기는 하지만 아직 튜닝이 되지 않은 날것의 model이기도 하고, 이미 Cursor나 VSCode에서 제공하는 멋지고 간편하고 잘 굴러가는 MCP Client를 이길 수가 없다. 무엇보다도 Notion 하나를 연결하기 위해 고려해야 할 것들이 너무 많다. 앞으로 깃허브와 figma, 및 기타 등등의 API 연결이 필요할 때 마다 일일이 MCP Client를 만들다가는 튜닝은 어림도 없을 것 같았다.
때문에 남들이 만들어둔 멋진 라이브러리를 찾던 중 방법을 찾았다.
Ollama를 활용한 로컬 MCP 서버 배포 가이드
최근 AI 기술의 발전으로 Model Context Protocol(MCP)은 AI 에이전트가 외부 도구와 서비스에 접근하여 다양한 작업을 수행할 수 있게 하는 중요한 프로토콜로 자리잡고 있습니다. 특히 로컬 환경에서 Ol
velog.io
장고를 사용하는 상황이다보니, Python에서 사용할 수 있는 MCP Client가 필요했고, 이 dolphin-mcp가 그 역할을 수행할 수 있을 것 같다.
다음 포스트에서는 한번 이 라이브러리를 장고 앱에 붙여보려고 한다.
반응형
'개발 > 바이브 코딩' 카테고리의 다른 글
| Ollama + MCP Server 붙이기 - 3 (5) | 2025.07.27 |
|---|---|
| Ollama + MCP Server 붙이기 - 2 (9) | 2025.07.20 |
| Ollama LLM 사용기 (12) | 2025.06.29 |
| MCP 도전기 (4) | 2025.06.22 |
