การเปิดให้เข้าถึงทรัพยากรฐานข้อมูลผ่าน MCP

เหตุผลที่ฐานข้อมูลต้องใช้ MCP

การออกแบบสคีมา MCP สำหรับฐานข้อมูล

การตั้งค่าการเชื่อมต่อฐานข้อมูล

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

การแสดงระเบียนเป็นเครื่องมือ

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

การเปิดเผยแถวฐานข้อมูลเป็น 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}')

การแบ่งหน้าสำหรับชุดผลลัพธ์ขนาดใหญ่

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'

เครื่องมือค้นหาข้อความแบบเต็ม

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

การดำเนินการเขียนพร้อมการยืนยัน

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

การกรองข้อมูลที่ละเอียดอ่อน

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

การบันทึกเพื่อตรวจสอบการเข้าถึงฐานข้อมูล

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

การทดสอบเซิร์ฟเวอร์ MCP สำหรับฐานข้อมูล

ตรวจสอบความเข้าใจ

สรุปบทเรียน