Muse Spark 1.1: Tutorial Coding dengan AI Meta Terbaru

python --version

Python 3.10.12

python -m venv muse-spark-env
source muse-spark-env/bin/activate  # Linux/Mac
# atau di Windows:
# muse-spark-env\Scripts\activate

touch .env

MUSE_SPARK_API_KEY=sk-muse-spark-your-api-key-here

echo ".env" >> .gitignore

pip install openai python-dotenv requests

Successfully installed openai-1.40.0 python-dotenv-1.0.1 requests-2.32.3

muse-spark-tutorial/
├── .env
├── .gitignore
├── main.py
├── chat.py
├── agent.py
├── tools.py
└── requirements.txt

pip freeze > requirements.txt

import os
from dotenv import load_dotenv

load_dotenv()

api_key = os.getenv("MUSE_SPARK_API_KEY")

if api_key:
    print(f"✅ API Key loaded: {api_key[:10]}...")
else:
    print("❌ API Key not found. Check your .env file.")

python main.py

✅ API Key loaded: sk-muse-sp...

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

# Inisialisasi client dengan endpoint Meta Model API
client = OpenAI(
    api_key=os.getenv("MUSE_SPARK_API_KEY"),
    base_url="https://api.meta.ai/v1"  # Endpoint Meta Model API
)

response = client.chat.completions.create(
    model="muse-spark-1.1",
    messages=[
        {
            "role": "system",
            "content": "Kamu adalah asisten coding yang helpful dan memberikan jawaban yang detail."
        },
        {
            "role": "user",
            "content": "Jelaskan apa itu recursion dalam programming dengan contoh Python sederhana."
        }
    ],
    temperature=0.7,
    max_tokens=1000
)

print(response.choices[0].message.content)
print(f"\n--- Token Usage ---")
print(f"Input tokens: {response.usage.prompt_tokens}")
print(f"Output tokens: {response.usage.completion_tokens}")
print(f"Total tokens: {response.usage.total_tokens}")

python chat.py

Recursion adalah teknik dalam programming di mana sebuah fungsi memanggil dirinya sendiri untuk menyelesaikan masalah. Konsep ini sangat berguna untuk masalah yang bisa dipecah menjadi sub-masalah yang lebih kecil dan serupa.

Berikut contoh sederhana dalam Python — menghitung faktorial:

```python
def factorial(n):
    if n <= 1:
        return 1
    return n * factorial(n - 1)

print(factorial(5))  # Output: 120

### Memahami Parameter Utama

Beberapa parameter penting yang perlu dipahami:

- **`model`**: Nama model, yaitu `muse-spark-1.1`
- **`messages`**: Array berisi conversation history dengan role `system`, `user`, dan `assistant`
- **`temperature`**: Kontrol kreativitas (0 = deterministik, 1 = lebih variatif)
- **`max_tokens`**: Batas maksimum token output
- **`top_p`**: Nucleus sampling, alternatif dari temperature untuk kontrol diversity

**Error umum di langkah ini:**

| Error | Penyebab | Solusi |
|---|---|---|
| `AuthenticationError` | API key salah atau tidak valid | Cek ulang API key di `.env` |
| `RateLimitError` | Limit preview tercapai | Tunggu sebentar, lalu retry dengan backoff |
| `BadRequestError` | Parameter tidak valid | Cek nama model dan format request |
| `ConnectionError` | Masalah jaringan atau endpoint salah | Verifikasi `base_url` dan koneksi internet |

## Step 4: Menggunakan Tool Calling untuk Tugas Agentic

Salah satu kekuatan utama Muse Spark 1.1 adalah **tool calling** — kemampuan model untuk memanggil fungsi eksternal sebagai bagian dari reasoning-nya. Ini fondasi dari agent yang bisa melakukan tugas nyata.

**Kenapa langkah ini penting?** Tool calling adalah apa yang membedakan chatbot biasa dari agent yang bisa menyelesaikan workflow. Dengan tool calling, model bisa mencari informasi, menjalankan kode, akses database, dan banyak lagi — semua dalam satu sesi.

### Mendefinisikan Tools

Buat file `tools.py`:

```python
import json
from datetime import datetime

