إتاحة موارد قواعد البيانات عبر 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))]

إتاحة صفوف قاعدة البيانات كموارد

@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 لقاعدة البيانات

تحقق سريع

مراجعة الدرس