Claude 06, Mengenal Model Context Protocol (MCP): Cara Menghubungkan Claude ke Tools dan Data Eksternal

# Buat virtual environment
python -m venv mcp-env
source mcp-env/bin/activate  # Windows: mcp-env\Scripts\activate

# Install paket MCP
pip install mcp

# Install dependensi tambahan
pip install anthropic httpx pydantic

# Verifikasi instalasi
python -c 'from mcp.server.fastmcp import FastMCP; print("MCP OK")'

my-mcp-server/
├── server.py      # MCP server utama
├── client.py      # MCP client untuk testing
├── tools/         # Modul helper (opsional)
│   └── database.py
└── requirements.txt

# server.py
from mcp.server.fastmcp import FastMCP
import json
from datetime import datetime

# Inisialisasi server
mcp = FastMCP("Document Manager")

# Storage sederhana untuk demo
documents = {}


# ── TOOLS: Operasi dengan side effects ──────────────────────────

@mcp.tool()
def create_document(title: str, content: str) -> str:
    """
    Buat dokumen baru dengan judul dan konten.
    
    Gunakan tool ini ketika user meminta membuat, menulis,
    atau menyimpan dokumen atau catatan baru.
    Mengembalikan ID dokumen yang dibuat.
    """
    doc_id = f'doc_{len(documents) + 1:03d}'
    documents[doc_id] = {
        'title': title,
        'content': content,
        'created_at': datetime.now().isoformat(),
        'updated_at': datetime.now().isoformat()
    }
    return json.dumps({'success': True, 'doc_id': doc_id, 'title': title})


@mcp.tool()
def edit_document(doc_id: str, new_content: str) -> str:
    """
    Edit konten dokumen yang sudah ada berdasarkan ID-nya.
    Gunakan untuk memperbarui atau merevisi dokumen yang sudah dibuat.
    JANGAN gunakan untuk membuat dokumen baru — gunakan create_document.
    """
    if doc_id not in documents:
        return json.dumps({'success': False, 'error': f'Dokumen {doc_id} tidak ditemukan'})
    
    documents[doc_id]['content'] = new_content
    documents[doc_id]['updated_at'] = datetime.now().isoformat()
    return json.dumps({'success': True, 'doc_id': doc_id})


@mcp.tool()
def search_documents(query: str) -> str:
    """
    Cari dokumen berdasarkan kata kunci dalam judul atau konten.
    Mengembalikan semua dokumen yang mengandung kata kunci tersebut.
    """
    results = []
    for doc_id, doc in documents.items():
        if (query.lower() in doc['title'].lower() or
                query.lower() in doc['content'].lower()):
            results.append({
                'doc_id': doc_id,
                'title': doc['title'],
                'preview': doc['content'][:100] + '...' if len(doc['content']) > 100 else doc['content']
            })
    
    return json.dumps({'count': len(results), 'results': results})


# ── RESOURCES: Data read-only ────────────────────────────────────

@mcp.resource(
    'documents://list',
    description='Daftar semua dokumen yang tersimpan di server'
)
def list_documents() -> str:
    """List semua dokumen."""
    doc_list = [
        {
            'doc_id': doc_id,
            'title': doc['title'],
            'created_at': doc['created_at'],
            'word_count': len(doc['content'].split())
        }
        for doc_id, doc in documents.items()
    ]
    return json.dumps({'total': len(doc_list), 'documents': doc_list})


@mcp.resource(
    'documents://{doc_id}',
    description='Baca konten lengkap dokumen berdasarkan ID-nya'
)
def get_document_by_id(doc_id: str) -> str:
    """Ambil dokumen spesifik berdasarkan ID."""
    if doc_id not in documents:
        return json.dumps({'error': f'Dokumen {doc_id} tidak ditemukan'})
    
    doc = documents[doc_id]
    return json.dumps({
        'doc_id': doc_id,
        'title': doc['title'],
        'content': doc['content'],
        'created_at': doc['created_at'],
        'updated_at': doc['updated_at']
    }, ensure_ascii=False)


# ── PROMPTS: Template reusable ───────────────────────────────────

@mcp.prompt()
def format_document(style: str = 'professional', language: str = 'Indonesia') -> str:
    """
    Template untuk memformat dokumen dengan gaya dan bahasa tertentu.
    Tersedia sebagai slash command /format_document di Claude Desktop.
    """
    style_instructions = {
        'professional': 'Gunakan bahasa formal dan terstruktur dengan heading yang jelas.',
        'casual': 'Gunakan bahasa santai dan ramah dengan bullet points.',
        'technical': 'Sertakan spesifikasi teknis dan terminologi yang presisi.',
        'executive': 'Ringkas dan to-the-point. Executive summary di awal.'
    }
    
    instruction = style_instructions.get(style, style_instructions['professional'])
    
    return f"""Format ulang dokumen berikut sesuai panduan ini:
    
Gaya: {style} — {instruction}
Bahasa output: {language}

Dokumen yang akan diformat:
[Paste dokumen di sini atau sebutkan doc_id-nya]"""


if __name__ == '__main__':
    mcp.run()  # Default: transport stdio

from typing import Optional, Annotated
from pydantic import Field

@mcp.tool()
def create_task(
    title: Annotated[str, Field(description='Judul task yang ringkas dan jelas')],
    priority: Annotated[str, Field(description='Prioritas: LOW, MEDIUM, HIGH, CRITICAL')],
    assignee: Annotated[Optional[str], Field(description='Username penerima. None jika belum ditentukan')] = None
) -> str:
    """
    Buat task baru dalam sistem manajemen proyek.
    
    Gunakan tool ini ketika user ingin membuat task atau to-do baru.
    JANGAN gunakan untuk memperbarui task yang sudah ada — gunakan update_task.
    """
    # implementasi...