# Definisi tools yang bisa dipanggil model
def get_current_time():
    """Mendapatkan waktu saat ini"""
    return json.dumps({
        "current_time": datetime.now().isoformat(),
        "timezone": "UTC"
    })

def calculate(expression: str) -> str:
    """Menghitung ekspresi matematika dengan aman"""
    try:
        # Hanya izinkan angka dan operator dasar
        allowed_chars = set("0123456789+-*/.() ")
        if not all(c in allowed_chars for c in expression):
            return json.dumps({"error": "Karakter tidak diizinkan"})
        
        result = eval(expression)
        return json.dumps({"expression": expression, "result": result})
    except Exception as e:
        return json.dumps({"error": str(e)})

def search_codebase(query: str, file_path: str = None) -> str:
    """Mencari kode dalam codebase (simulasi)"""
    results = [
        {"file": "src/utils.py", "line": 42, "match": f"Found query: {query}"},
        {"file": "src/models.py", "line": 15, "match": f"Related: {query}"}
    ]
    return json.dumps({"results": results})

# Schema tools untuk API
TOOL_DEFINITIONS = [
    {
        "type": "function",
        "function": {
            "name": "get_current_time",
            "description": "Mendapatkan waktu dan tanggal saat ini",
            "parameters": {
                "type": "object",
                "properties": {},
                "required": []
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculate",
            "description": "Menghitung ekspresi matematika. Contoh: '2 + 3 * 4'",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "Ekspresi matematika yang akan dihitung"
                    }
                },
                "required": ["expression"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "search_codebase",
            "description": "Mencari kode atau fungsi dalam codebase",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Kata kunci pencarian"
                    },
                    "file_path": {
                        "type": "string",
                        "description": "File spesifik untuk dicari (opsional)"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

# Mapping nama fungsi ke implementasi
TOOL_FUNCTIONS = {
    "get_current_time": get_current_time,
    "calculate": calculate,
    "search_codebase": search_codebase
}

import os
import json
from openai import OpenAI
from dotenv import load_dotenv
from tools import TOOL_DEFINITIONS, TOOL_FUNCTIONS

load_dotenv()

client = OpenAI(
    api_key=os.getenv("MUSE_SPARK_API_KEY"),
    base_url="https://api.meta.ai/v1"
)

def run_agent(user_message: str, max_iterations: int = 5):
    """
    Menjalankan agent loop dengan tool calling.
    Agent akan terus memanggil tools sampai mendapat jawaban final
    atau mencapai batas iterasi.
    """
    messages = [
        {
            "role": "system",
            "content": (
                "Kamu adalah agent coding yang bisa menggunakan tools untuk "
                "menyelesaikan tugas. Gunakan tools yang tersedia ketika "
                "diperlukan, dan berikan jawaban final setelah mendapatkan "
                "hasil dari tools."
            )
        },
        {"role": "user", "content": user_message}
    ]
    
    for i in range(max_iterations):
        print(f"\n--- Iterasi {i + 1} ---")
        
        response = client.chat.completions.create(
            model="muse-spark-1.1",
            messages=messages,
            tools=TOOL_DEFINITIONS,
            tool_choice="auto",
            temperature=0.3
        )
        
        message = response.choices[0].message
        
        # Jika model ingin memanggil tool
        if message.tool_calls:
            messages.append(message)
            
            for tool_call in message.tool_calls:
                function_name = tool_call.function.name
                function_args = json.loads(tool_call.function.arguments)
                
                print(f"🔧 Memanggil tool: {function_name}")
                print(f"   Argumen: {function_args}")
                
                # Eksekusi fungsi
                if function_name in TOOL_FUNCTIONS:
                    result = TOOL_FUNCTIONS[function_name](**function_args)
                else:
                    result = json.dumps({"error": f"Function {function_name} not found"})
                
                print(f"   Hasil: {result}")
                
                # Tambahkan hasil tool ke conversation
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": result
                })
        else:
            # Model memberikan jawaban final
            print(f"\n✅ Jawaban final:")
            print(message.content)
            print(f"\n--- Token Usage ---")
            print(f"Input: {response.usage.prompt_tokens}")
            print(f"Output: {response.usage.completion_tokens}")
            print(f"Total: {response.usage.total_tokens}")
            return message.content
    
    print("⚠️ Batas iterasi tercapai tanpa jawaban final")
    return None

# Test agent
if __name__ == "__main__":
    result = run_agent(
        "Berapa hasil dari (15 + 27) * 3? Lalu cari kode yang berhubungan "
        "dengan 'authentication' di codebase. Juga beri tahu saya waktu saat ini."
    )

python agent.py

--- Iterasi 1 ---
🔧 Memanggil tool: calculate
   Argumen: {'expression': '(15 + 27) * 3'}
   Hasil: {"expression": "(15 + 27) * 3", "result": 126}
🔧 Memanggil tool: search_codebase
   Argumen: {'query': 'authentication'}
   Hasil: {"results": [{"file": "src/utils.py", "line": 42, "match": "Found query: authentication"}, {"file": "src/models.py", "line": 15, "match": "Related: authentication"}]}
🔧 Memanggil tool: get_current_time
   Argumen: {}
   Hasil: {"current_time": "2026-07-09T17:30:00", "timezone": "UTC"}

--- Iterasi 2 ---

✅ Jawaban final:
Berikut hasil dari permintaan Anda:

1. **Perhitungan**: (15 + 27) * 3 = 126
2. **Pencarian kode 'authentication'**: Ditemukan di 2 file:
   - `src/utils.py` baris 42
   - `src/models.py` baris 15
3. **Waktu saat ini**: 9 Juli 2026, 17:30 UTC

--- Token Usage ---
Input: 485
Output: 142
Total: 627

import os
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("MUSE_SPARK_API_KEY"),
    base_url="https://api.meta.ai/v1"
)

