Menyediakan Sumber Daya Basis Data melalui MCP

Mengapa Basis Data Membutuhkan MCP

Merancang Skema MCP Basis Data

Menyiapkan Koneksi Basis Data

import asyncpg
import os
from mcp.server import Server
from mcp import types
import asyncio

app = Server('database-mcp-server')
pool = None  # Global connection pool

async def get_pool():
    global pool
    if pool is None:
        pool = await asyncpg.create_pool(
            os.environ['DATABASE_URL'],
            min_size=2,
            max_size=10,
            command_timeout=30
        )
    return pool

# Initialize pool at startup
async def startup():
    await get_pool()
    print('Database pool initialized', flush=False)  # stderr only via logger

Menyediakan Record sebagai Alat

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name='list_products',
            description='List products from the catalog with optional filtering by category and price range.',
            inputSchema={
                'type': 'object',
                'properties': {
                    'category': {'type': 'string', 'description': 'Filter by product category.'},
                    'max_price': {'type': 'number', 'description': 'Maximum price in USD.'},
                    'limit': {'type': 'integer', 'default': 20, 'maximum': 100}
                },
                'required': []
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict):
    if name == 'list_products':
        return await list_products_handler(arguments)

async def list_products_handler(args: dict):
    pool = await get_pool()
    query = 'SELECT id, name, category, price, stock FROM products WHERE 1=1'
    params = []

    if 'category' in args:
        params.append(args['category'])
        query += f' AND category = ${len(params)}'
    if 'max_price' in args:
        params.append(args['max_price'])
        query += f' AND price <= ${len(params)}'

    limit = min(args.get('limit', 20), 100)
    params.append(limit)
    query += f' ORDER BY name LIMIT ${len(params)}'

    async with pool.acquire() as conn:
        rows = await conn.fetch(query, *params)

    if not rows:
        return [types.TextContent(type='text', text='No products found matching your criteria.')]

    lines = ['id | name | category | price | stock']
    lines += [f'{r["id"]} | {r["name"]} | {r["category"]} | ${r["price"]} | {r["stock"]}' for r in rows]
    return [types.TextContent(type='text', text='\n'.join(lines))]

Mengekspos Baris Basis Data sebagai Resource

@app.list_resources()
async def list_resources():
    pool = await get_pool()
    async with pool.acquire() as conn:
        # List most recently updated products as resources
        rows = await conn.fetch(
            'SELECT id, name FROM products ORDER BY updated_at DESC LIMIT 50'
        )
    return [
        types.Resource(
            uri=f'db://products/{row["id"]}',
            name=row['name'],
            description=f'Product record for {row["name"]}',
            mimeType='application/json'
        )
        for row in rows
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    import json
    if uri.startswith('db://products/'):
        product_id = int(uri.split('/')[-1])
        pool = await get_pool()
        async with pool.acquire() as conn:
            row = await conn.fetchrow(
                'SELECT id, name, category, price, description, stock FROM products WHERE id = $1',
                product_id
            )
        if row is None:
            raise ValueError(f'Product {product_id} not found')
        return json.dumps(dict(row), default=str)
    raise ValueError(f'Unknown resource URI: {uri}')

Paginasi untuk Kumpulan Hasil Besar

async def paginated_list(table: str, limit: int = 20, after_id: int = 0) -> tuple:
    '''Returns (rows, has_more, next_cursor).'''
    pool = await get_pool()
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            f'SELECT * FROM {table} WHERE id > $1 ORDER BY id LIMIT $2',
            after_id,
            limit + 1  # Fetch one extra to detect if more pages exist
        )

    has_more = len(rows) > limit
    rows = rows[:limit]
    next_cursor = rows[-1]['id'] if rows else None
    return rows, has_more, next_cursor

# In your tool result, include pagination info:
# 'Showing 20 products. To see the next page, call list_products with after_id=120'

Alat Pencarian Teks Lengkap

@app.list_tools()
async def list_tools():
    return [
        types.Tool(
            name='search_products',
            description='Full-text search across product names and descriptions. Use for finding products by natural language, not by category filter.',
            inputSchema={
                'type': 'object',
                'properties': {
                    'query': {'type': 'string', 'description': 'Natural language search query.'},
                    'limit': {'type': 'integer', 'default': 10, 'maximum': 50}
                },
                'required': ['query']
            }
        )
    ]

async def search_products_handler(args: dict):
    pool = await get_pool()
    async with pool.acquire() as conn:
        rows = await conn.fetch(
            '''SELECT id, name, price, ts_rank(search_vector, query) AS rank
               FROM products, plainto_tsquery('english', $1) AS query
               WHERE search_vector @@ query
               ORDER BY rank DESC
               LIMIT $2''',
            args['query'],
            args.get('limit', 10)
        )
    if not rows:
        return [types.TextContent(type='text', text=f'No products found matching "{args["query"]}".  ')]
    lines = [f'{r["id"]}: {r["name"]} — ${r["price"]}' for r in rows]
    return [types.TextContent(type='text', text='\n'.join(lines))]

Operasi Tulis dengan Konfirmasi

import uuid
pending_actions = {}  # In-memory store; use Redis in production

async def create_order_preview(args: dict) -> list:
    action_id = str(uuid.uuid4())[:8]
    pending_actions[action_id] = {'type': 'create_order', 'data': args}
    return [types.TextContent(
        type='text',
        text=f'Preview: Create order for customer {args["customer_id"]} with {len(args["items"])} items, '
             f'total ${args["total"]}. Confirm with: confirm_action(action_id="{action_id}")')
    ]

async def confirm_action_handler(args: dict) -> list:
    action_id = args.get('action_id')
    action = pending_actions.pop(action_id, None)
    if not action:
        return [types.TextContent(type='text', text='Action expired or not found.')]
    # Execute the pending action
    result = await execute_action(action)
    return [types.TextContent(type='text', text=f'Done: {result}')]

Memfilter Data Sensitif

async def get_safe_customer(customer_id: int) -> dict:
    '''Fetch customer data with sensitive fields excluded.'''
    pool = await get_pool()
    async with pool.acquire() as conn:
        row = await conn.fetchrow(
            '''SELECT id, name, created_at, country, tier
               FROM customers
               WHERE id = $1
               -- NEVER select: password_hash, api_key, full_address, payment_method_id
            ''',
            customer_id
        )
    return dict(row) if row else {}

Logging Audit untuk Akses Basis Data

import logging
import sys
import time

audit_logger = logging.getLogger('mcp.audit')
audit_logger.setLevel(logging.INFO)
handler = logging.StreamHandler(sys.stderr)
audit_logger.addHandler(handler)

async def audited_tool_call(name: str, arguments: dict, session_id: str = None) -> list:
    start = time.time()
    result = await call_tool(name, arguments)
    elapsed_ms = round((time.time() - start) * 1000)

    audit_logger.info({
        'event': 'tool_call',
        'tool': name,
        'args': {k: v for k, v in arguments.items() if k != 'password'},
        'result_count': len(result),
        'elapsed_ms': elapsed_ms,
        'session_id': session_id
    })
    return result

Menguji Server MCP Basis Data

Pemeriksaan Cepat

Ringkasan Pelajaran