# Cara termudah — tidak perlu install apapun
npx @modelcontextprotocol/inspector python server.py

# Atau via uvx
uvx mcp dev server.py

# client.py
import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client


async def main():
    # Konfigurasi server yang akan dihubungi
    server_params = StdioServerParameters(
        command='python',
        args=['server.py'],
        env=None
    )
    
    # Koneksi ke server via stdio
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # WAJIB: initialize handshake dulu
            await session.initialize()
            print('Terhubung ke MCP server!')
            
            # Discover tools yang tersedia
            tools_response = await session.list_tools()
            print(f'\nTools tersedia ({len(tools_response.tools)}):')
            for tool in tools_response.tools:
                print(f'  - {tool.name}: {tool.description[:60]}...')
            
            # Panggil tool
            result = await session.call_tool(
                'create_document',
                arguments={
                    'title': 'Laporan Q4 2024',
                    'content': 'Ini adalah laporan kuartal keempat...'
                }
            )
            print(f'\nHasil create_document:')
            for content_block in result.content:
                print(f'  {content_block.text}')
            
            # Baca resource
            resource_result = await session.read_resource('documents://list')
            print(f'\nDaftar dokumen:')
            for content in resource_result.contents:
                data = json.loads(content.text)
                for doc in data.get('documents', []):
                    print(f'  [{doc["doc_id"]}] {doc["title"]}')


if __name__ == '__main__':
    asyncio.run(main())

from mcp.client.streamable_http import streamablehttp_client

async with streamablehttp_client('http://localhost:8000/mcp') as (read, write, _):
    async with ClientSession(read, write) as session:
        await session.initialize()
        # API identik dengan stdio setelah ini

import asyncio
import anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

anthropic_client = anthropic.Anthropic()


def mcp_tools_to_anthropic(mcp_tools) -> list:
    """Konversi MCP tool objects ke format Anthropic API."""
    return [
        {
            'name': tool.name,
            'description': tool.description or '',
            'input_schema': tool.inputSchema
        }
        for tool in mcp_tools
    ]


async def claude_with_mcp(task: str, server_script: str) -> str:
    """Jalankan task menggunakan Claude dengan akses ke MCP server."""
    server_params = StdioServerParameters(command='python', args=[server_script])
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # Dapatkan tools dari MCP server dan konversi ke format Anthropic
            tools_resp = await session.list_tools()
            anthropic_tools = mcp_tools_to_anthropic(tools_resp.tools)
            
            # Mulai agentic loop
            messages = [{'role': 'user', 'content': task}]
            max_iterations = 10  # Safeguard penting — selalu pasang ini
            
            for iteration in range(max_iterations):
                response = anthropic_client.messages.create(
                    model='claude-sonnet-4-6',
                    max_tokens=4096,
                    tools=anthropic_tools,
                    messages=messages
                )
                
                # Claude selesai — tidak ada tool call lagi
                if response.stop_reason == 'end_turn':
                    return next(
                        (b.text for b in response.content if hasattr(b, 'text')),
                        'Task selesai.'
                    )
                
                # Claude ingin memanggil tool
                if response.stop_reason == 'tool_use':
                    messages.append({'role': 'assistant', 'content': response.content})
                    tool_results = []
                    
                    for block in response.content:
                        if block.type == 'tool_use':
                            print(f'  → Calling: {block.name}({block.input})')
                            
                            # Eksekusi tool via MCP session
                            mcp_result = await session.call_tool(block.name, block.input)
                            result_text = '\n'.join(
                                c.text for c in mcp_result.content if hasattr(c, 'text')
                            )
                            
                            tool_results.append({
                                'type': 'tool_result',
                                'tool_use_id': block.id,
                                'content': result_text
                            })
                    
                    messages.append({'role': 'user', 'content': tool_results})
            
            return 'Max iterations reached.'


# Test
if __name__ == '__main__':
    result = asyncio.run(claude_with_mcp(
        task='Buat 3 dokumen: tentang Python, MCP, dan AI. '
             'Setelah selesai, cari dokumen yang berisi kata Python.',
        server_script='server.py'
    ))
    print(f'\nHasil: {result}')

{
  "mcpServers": {
    "document-manager": {
      "command": "python",
      "args": ["/path/absolut/ke/server.py"],
      "env": {
        "PYTHONPATH": "/path/ke/project"
      }
    }
  }
}

from mcp.types import McpError, ErrorCode

async def safe_call_tool(session: ClientSession, tool_name: str, arguments: dict) -> str:
    """Call tool dengan error handling yang lengkap."""
    try:
        result = await session.call_tool(tool_name, arguments)
        
        # Cek apakah result berisi error dari server
        if result.isError:
            error_text = '\n'.join(
                c.text for c in result.content if hasattr(c, 'text')
            )
            return json.dumps({'error': error_text})
        
        return '\n'.join(c.text for c in result.content if hasattr(c, 'text'))
    
    except McpError as e:
        if e.error.code == ErrorCode.MethodNotFound:
            return json.dumps({'error': f'Tool {tool_name} tidak ditemukan'})
        elif e.error.code == ErrorCode.InvalidParams:
            return json.dumps({'error': f'Parameter tidak valid: {e.error.message}'})
        else:
            return json.dumps({'error': f'MCP Error: {e.error.message}'})
    
    except Exception as e:
        return json.dumps({'error': f'Unexpected error: {str(e)}'})