def analyze_large_file(file_path: str, question: str):
    """
    Membaca file besar dan menganalisis isinya dengan Muse Spark 1.1.
    Memanfaatkan jendela konteks 1 juta token.
    """
    with open(file_path, 'r', encoding='utf-8') as f:
        file_content = f.read()
    
    token_estimate = len(file_content) // 4  # Estimasi kasar: 1 token ≈ 4 karakter
    print(f"File size: {len(file_content)} characters (~{token_estimate} tokens)")
    
    response = client.chat.completions.create(
        model="muse-spark-1.1",
        messages=[
            {
                "role": "system",
                "content": (
                    "Kamu adalah code analyst. Kamu akan diberikan isi file "
                    "kode lengkap dan harus menjawab pertanyaan tentang file "
                    "tersebut dengan akurat. Gunakan informasi dari kode "
                    "yang diberikan, jangan mengarang."
                )
            },
            {
                "role": "user",
                "content": f"Berikut isi file:\n\n```\n{file_content}\n```\n\nPertanyaan: {question}"
            }
        ],
        temperature=0.2,
        max_tokens=2000
    )
    
    return response.choices[0].message.content

# Contoh: analisis file Python
if __name__ == "__main__":
    # Buat file dummy besar untuk testing
    sample_code = """
class UserManager:
    def __init__(self, db_connection):
        self.db = db_connection
        self.cache = {}
    
    def get_user(self, user_id: int) -> dict:
        if user_id in self.cache:
            return self.cache[user_id]
        
        user = self.db.query("SELECT * FROM users WHERE id = ?", user_id)
        if user:
            self.cache[user_id] = user
        return user
    
    def create_user(self, name: str, email: str) -> dict:
        existing = self.db.query("SELECT * FROM users WHERE email = ?", email)
        if existing:
            raise ValueError("Email already exists")
        
        user = self.db.query(
            "INSERT INTO users (name, email) VALUES (?, ?) RETURNING *",
            name, email
        )
        self.cache[user['id']] = user
        return user
    
    def update_user(self, user_id: int, **fields) -> dict:
        allowed_fields = {'name', 'email', 'status'}
        update_fields = {k: v for k, v in fields.items() if k in allowed_fields}
        
        if not update_fields:
            raise ValueError("No valid fields to update")
        
        set_clause = ", ".join(f"{k} = ?" for k in update_fields)
        query = f"UPDATE users SET {set_clause} WHERE id = ? RETURNING *"
        params = list(update_fields.values()) + [user_id]
        
        user = self.db.query(query, *params)
        if user:
            self.cache[user_id] = user
        return user
    
    def delete_user(self, user_id: int) -> bool:
        result = self.db.execute("DELETE FROM users WHERE id = ?", user_id)
        if result and user_id in self.cache:
            del self.cache[user_id]
        return result > 0
""".strip()
    
    # Tulis ke file sementara
    with open("sample_user_manager.py", "w") as f:
        f.write(sample_code)
    
    result = analyze_large_file(
        "sample_user_manager.py",
        "Fungsi apa saja yang ada di class UserManager? Apakah ada potensi bug atau masalah keamanan?"
    )
    print(result)

