MCP로 데이터베이스 리소스 제공

Why Databases Need MCP

Designing Your Database MCP Schema

Database Connection Setup

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

Listing Records as a Tool

@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))]

Exposing Database Rows as Resources

@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}')

Pagination for Large Result Sets

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'

Full-Text Search Tool

@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))]

Write Operations with Confirmation

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}')]

Filtering Sensitive Data

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 {}

Audit Logging for Database Access

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

Testing Your Database MCP Server

Quick Check

Lesson Recap