File System MCP Server

#!/usr/bin/env python3 """ 개선된 Ollama + MCP 브리지 Ollama가 MCP 도구 호출 JSON을 생성하도록 명확한 프롬프트를 사용합니다. """ import requests import json import re import time import sys # 서비스 URL 설정 OLLAMA_URL = "http://localhost:11434/api/generate" MCP_URL = "http://localhost:8000/invoke_tool" def check_services(): """모든 서비스가 실행 중인지 확인""" services = { "Backend API": "http://localhost:8001/health", "MCP Server": "http://localhost:8000/health", "Ollama": "http://localhost:11434/api/tags", } print("🔍 서비스 상태 확인 중...") for name, url in services.items(): try: response = requests.get(url, timeout=5) if response.status_code == 200: print(f"✅ {name}: 실행 중") else: print(f"❌ {name}: 응답 오류 ({response.status_code})") return False except requests.exceptions.RequestException as e: print(f"❌ {name}: 연결 실패 - {e}") return False print("✅ 모든 서비스가 정상 실행 중입니다!\n") return True def create_mcp_prompt(user_request): """MCP 도구 호출을 위한 명확한 프롬프트 생성""" return f"""당신은 파일 시스템 작업을 수행하는 AI 어시스턴트입니다. 사용자의 요청을 분석하여 적절한 MCP 도구를 호출해야 합니다. 사용 가능한 MCP 도구들: - list_directory: 디렉토리 목록 보기 (inputs: {{"path": "경로"}}) - read_file: 파일 읽기 (inputs: {{"file_path": "파일경로"}}) - write_file: 파일 쓰기 (inputs: {{"file_path": "파일경로", "content": "내용"}}) - get_file_info: 파일 정보 확인 (inputs: {{"path": "경로"}}) - create_directory: 디렉토리 생성 (inputs: {{"dir_path": "디렉토리경로"}}) - delete_file: 파일/디렉토리 삭제 (inputs: {{"path": "경로"}}) - copy_file: 파일 복사 (inputs: {{"source_path": "원본경로", "destination_path": "대상경로"}}) 사용자 요청: {user_request} 위 요청을 처리하기 위해 적절한 MCP 도구를 JSON 형식으로 응답하세요. 응답은 반드시 다음 형식의 JSON만 포함해야 합니다: {{"tool": "도구명", "inputs": {{매개변수들}}}} 예시: - "파일 목록 보여줘" → {{"tool": "list_directory", "inputs": {{"path": "."}}}} - "test.txt 읽어줘" → {{"tool": "read_file", "inputs": {{"file_path": "test.txt"}}}} - "test.txt에 Hello 써줘" → {{"tool": "write_file", "inputs": {{"file_path": "test.txt", "content": "Hello"}}}} JSON만 응답하고 다른 설명은 하지 마세요.""" def ask_ollama(prompt, model="mistral"): """Ollama에 프롬프트를 보내고 응답을 받습니다""" payload = {"model": model, "prompt": prompt, "stream": False} try: response = requests.post(OLLAMA_URL, json=payload, timeout=30) response.raise_for_status() return response.json()["response"] except requests.exceptions.RequestException as e: print(f"❌ Ollama 요청 실패: {e}") return None def extract_tool_call(text): """LLM 응답에서 MCP 도구 호출 JSON을 추출합니다""" # 먼저 전체 텍스트가 JSON인지 확인 text = text.strip() if text.startswith("{") and text.endswith("}"): try: return json.loads(text) except json.JSONDecodeError: pass # JSON 패턴 매칭 (여러 패턴 시도) patterns = [ r'\{[\s\S]*?"tool"\s*:\s*".+?"[\s\S]*?\}', # 기본 패턴 r"```json\s*(\{[\s\S]*?\})\s*```", # 코드 블록 내 JSON r'(\{[\s\S]*?"tool"[\s\S]*?\})', # 단순 JSON ] for pattern in patterns: match = re.search(pattern, text, re.IGNORECASE) if match: try: json_str = match.group(1) if len(match.groups()) > 0 else match.group(0) return json.loads(json_str) except json.JSONDecodeError: continue return None def call_mcp_api(tool_json): """MCP API를 호출하여 도구를 실행합니다""" try: response = requests.post(MCP_URL, json=tool_json, timeout=30) response.raise_for_status() return response.json() except requests.exceptions.RequestException as e: print(f"❌ MCP API 호출 실패: {e}") return None def format_result(result): """결과를 보기 좋게 포맷팅합니다""" if isinstance(result, dict): if "content" in result: return result["content"] elif "files" in result: files = result["files"] if isinstance(files, list): return "\n".join([f"📁 {f}" for f in files]) else: return str(files) elif "message" in result: return result["message"] else: return json.dumps(result, indent=2, ensure_ascii=False) else: return str(result) def main(): """메인 실행 함수""" print("🤖 개선된 Ollama + MCP 브리지 시스템") print("=" * 50) # 서비스 상태 확인 if not check_services(): print("❌ 일부 서비스가 실행되지 않았습니다.") print("다음 명령어로 서비스를 시작하세요:") print("1. python backend_api.py") print("2. python mcp_server.py") print("3. ollama serve") sys.exit(1) print("💡 사용 가능한 명령 예시:") print("- '현재 디렉토리 파일 목록 보여줘'") print("- 'test_source.txt 파일 읽어줘'") print("- '새 파일 test.txt에 Hello World 써줘'") print("- 'test_source.txt 파일 정보 보여줘'") print("- '새 폴더 backup 만들어줘'") print("- 'test_source.txt를 test_copy.txt로 복사해줘'") print("- 'test_copy.txt 파일 삭제해줘'") print("=" * 50) while True: try: user_input = input("\n👤 사용자: ").strip() if user_input.lower() in ["quit", "exit", "종료", "q"]: print("👋 브리지 시스템을 종료합니다.") break if not user_input: continue # MCP 도구 호출을 위한 명확한 프롬프트 생성 mcp_prompt = create_mcp_prompt(user_input) # Ollama에 프롬프트 전송 print("🤖 Ollama 처리 중...") llm_response = ask_ollama(mcp_prompt) if not llm_response: print("❌ Ollama 응답을 받지 못했습니다.") continue print(f"📝 Ollama 응답:\n{llm_response}\n") # MCP 도구 호출 추출 tool_call = extract_tool_call(llm_response) if tool_call: print(f"🔧 MCP 도구 호출: {tool_call['tool']}") print(f"📋 매개변수: {tool_call['inputs']}") # MCP API 호출 result = call_mcp_api(tool_call) if result: print(f"✅ 실행 결과:\n{format_result(result)}") else: print("❌ MCP 도구 실행 실패") else: print("❌ 실행할 MCP 도구 명령을 찾지 못했습니다.") print("Ollama 응답에서 JSON을 추출할 수 없습니다.") except KeyboardInterrupt: print("\n👋 브리지 시스템을 종료합니다.") break except Exception as e: print(f"❌ 오류 발생: {e}") if __name__ == "__main__": main()