File size: 1234 characters (~308 tokens)

Class `UserManager` memiliki 4 fungsi utama:

1. **`get_user(user_id)`** — Mengambil user berdasarkan ID, dengan caching sederhana
2. **`create_user(name, email)`** — Membuat user baru, mengecek duplikasi email
3. **`update_user(user_id, **fields)`** — Update user dengan whitelist field yang diizinkan
4. **`delete_user(user_id)`** — Hapus user dan hapus dari cache

**Potensi masalah:**

- **SQL Injection**: Query menggunakan parameterized queries (`?`), yang sudah baik. Tapi di `update_user`, nama field di-inject langsung ke query string (`f"{k} = ?"`). Meskipun sudah difilter dengan `allowed_fields`, ini tetap pola yang berisiko jika `allowed_fields` di-expand tanpa validasi ketat di masa depan.

- **Cache invalidation**: Cache tidak ada TTL atau mekanisme expiry. Jika data user berubah di database oleh proses lain, cache akan stale.

- **Race condition**: `create_user` mengecek email existing lalu insert — ada race condition antara check dan insert. Sebaiknya gunakan unique constraint di database.

import os
import json
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("MUSE_SPARK_API_KEY"),
    base_url="https://api.meta.ai/v1"
)

def call_muse_spark(messages, tools=None, temperature=0.3):
    """Helper function untuk memanggil Muse Spark 1.1"""
    kwargs = {
        "model": "muse-spark-1.1",
        "messages": messages,
        "temperature": temperature,
        "max_tokens": 2000
    }
    if tools:
        kwargs["tools"] = tools
        kwargs["tool_choice"] = "auto"
    
    return client.chat.completions.create(**kwargs)

def subagent_research(topic: str) -> str:
    """Subagent yang fokus pada riset"""
    messages = [
        {
            "role": "system",
            "content": (
                "Kamu adalah research subagent. Tugasmu adalah mencari "
                "dan merangkum informasi teknis tentang topik yang diberikan. "
                "Berikan jawaban yang terstruktur dan faktual."
            )
        },
        {"role": "user", "content": f"Riset tentang: {topic}"}
    ]
    
    response = call_muse_spark(messages, temperature=0.2)
    return response.choices[0].message.content

def subagent_code(task: str, context: str = "") -> str:
    """Subagent yang fokus pada penulisan kode"""
    messages = [
        {
            "role": "system",
            "content": (
                "Kamu adalah coding subagent. Tugasmu adalah menulis kode "
                "yang bersih, efisien, dan well-documented. Selalu sertakan "
                "komentar dan type hints."
            )
        },
        {
            "role": "user",
            "content": f"Konteks: {context}\n\nTugas: {task}"
        }
    ]
    
    response = call_muse_spark(messages, temperature=0.3)
    return response.choices[0].message.content

def subagent_review(code: str) -> str:
    """Subagent yang fokus pada code review"""
    messages = [
        {
            "role": "system",
            "content": (
                "Kamu adalah code review subagent. Tugasmu adalah mengreview "
                "kode yang diberikan, mencari bug, masalah keamanan, dan "
                "area improvement. Berikan feedback yang spesifik dan actionable."
            )
        },
        {"role": "user", "content": f"Review kode berikut:\n\n```\n{code}\n```"}
    ]
    
    response = call_muse_spark(messages, temperature=0.2)
    return response.choices[0].message.content

