Claude 09, MCP Advanced Topics: Panduan Teknis Sampling, Notifications, Roots, dan Transport untuk Developer

# sampling_server.py
from mcp.server.fastmcp import FastMCP, Context
from mcp.server.session import ServerSession
from mcp.types import SamplingMessage, TextContent, ModelPreferences, ModelHint

mcp = FastMCP('Sampling Demo Server')


@mcp.tool()
async def analyze_and_summarize(
    content: str,
    ctx: Context[ServerSession, None]
) -> str:
    """
    Analisis konten dan hasilkan ringkasan menggunakan LLM client.
    Server mendelegasikan pembuatan ringkasan ke model yang dimiliki client.
    """
    result = await ctx.session.create_message(
        messages=[
            SamplingMessage(
                role="user",
                content=TextContent(
                    type="text",
                    text=f"Buat ringkasan 3 poin dari teks berikut:\n\n{content}"
                )
            )
        ],
        max_tokens=500,
        # Preferensi model — client yang memutuskan final
        model_preferences=ModelPreferences(
            hints=[ModelHint(name='claude-sonnet')],
            speed_priority=0.3,
            intelligence_priority=0.8
        ),
        system_prompt="Kamu adalah analis profesional. Berikan ringkasan yang ringkas dan akurat.",
        include_context="none"
    )

    if result.content.type == 'text':
        return f'Ringkasan (dari {result.model}):\n{result.content.text}'
    return str(result.content)

# sampling_client.py
import asyncio
import anthropic
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import CreateMessageRequestParams, CreateMessageResult, TextContent

anthropic_client = anthropic.Anthropic()


async def handle_sampling_request(
    params: CreateMessageRequestParams
) -> CreateMessageResult:
    """
    Callback yang dipanggil saat server mengirim sampling request.
    Di sinilah client mengeksekusi actual LLM call.
    """
    # Konversi MCP messages ke format Anthropic
    anthropic_messages = []
    for msg in params.messages:
        if msg.content.type == 'text':
            anthropic_messages.append({
                'role': msg.role,
                'content': msg.content.text
            })

    # Client yang memutuskan model final, bukan server
    model = 'claude-sonnet-4-6'
    if params.model_preferences and params.model_preferences.hints:
        hint = params.model_preferences.hints[0].name
        if hint and 'haiku' in hint.lower():
            model = 'claude-haiku-4-5'

    response = anthropic_client.messages.create(
        model=model,
        max_tokens=params.max_tokens or 1024,
        system=params.system_prompt if params.system_prompt else None,
        messages=anthropic_messages
    )

    return CreateMessageResult(
        role='assistant',
        content=TextContent(type='text', text=response.content[0].text),
        model=model,
        stop_reason='end_turn'
    )


