Esporre risorse del database tramite MCP

Perché i database hanno bisogno di MCP

Progettazione dello schema MCP del database

Configurazione della connessione al database

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

Elenco dei record come strumento

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

Esposizione delle righe del database come risorse

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

Impaginazione di grandi insiemi di risultati

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'

Strumento di ricerca full-text

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

Operazioni di scrittura con conferma

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

Filtraggio dei dati sensibili

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 di audit per l'accesso al database

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

Test del server MCP del database

Verifica rapida

Riepilogo della lezione