mcp-vultr/chatbot.py

#!/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()