def main_agent(task: str):
    """
    Main agent yang mengkoordinasi subagent untuk menyelesaikan tugas.
    """
    print(f"🎯 Main Agent: Menerima tugas — {task}")
    
    # Fase 1: Riset
    print("\n📋 Fase 1: Delegasi ke Research Subagent...")
    research_result = subagent_research(
        "Best practices untuk REST API authentication dengan JWT di Python FastAPI"
    )
    print(f"✅ Research selesai ({len(research_result)} chars)")
    
    # Fase 2: Coding
    print("\n💻 Fase 2: Delegasi ke Coding Subagent...")
    code_result = subagent_code(
        task="Buat endpoint login dengan JWT authentication menggunakan FastAPI",
        context=research_result
    )
    print(f"✅ Code selesai ({len(code_result)} chars)")
    
    # Fase 3: Review
    print("\n🔍 Fase 3: Delegasi ke Review Subagent...")
    review_result = subagent_review(code_result)
    print(f"✅ Review selesai ({len(review_result)} chars)")
    
    # Fase 4: Main agent synthesis
    print("\n🧠 Fase 4: Main Agent melakukan synthesis...")
    synthesis_messages = [
        {
            "role": "system",
            "content": (
                "Kamu adalah main agent. Kamu telah menerima hasil dari "
                "tiga subagent: research, coding, dan review. Tugasmu "
                "adakah menggabungkan semua hasil menjadi output final "
                "yang siap pakai, dengan mempertimbangkan feedback dari review."
            )
        },
        {
            "role": "user",
            "content": (
                f"Task: {task}\n\n"
                f"Research Result:\n{research_result}\n\n"
                f"Code Result:\n{code_result}\n\n"
                f"Review Result:\n{review_result}\n\n"
                f"Buat output final yang menggabungkan kode yang sudah diperbaiki "
                f"berdasarkan review."
            )
        }
    ]
    
    final_response = call_muse_spark(synthesis_messages, temperature=0.3)
    return final_response.choices[0].message.content

if __name__ == "__main__":
    result = main_agent("Buat endpoint autentikasi JWT untuk REST API dengan FastAPI")
    print("\n" + "="*60)
    print("HASIL FINAL:")
    print("="*60)
    print(result)

import os
import base64
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("MUSE_SPARK_API_KEY"),
    base_url="https://api.meta.ai/v1"
)

def analyze_screenshot(image_path: str, code_context: str = ""):
    """
    Mengirim screenshot ke Muse Spark 1.1 untuk dianalisis.
    Bisa dikombinasikan dengan kode untuk debugging yang lebih akurat.
    """
    with open(image_path, 'rb') as f:
        image_data = base64.b64encode(f.read()).decode('utf-8')
    
    messages = [
        {
            "role": "system",
            "content": (
                "Kamu adalah debugging assistant yang ahli dalam menganalisis "
                "screenshot error dan kode. Berikan diagnosis yang spesifik "
                "dan solusi yang actionable."
            )
        },
        {
            "role": "user",
            "content": [
                {
                    "type": "text",
                    "text": (
                        f"Saya menemukan error ini di aplikasi saya. "
                        f"Berikut screenshot error dan kode yang relevan:\n\n"
                        f"```\n{code_context}\n```"
                    )
                },
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/png;base64,{image_data}"
                    }
                }
            ]
        }
    ]
    
    response = client.chat.completions.create(
        model="muse-spark-1.1",
        messages=messages,
        temperature=0.2,
        max_tokens=2000
    )
    
    return response.choices[0].message.content

# Contoh penggunaan
if __name__ == "__main__":
    # Simulasi: kode yang menyebabkan error
    buggy_code = """
@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    user = db.users.find(user_id)
    return jsonify(user)
"""
    
    # Catatan: ganti 'error_screenshot.png' dengan file screenshot aktual
    # result = analyze_screenshot("error_screenshot.png", buggy_code)
    # print(result)
    print("Untuk menggunakan fitur ini, sediakan screenshot error dan jalankan:")
    print("result = analyze_screenshot('screenshot.png', kode_buggy)")

def estimate_cost(input_tokens: int, output_tokens: int):
    """
    Menghitung estimasi biaya pemakaian Muse Spark 1.1.
    """
    input_cost = (input_tokens / 1_000_000) * 1.25
    output_cost = (output_tokens / 1_000_000) * 4.25
    total = input_cost + output_cost
    
    print(f"Input tokens:  {input_tokens:>10,} → ${input_cost:.4f}")
    print(f"Output tokens: {output_tokens:>10,} → ${output_cost:.4f}")
    print(f"Total cost:                    ${total:.4f}")
    
    # Berapa kali bisa dipakai dengan $20 kredit gratis
    if total > 0:
        free_credits_uses = 20 / total
        print(f"\nDengan $20 kredit gratis: ~{free_credits_uses:.0f} request")
    
    return total