async def run_with_sampling():
    server_params = StdioServerParameters(
        command='python', args=['sampling_server.py']
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(
            read, write,
            sampling_callback=handle_sampling_request  # ← Kunci untuk sampling
        ) as session:
            await session.initialize()

            result = await session.call_tool(
                'analyze_and_summarize',
                {'content': 'Laporan kuartal menunjukkan pertumbuhan 15% di Q3...'}
            )
            print(result.content[0].text)


asyncio.run(run_with_sampling())

# notification_server.py
from mcp.server.fastmcp import FastMCP, Context
import asyncio

mcp = FastMCP('Notification Demo')


@mcp.tool()
async def batch_process_files(
    file_paths: list[str],
    ctx: Context
) -> str:
    """Proses sekumpulan file dengan progress reporting real-time."""
    total = len(file_paths)
    processed = 0
    errors = 0

    await ctx.info(f'Memulai batch processing {total} file')

    for i, file_path in enumerate(file_paths):
        # Kirim progress update ke client
        await ctx.report_progress(i, total)
        await ctx.debug(f'Memproses: {file_path}')

        try:
            await asyncio.sleep(0.05)  # Simulasi pemrosesan
            processed += 1
            await ctx.info(f'File {file_path} berhasil diproses')
        except Exception as e:
            errors += 1
            await ctx.error(f'Gagal memproses {file_path}: {e}')

    # 100% saat selesai
    await ctx.report_progress(total, total)
    await ctx.info(f'Batch selesai: {processed} berhasil, {errors} gagal')

    return f'Batch processing selesai: {processed}/{total} file berhasil'


@mcp.tool()
async def analyze_large_dataset(
    dataset_name: str,
    num_records: int,
    ctx: Context
) -> dict:
    """Analisis dataset besar dengan progress per fase."""

    PHASES = [
        ('Memuat data', 0.1),
        ('Validasi schema', 0.15),
        ('Preprocessing', 0.25),
        ('Analisis statistik', 0.30),
        ('Deteksi anomali', 0.15),
        ('Generasi laporan', 0.05),
    ]

    results = {}
    current_progress = 0

    await ctx.info(f'Memulai analisis {dataset_name} ({num_records:,} record)')

    for phase_name, phase_weight in PHASES:
        await ctx.info(f'Phase: {phase_name}...')

        steps = 10
        for step in range(steps):
            await asyncio.sleep(0.01)
            step_progress = current_progress + (phase_weight * step / steps * num_records)
            await ctx.report_progress(int(step_progress), num_records)

        current_progress += phase_weight * num_records
        results[phase_name] = 'Selesai'
        await ctx.debug(f'Phase {phase_name} selesai')

    await ctx.report_progress(num_records, num_records)  # 100%
    await ctx.info('Analisis dataset selesai!')

    return {'status': 'success', 'phases': results, 'total_records': num_records}

# roots_server.py
from mcp.server.fastmcp import FastMCP, Context
from pathlib import Path
import json

mcp = FastMCP('File Access Server')


@mcp.tool()
async def list_accessible_files(
    extension_filter: str = '',
    ctx: Context
) -> str:
    """
    List semua file yang accessible berdasarkan roots yang dikonfigurasi client.
    Hanya membaca direktori yang sudah di-grant, tidak lebih.
    """
    roots_result = await ctx.session.list_roots()

    if not roots_result.roots:
        return 'Tidak ada roots yang dikonfigurasi. Client belum memberikan akses direktori.'

    all_files = []

    for root in roots_result.roots:
        root_uri = root.uri
        root_name = root.name or root_uri

        await ctx.info(f'Scanning root: {root_name}')

        if root_uri.startswith('file://'):
            local_path = root_uri.replace('file://', '')
            dir_path = Path(local_path)

            if dir_path.exists() and dir_path.is_dir():
                for file_path in dir_path.iterdir():
                    if file_path.is_file():
                        if not extension_filter or file_path.suffix == extension_filter:
                            all_files.append({
                                'path': str(file_path),
                                'name': file_path.name,
                                'size': file_path.stat().st_size,
                                'root': root_name
                            })
            else:
                await ctx.warning(f'Path tidak ditemukan: {local_path}')

    return json.dumps({'total': len(all_files), 'files': all_files})


@mcp.tool()
async def read_file_safe(filename: str, ctx: Context) -> str:
    """
    Baca file HANYA jika ada dalam salah satu root yang diotorisasi.
    Mencegah path traversal dan akses ke lokasi yang tidak diizinkan.
    """
    roots_result = await ctx.session.list_roots()
    target_path = Path(filename).resolve()  # Normalize path

    # Validasi: pastikan file ada dalam salah satu root
    is_authorized = False
    for root in roots_result.roots:
        if root.uri.startswith('file://'):
            root_path = Path(root.uri.replace('file://', '')).resolve()
            try:
                target_path.relative_to(root_path)  # Akan raise ValueError jika di luar root
                is_authorized = True
                break
            except ValueError:
                continue

    if not is_authorized:
        await ctx.error(f'AKSES DITOLAK: {filename} tidak dalam root yang diotorisasi')
        return json.dumps({'error': 'Akses ditolak — file di luar root yang diizinkan'})

    try:
        with open(target_path, 'r', encoding='utf-8') as f:
            content = f.read()
        await ctx.info(f'File dibaca: {filename}')
        return json.dumps({'filename': filename, 'content': content})
    except Exception as e:
        await ctx.error(f'Gagal membaca {filename}: {e}')
        return json.dumps({'error': str(e)})

# roots_client.py
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import Root


async def run_with_roots():
    server_params = StdioServerParameters(command='python', args=['roots_server.py'])

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(
            read, write,
            roots=[
                Root(uri='file:///home/user/documents', name='My Documents'),
                Root(uri='file:///home/user/projects', name='Projects'),
                # TIDAK memberikan akses ke /etc atau direktori sensitif
            ]
        ) as session:
            await session.initialize()

            # List file yang accessible
            result = await session.call_tool('list_accessible_files', {})
            print(result.content[0].text)

            # Coba baca file di luar root — akan ditolak
            result3 = await session.call_tool(
                'read_file_safe',
                {'filename': '/etc/passwd'}  # Di luar root!
            )
            print(result3.content[0].text)  # {'error': 'Akses ditolak...'}


asyncio.run(run_with_roots())

// Request: Client memanggil tool
{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "tools/call",
  "params": {
    "name": "read_file",
    "arguments": {"filename": "report.txt"}
  }
}

// Response sukses
{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "content": [{"type": "text", "text": "Isi file..."}],
    "isError": false
  }
}

