#!/usr/bin/env python3 """ Chatbot yang mengintegrasikan Deepseek API dengan MCP server Vultr. """ import os import sys import json import requests from typing import Dict, List, Any from dotenv import load_dotenv # Load .env file load_dotenv() # Konfigurasi DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY") if not DEEPSEEK_API_KEY: print("Error: DEEPSEEK_API_KEY tidak ditemukan di environment.") print("Silakan set DEEPSEEK_API_KEY dengan token Anda dari https://platform.deepseek.com/") sys.exit(1) MCP_SERVER_URL = "http://localhost:8000" DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions" def get_mcp_tools() -> List[Dict[str, Any]]: try: response = requests.get(f"{MCP_SERVER_URL}/tools", timeout=10) response.raise_for_status() return response.json() except Exception as e: print(f"Error mengambil tools dari MCP server: {e}") return [] def convert_tools_to_functions(tools: List[Dict[str, Any]]) -> List[Dict[str, Any]]: functions = [] for tool in tools: # DeepSeek/OpenAI expect tools in format: {"type": "function", "function": {...}} func = { "type": "function", "function": { "name": tool["name"], "description": tool.get("description", ""), "parameters": tool.get("inputSchema", {"type": "object", "properties": {}}) } } functions.append(func) return functions def call_mcp_tool(tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: try: payload = { "name": tool_name, "arguments": arguments } response = requests.post( f"{MCP_SERVER_URL}/tools/call", json=payload, timeout=30 ) response.raise_for_status() return response.json() except Exception as e: return { "error": str(e), "content": [{"type": "text", "text": f"Error memanggil tool {tool_name}: {e}"}] } def chat_with_deepseek(messages: List[Dict[str, str]], functions: List[Dict[str, Any]]) -> Dict[str, Any]: headers = { "Authorization": f"Bearer {DEEPSEEK_API_KEY}", "Content-Type": "application/json" } payload = { "model": "deepseek-chat", "messages": messages, "tools": functions if functions else None } if not functions: del payload["tools"] # DEBUG: Print payload # print(f"DEBUG PAYLOAD: {json.dumps(payload, indent=2)}") try: response = requests.post(DEEPSEEK_API_URL, headers=headers, json=payload, timeout=60) if response.status_code != 200: print(f"DEBUG RESPONSE: {response.text}") response.raise_for_status() return response.json() except Exception as e: return {"error": str(e)} def main(): print("=" * 60) print("CHATBOT DEEPSEEK-MCP VULTR") print("=" * 60) print("Chatbot ini menggunakan Deepseek API dan MCP server Vultr.") print("Anda dapat menanyakan tentang Vultr instances, DNS, dll.") print("Contoh: 'List instances saya di Vultr' atau 'Buat domain example.com'") print("Ketik 'quit' untuk keluar.") print("=" * 60) print("\nMengambil tools dari MCP server...") tools = get_mcp_tools() if not tools: print("Tidak ada tools yang ditemukan. Pastikan MCP server berjalan.") return functions = convert_tools_to_functions(tools) print(f"Ditemukan {len(functions)} tools:") for func_wrapper in functions: func = func_wrapper["function"] print(f" - {func['name']}: {func['description']}") messages = [ { "role": "system", "content": "Anda adalah asisten AI yang membantu mengelola Vultr cloud services melalui MCP server. Anda dapat menggunakan tools yang tersedia untuk melakukan operasi seperti melihat informasi akun, mengelola instances, dan mengelola DNS. Gunakan tools yang sesuai ketika pengguna meminta. Jika pengguna meminta sesuatu yang membutuhkan tool, panggil tool tersebut. Jika tidak ada tool yang sesuai, berikan jawaban umum. Gunakan bahasa Indonesia yang baik dan sopan." } ] while True: user_input = input("\nAnda: ").strip() if user_input.lower() in ['quit', 'exit', 'keluar']: print("Terima kasih! Sampai jumpa.") break if not user_input: continue messages.append({"role": "user", "content": user_input}) print("\nBot: ", end="", flush=True) response = chat_with_deepseek(messages, functions) if "error" in response: print(f"Error dari Deepseek API: {response['error']}") messages.append({"role": "assistant", "content": f"Maaf, terjadi error: {response['error']}"}) continue choice = response.get("choices", [{}])[0] message = choice.get("message", {}) if "tool_calls" in message: tool_calls = message["tool_calls"] assistant_message = { "role": "assistant", "content": message.get("content") or "", "tool_calls": tool_calls } messages.append(assistant_message) for tool_call in tool_calls: tool_name = tool_call["function"]["name"] tool_args = json.loads(tool_call["function"]["arguments"]) print(f"\n Memanggil tool: {tool_name} dengan args: {tool_args}") tool_result = call_mcp_tool(tool_name, tool_args) result_text = "" if "content" in tool_result: for content in tool_result["content"]: if content["type"] == "text": result_text += content["text"] + "\n" else: result_text = json.dumps(tool_result, indent=2) print(f" Hasil: {result_text[:200]}..." if len(result_text) > 200 else f" Hasil: {result_text}") messages.append({ "role": "tool", "tool_call_id": tool_call["id"], "content": result_text }) follow_up = chat_with_deepseek(messages, functions) if "error" in follow_up: print(f"Error follow-up: {follow_up['error']}") continue follow_up_choice = follow_up.get("choices", [{}])[0] follow_up_message = follow_up_choice.get("message", {}) if "content" in follow_up_message: print(follow_up_message["content"]) messages.append({"role": "assistant", "content": follow_up_message["content"]}) else: print("Selesai memproses tool calls.") else: assistant_response = message.get("content", "") print(assistant_response) messages.append({"role": "assistant", "content": assistant_response}) if __name__ == "__main__": main()