# Contoh: sesi coding agent dengan context panjang
estimate_cost(
    input_tokens=50_000,   # 50K token input (kode + konteks)
    output_tokens=5_000    # 5K token output (jawaban + kode)
)

Input tokens:     50,000 → $0.0625
Output tokens:     5,000 → $0.0213
Total cost:                    $0.0838

Dengan $20 kredit gratis: ~239 request

import os
import json
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("MUSE_SPARK_API_KEY"),
    base_url="https://api.meta.ai/v1"
)

# Definisi MCP-style tools (disederhanakan untuk demo)
MCP_TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "read_file",
            "description": "Membaca isi file dari filesystem",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path file yang akan dibaca"
                    }
                },
                "required": ["path"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "write_file",
            "description": "Menulis konten ke file",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path file tujuan"
                    },
                    "content": {
                        "type": "string",
                        "description": "Konten yang akan ditulis"
                    }
                },
                "required": ["path", "content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "list_directory",
            "description": "Melihat isi direktori",
            "parameters": {
                "type": "object",
                "properties": {
                    "path": {
                        "type": "string",
                        "description": "Path direktori",
                        "default": "."
                    }
                },
                "required": []
            }
        }
    }
]

def execute_mcp_tool(name: str, args: dict) -> str:
    """Eksekusi MCP tool (dengan safety checks)"""
    try:
        if name == "read_file":
            with open(args["path"], "r") as f:
                return json.dumps({"content": f.read()[:5000]})  # Limit 5K chars
        elif name == "write_file":
            with open(args["path"], "w") as f:
                f.write(args["content"])
            return json.dumps({"status": "success", "path": args["path"]})
        elif name == "list_directory":
            import os
            entries = os.listdir(args.get("path", "."))
            return json.dumps({"entries": entries})
    except Exception as e:
        return json.dumps({"error": str(e)})

def mcp_agent_loop(task: str, max_iterations: int = 10):
    """Agent loop dengan MCP tools"""
    messages = [
        {
            "role": "system",
            "content": (
                "Kamu adalah coding agent yang bisa membaca dan menulis file. "
                "Gunakan tools yang tersedia untuk menyelesaikan tugas. "
                "Selalu verifikasi hasil sebelum memberikan jawaban final."
            )
        },
        {"role": "user", "content": task}
    ]
    
    for i in range(max_iterations):
        response = client.chat.completions.create(
            model="muse-spark-1.1",
            messages=messages,
            tools=MCP_TOOLS,
            tool_choice="auto",
            temperature=0.2
        )
        
        msg = response.choices[0].message
        
        if msg.tool_calls:
            messages.append(msg)
            for tc in msg.tool_calls:
                fn_name = tc.function.name
                fn_args = json.loads(tc.function.arguments)
                print(f"  → {fn_name}({fn_args})")
                
                result = execute_mcp_tool(fn_name, fn_args)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result
                })
        else:
            print(f"\n✅ {msg.content}")
            return msg.content
    
    return "Batas iterasi tercapai"

if __name__ == "__main__":
    mcp_agent_loop(
        "Baca file main.py di direktori ini, lalu buat file baru bernama "
        "test_main.py dengan unit test dasar berdasarkan kode yang kamu baca."
    )

import os
import json
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(
    api_key=os.getenv("MUSE_SPARK_API_KEY"),
    base_url="https://api.meta.ai/v1"
)