// Notification progress (tidak ada id, tidak ada respons)
{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": {
    "progressToken": "proc-001",
    "progress": 45,
    "total": 100
  }
}

# stdio_server.py
from mcp.server.fastmcp import FastMCP
import sys
import logging

# KRITIS: logging ke stderr, BUKAN stdout
logging.basicConfig(
    stream=sys.stderr,  # ← PENTING
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(message)s'
)
logger = logging.getLogger(__name__)

mcp = FastMCP('STDIO Demo Server')


@mcp.tool()
def calculate(expression: str) -> str:
    """Evaluasi ekspresi matematika sederhana."""
    logger.debug(f'Evaluating: {expression}')  # Aman — ke stderr
    try:
        result = eval(expression, {'__builtins__': {}},
                      {'abs': abs, 'round': round, 'min': min, 'max': max})
        return str(result)
    except Exception as e:
        logger.error(f'Error: {e}')
        return f'Error: {e}'


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

{
  "mcpServers": {
    "calculator": {
      "command": "python",
      "args": ["/absolute/path/to/stdio_server.py"],
      "env": {
        "PYTHONPATH": "/path/to/project",
        "DATABASE_URL": "postgresql://localhost/mydb"
      }
    }
  }
}

# streamable_http_server.py
from mcp.server.fastmcp import FastMCP

# Konfigurasi untuk production: stateless + JSON response
# Ideal untuk horizontal scaling dengan load balancer
mcp = FastMCP(
    'Production Server',
    stateless_http=True,   # Tidak ada session state
    json_response=True     # Respons JSON langsung, tidak ada SSE streaming
)


@mcp.tool()
async def get_server_time() -> str:
    """Kembalikan waktu server saat ini."""
    from datetime import datetime
    return datetime.now().isoformat()


@mcp.tool()
async def search_database(query: str, limit: int = 10) -> list:
    """Cari database berdasarkan query."""
    return [{'id': i, 'result': f'Result {i} for {query}'} for i in range(limit)]


if __name__ == '__main__':
    mcp.run(
        transport='streamable-http',
        host='0.0.0.0',
        port=8000
    )
    # Accessible di: http://localhost:8000/mcp

# multi_server_app.py
import contextlib
from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP

# Server 1
data_mcp = FastMCP('Data API', stateless_http=True, json_response=True)

@data_mcp.tool()
def get_user(user_id: int) -> dict:
    return {'id': user_id, 'name': f'User {user_id}'}

# Server 2
analytics_mcp = FastMCP('Analytics API', stateless_http=True, json_response=True)

@analytics_mcp.tool()
def get_stats() -> dict:
    return {'requests': 1000, 'errors': 5}

# Mount keduanya ke satu FastAPI app
@contextlib.asynccontextmanager
async def lifespan(app: FastAPI):
    async with data_mcp.lifespan(), analytics_mcp.lifespan():
        yield

app = FastAPI(lifespan=lifespan)
app.mount('/data', data_mcp.streamable_http_app())
app.mount('/analytics', analytics_mcp.streamable_http_app())

# Accessible di:
# http://localhost:8000/data/mcp
# http://localhost:8000/analytics/mcp