def migration_agent(codebase_path: str):
    """
    Agent untuk migrasi codebase Flask ke FastAPI.
    """
    # Fase 1: Kumpulkan struktur codebase
    print("📂 Fase 1: Menganalisis struktur codebase...")
    
    file_list = []
    for root, dirs, files in os.walk(codebase_path):
        for f in files:
            if f.endswith('.py'):
                file_list.append(os.path.join(root, f))
    
    print(f"   Ditemukan {len(file_list)} file Python")
    
    # Fase 2: Baca dan analisis setiap file
    print("\n🔍 Fase 2: Menganalisis kode Flask...")
    
    codebase_content = ""
    for file_path in file_list[:10]:  # Limit 10 file untuk demo
        with open(file_path, 'r') as f:
            content = f.read()
            codebase_content += f"\n# === {file_path} ===\n{content}\n"
    
    analysis_response = client.chat.completions.create(
        model="muse-spark-1.1",
        messages=[
            {
                "role": "system",
                "content": (
                    "Kamu adalah migration specialist. Analisis codebase "
                    "Flask yang diberikan dan buat rencana migrasi ke FastAPI. "
                    "Identifikasi: 1) endpoint yang perlu dikonversi, "
                    "2) dependency yang perlu diganti, 3) urutan migrasi "
                    "yang aman, 4) potensi masalah."
                )
            },
            {
                "role": "user",
                "content": f"Codebase:\n```\n{codebase_content[:80000]}\n```"
            }
        ],
        temperature=0.2,
        max_tokens=3000
    )
    
    migration_plan = analysis_response.choices[0].message.content
    print(f"   Rencana migrasi dibuat ({len(migration_plan)} chars)")
    
    # Fase 3: Generate kode FastAPI per endpoint
    print("\n🔧 Fase 3: Generating kode FastAPI...")
    
    fastapi_response = client.chat.completions.create(
        model="muse-spark-1.1",
        messages=[
            {
                "role": "system",
                "content": (
                    "Kamu adalah FastAPI expert. Berdasarkan rencana migrasi, "
                    "generate kode FastAPI yang ekuivalen dengan kode Flask. "
                    "Sertakan: type hints, Pydantic models, async support, "
                    "dan error handling yang proper."
                )
            },
            {
                "role": "user",
                "content": (
                    f"Rencana migrasi:\n{migration_plan}\n\n"
                    f"Generate kode FastAPI untuk endpoint pertama."
                )
            }
        ],
        temperature=0.3,
        max_tokens=3000
    )
    
    fastapi_code = fastapi_response.choices[0].message.content
    print(f"   Kode FastAPI di-generate ({len(fastapi_code)} chars)")
    
    # Fase 4: Review dan validasi
    print("\n✅ Fase 4: Review dan validasi...")
    
    review_response = client.chat.completions.create(
        model="muse-spark-1.1",
        messages=[
            {
                "role": "system",
                "content": (
                    "Kamu adalah code reviewer. Bandingkan kode Flask asli "
                    "dengan kode FastAPI yang di-generate. Pastikan: "
                    "1) Semua endpoint ter-cover, 2) Logic bisnis sama, "
                    "3) Error handling ekuivalen, 4) Tidak ada kebocoran "
                    "security. Berikan skor 1-10 dan list perbaikan."
                )
            },
            {
                "role": "user",
                "content": (
                    f"Kode Flask:\n{codebase_content[:30000]}\n\n"
                    f"Kode FastAPI:\n{fastapi_code}"
                )
            }
        ],
        temperature=0.2,
        max_tokens=2000
    )
    
    review = review_response.choices[0].message.content
    print(f"   Review selesai ({len(review)} chars)")
    
    # Estimasi biaya
    total_input = (
        analysis_response.usage.prompt_tokens +
        fastapi_response.usage.prompt_tokens +
        review_response.usage.prompt_tokens
    )
    total_output = (
        analysis_response.usage.completion_tokens +
        fastapi_response.usage.completion_tokens +
        review_response.usage.completion_tokens
    )
    
    cost = (total_input / 1_000_000 * 1.25) + (total_output / 1_000_000 * 4.25)
    
    print(f"\n💰 Estimasi biaya: ${cost:.4f}")
    print(f"   Input: {total_input:,} tokens")
    print(f"   Output: {total_output:,} tokens")
    
    return {
        "plan": migration_plan,
        "code": fastapi_code,
        "review": review,
        "cost": cost
    }

# Jalankan migrasi
if __name__ == "__main__":
    result = migration_agent("./flask_app")

response = client.chat.completions.create(
    model="muse-spark-1.1",
    messages=messages,
    response_format={"type": "json_object"},
    temperature=0.2
)

import time
from openai import RateLimitError, APIError

def call_with_retry(client, max_retries=3, **kwargs):
    """Memanggil API dengan retry dan exponential backoff."""
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(**kwargs)
        except RateLimitError:
            wait_time = 2 ** attempt
            print(f"Rate limited. Menunggu {wait_time}s...")
            time.sleep(wait_time)
        except APIError as e:
            if attempt == max_retries - 1:
                raise
            print(f"API Error: {e}. Retry dalam {2 ** attempt}s...")
            time.sleep(2 ** attempt)
    
    raise Exception(f"Failed after {max_retries} retries")

def stream_response(client, messages, **kwargs):
    """Streaming response untuk real-time output."""
    stream = client.chat.completions.create(
        model="muse-spark-1.1",
        messages=messages,
        stream=True,
        **kwargs
    )
    
    full_response = ""
    for chunk in stream:
        if chunk.choices[0].delta.content is not None:
            content = chunk.choices[0].delta.content
            full_response += content
            print(content, end="", flush=True)
    
    print()  # Newline
    return full_response

def manage_context(messages: list, max_tokens: int = 800_000):
    """
    Mengelola context window untuk sesi agent yang panjang.
    Meringkas pesan lama dan menyimpan yang penting.
    """
    estimated_tokens = sum(len(m["content"]) // 4 for m in messages if isinstance(m.get("content"), str))
    
    if estimated_tokens < max_tokens:
        return messages
    
    # Simpan system prompt dan pesan terakhir
    system_msgs = [m for m in messages if m["role"] == "system"]
    recent_msgs = messages[-6:]  # 3 pasang terakhir
    
    # Ringkas pesan tengah
    middle_msgs = messages[len(system_msgs):-6]
    summary_content = "Ringkasan percakapan sebelumnya:\n"
    for m in middle_msgs:
        if m["role"] == "user":
            summary_content += f"- User: {m['content'][:200]}\n"
        elif m["role"] == "assistant":
            summary_content += f"- Assistant: {m['content'][:200]}\n"
    
    return system_msgs + [
        {"role": "system", "content": summary_content}
    ] + recent_msgs

# Contoh integrasi dengan LangChain
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import StrOutputParser

# Gunakan Muse Spark 1.1 lewat LangChain
llm = ChatOpenAI(
    model="muse-spark-1.1",
    api_key=os.getenv("MUSE_SPARK_API_KEY"),
    base_url="https://api.meta.ai/v1",
    temperature=0.3
)

prompt = ChatPromptTemplate.from_messages([
    ("system", "Kamu adalah coding assistant yang expert."),
    ("user", "{input}")
])

chain = prompt | llm | StrOutputParser()

result = chain.invoke({"input": "Buat function Python untuk merge sort dengan type hints."})
print(result)

# Contoh integrasi dengan tool calling di LangChain
from langchain.tools import tool

@tool
def search_docs(query: str) -> str:
    """Cari dokumentasi berdasarkan query."""
    # Implementasi pencarian
    return f"Hasil pencarian untuk: {query}"

@tool
def run_tests(test_file: str) -> str:
    """Jalankan test dan return hasil."""
    # Implementasi
    return f"Tests di {test_file}: 5 passed, 0 failed"

tools = [search_docs, run_tests]
llm_with_tools = llm.bind_tools(tools)

response = llm_with_tools.invoke(
    "Cari dokumentasi tentang FastAPI middleware, lalu jalankan test di test_app.py"
)
print(response.content)

def benchmark_model(client, test_cases: list, model_name: str):
    """
    Benchmark model dengan test cases internal.
    """
    results = []
    total_cost = 0
    
    for i, test in enumerate(test_cases):
        response = client.chat.completions.create(
            model=model_name,
            messages=test["messages"],
            temperature=test.get("temperature", 0.3),
            max_tokens=test.get("max_tokens", 2000)
        )
        
        cost = (
            response.usage.prompt_tokens / 1_000_000 * 1.25 +
            response.usage.completion_tokens / 1_000_000 * 4.25
        )
        total_cost += cost
        
        results.append({
            "test_id": i,
            "response": response.choices[0].message.content,
            "input_tokens": response.usage.prompt_tokens,
            "output_tokens": response.usage.completion_tokens,
            "cost": cost
        })
    
    print(f"Total cost for {len(test_cases)} tests: ${total_cost:.4f}")
    print(f"Average cost per test: ${total_cost/len(test_cases):.4f}")
    
    return results