Cómo extender tu servidor MCP con acceso a base de datos

Conectar tu servidor MCP a una base de datos real lo transforma de una simple demostración a un backend de IA listo para producción. En esta guía, verás exactamente cómo integrar PostgreSQL en tu servidor MCP y exponerlo de manera segura a un LLM.

Por qué conectar una base de datos a MCP

Sin acceso a una base de datos, tu LLM está ciego a los datos reales de tu aplicación. Al conectarlo, puedes: Permitir que la IA responda preguntas basadas en usuarios reales, pedidos, tickets, etc; Automatizar acciones como crear entradas o actualizar registros; Construir agentes internos inteligentes sin necesidad de APIs separadas; Habilitar respuestas de IA conscientes del contexto utilizando el estado de tu aplicación; Crear análisis impulsados por IA sobre tus datos de negocio. Este es el primer paso para convertir un modelo en un verdadero asistente de aplicación que proporciona un valor comercial genuino.

Lo que necesitas antes de empezar

• Una base de datos PostgreSQL en funcionamiento (v12+)
• Un servidor MCP en Python (implementación básica)
• Python 3.10+ con soporte asíncrono
• La biblioteca asyncpg para acceso a la base de datos
• El paquete mcp-server (SDK oficial de Python)
• Python-dotenv para la configuración del entorno
• Conocimientos básicos de SQL y Python asíncrono

Visión general de la arquitectura

La arquitectura involucra varios componentes: (1) Cliente LLM: Claude u otro LLM que se comunica a través del protocolo MCP, (2) Servidor MCP: Tu servidor Python que expone recursos y herramientas, (3) Pool de conexiones: Gestiona las conexiones a la base de datos de manera eficiente, (4) PostgreSQL: La base de datos subyacente que almacena los datos de tu aplicación.

Esta configuración sigue una clara separación de responsabilidades: Los recursos proporcionan acceso de solo lectura para consultas, Las herramientas permiten operaciones de escritura controladas, El pooling de conexiones optimiza el rendimiento, La configuración del entorno mantiene las credenciales seguras.

Paso 1: Instalar y configurar dependencias de base de datos

Primero, instala los paquetes necesarios:

pip install asyncpg python-dotenv mcp-server

Crea una estructura de proyecto:

mcp-db-server/
├── .env                  # Variables de entorno (nunca lo subas a git)
├── requirements.txt      # Dependencias
├── server.py             # Archivo principal del servidor
├── database.py           # Módulo de conexión a la base de datos
├── resources/            # Recursos de base de datos
│   ├── __init__.py
│   └── users.py          # Recursos relacionados con usuarios
└── tools/                # Herramientas de base de datos
    ├── __init__.py
    └── users.py          # Herramientas relacionadas con usuarios

Paso 2: Configurar el entorno

Crea un archivo .env para tus credenciales de base de datos:

DB_USER=your_db_user
DB_PASSWORD=your_db_password
DB_NAME=your_db_name
DB_HOST=localhost
DB_PORT=5432

Nunca subas este archivo al control de versiones. Añádelo a .gitignore:

# .gitignore
.env
__pycache__/
*.py[cod]
*$py.class

Crea un archivo database.py para cargar estas variables de entorno:

import os
import asyncpg
from dotenv import load_dotenv

# Cargar variables de entorno
load_dotenv()

# Configuración de la base de datos
DB_CONFIG = {
    "user": os.getenv("DB_USER"),
    "password": os.getenv("DB_PASSWORD"),
    "database": os.getenv("DB_NAME"),
    "host": os.getenv("DB_HOST", "localhost"),
    "port": int(os.getenv("DB_PORT", "5432")),
}

Paso 3: Crear pool de conexiones a la base de datos

Extiende tu archivo database.py para incluir el pooling de conexiones:

import os
import asyncpg
import logging
from dotenv import load_dotenv

# Configurar logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mcp_database")

# Cargar variables de entorno
load_dotenv()

# Configuración de la base de datos
DB_CONFIG = {
    "user": os.getenv("DB_USER"),
    "password": os.getenv("DB_PASSWORD"),
    "database": os.getenv("DB_NAME"),
    "host": os.getenv("DB_HOST", "localhost"),
    "port": int(os.getenv("DB_PORT", "5432")),
}

# Variable global para el pool
db_pool = None

async def init_db():
    """Inicializar el pool de conexiones a la base de datos."""
    global db_pool
    try:
        db_pool = await asyncpg.create_pool(
            **DB_CONFIG,
            min_size=1,
            max_size=10,
            command_timeout=60,
            timeout=10,  # Tiempo de espera para adquisición de conexión
        )
        logger.info("Pool de conexiones a la base de datos establecido")
        # Probar la conexión
        async with db_pool.acquire() as connection:
            version = await connection.fetchval("SELECT version();")
            logger.info(f"Conectado a PostgreSQL: {version}")
        return db_pool
    except Exception as e:
        logger.error(f"Error al crear el pool de base de datos: {e}")
        raise

async def close_db():
    """Cerrar el pool de conexiones a la base de datos."""
    global db_pool
    if db_pool:
        await db_pool.close()
        logger.info("Pool de conexiones a la base de datos cerrado")

Esto te proporciona: Un pool de conexiones correctamente configurado, Uso de variables de entorno para seguridad, Logging para monitoreo, Optimización del tamaño del pool, Manejo de tiempos de espera de conexión, Función explícita de cierre para apagados limpios.

Paso 4: Exponer un recurso de solo lectura

Crea resources/users.py para exponer datos de usuarios:

from typing import List, Dict, Any, Optional
import logging
from database import db_pool

logger = logging.getLogger("mcp_database.resources.users")

async def fetch_recent_users(limit: int = 20) -> List[Dict[str, Any]]:
    """
    Obtener los usuarios más recientes de la base de datos.
    
    Args:
        limit: Número máximo de usuarios a devolver (por defecto: 20)
        
    Returns:
        Lista de objetos de usuario
    """
    try:
        if not db_pool:
            logger.error("Pool de base de datos no inicializado")
            return {"error": "Conexión a base de datos no disponible"}
            
        async with db_pool.acquire() as connection:
            # Usar una consulta parametrizada por seguridad
            query = """
                SELECT id, username, email, created_at 
                FROM users 
                ORDER BY created_at DESC 
                LIMIT $1;
            """
            rows = await connection.fetch(query, limit)
            
            # Convertir a diccionarios y manejar serialización de datetime
            users = []
            for row in rows:
                user = dict(row)
                # Convertir datetime a formato ISO para serialización JSON
                if "created_at" in user and user["created_at"]:
                    user["created_at"] = user["created_at"].isoformat()
                users.append(user)
                
            logger.info(f"Obtenidos {len(users)} usuarios recientes")
            return users
    except Exception as e:
        logger.error(f"Error al obtener usuarios recientes: {e}")
        return {"error": f"Error de base de datos: {str(e)}"}

Ahora, actualiza tu server.py para registrar este recurso:

import asyncio
import logging
from mcp_server import MCPServer
from database import init_db, close_db
from resources.users import fetch_recent_users

# Configurar logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp_server")

# Crear servidor MCP
server = MCPServer()

# Registrar el recurso
@server.resource(
    name="recent_users", 
    description="Obtener los usuarios más recientes de la base de datos."
)
async def recent_users_resource():
    return await fetch_recent_users(limit=20)

async def main():
    try:
        # Inicializar la base de datos
        await init_db()
        
        # Iniciar el servidor
        logger.info("Iniciando servidor MCP...")
        await server.start()
    except Exception as e:
        logger.error(f"Error del servidor: {e}")
    finally:
        # Cerrar conexiones de base de datos al apagar
        await close_db()

if __name__ == "__main__":
    asyncio.run(main())

Paso 5: Implementar capacidades de consulta avanzadas

Crea un recurso que permita consultas más flexibles con parámetros:

async def fetch_users_by_criteria(
    department: Optional[str] = None,
    role: Optional[str] = None,
    active: Optional[bool] = True,
    limit: int = 20
) -> List[Dict[str, Any]]:
    """
    Obtener usuarios que coincidan con criterios específicos.
    
    Args:
        department: Filtrar por departamento (opcional)
        role: Filtrar por rol (opcional)
        active: Filtrar por estado activo (por defecto: True)
        limit: Máximo de resultados a devolver (por defecto: 20)
        
    Returns:
        Lista de objetos de usuario coincidentes
    """
    try:
        if not db_pool:
            logger.error("Pool de base de datos no inicializado")
            return {"error": "Conexión a base de datos no disponible"}
            
        # Construir consulta dinámica
        conditions = ["active = $1"]
        params = [active]
        param_count = 1
        
        if department:
            param_count += 1
            conditions.append(f"department = ${param_count}")
            params.append(department)
            
        if role:
            param_count += 1
            conditions.append(f"role = ${param_count}")
            params.append(role)
            
        # Construir la consulta final
        query = f"""
            SELECT id, username, email, department, role, created_at 
            FROM users 
            WHERE {' AND '.join(conditions)}
            ORDER BY created_at DESC 
            LIMIT ${param_count + 1};
        """
        params.append(limit)
        
        async with db_pool.acquire() as connection:
            rows = await connection.fetch(query, *params)
            
            # Convertir a diccionarios y manejar serialización de datetime
            users = []
            for row in rows:
                user = dict(row)
                if "created_at" in user and user["created_at"]:
                    user["created_at"] = user["created_at"].isoformat()
                users.append(user)
                
            logger.info(f"Obtenidos {len(users)} usuarios que coinciden con los criterios")
            return users
    except Exception as e:
        logger.error(f"Error al obtener usuarios por criterios: {e}")
        return {"error": f"Error de base de datos: {str(e)}"}

Registra esto como un recurso parametrizado:

@server.resource(
    name="users_by_criteria", 
    description="Obtener usuarios que coincidan con criterios específicos como departamento o rol."
)
async def users_by_criteria_resource(data: dict):
    return await fetch_users_by_criteria(
        department=data.get("department"),
        role=data.get("role"),
        active=data.get("active", True),
        limit=data.get("limit", 20)
    )

Esto permite al LLM solicitar subconjuntos específicos de usuarios según las necesidades del negocio.

Paso 6: Crear una herramienta segura para insertar nuevos registros

Crea tools/users.py para operaciones de escritura:

from typing import Dict, Any, Optional
from pydantic import BaseModel, EmailStr, Field, validator
import logging
import re
from database import db_pool

logger = logging.getLogger("mcp_database.tools.users")

class CreateUserRequest(BaseModel):
    """Modelo de validación para solicitudes de creación de usuarios."""
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr
    department: Optional[str] = "General"
    role: Optional[str] = "User"
    
    @validator('username')
    def username_alphanumeric(cls, v):
        if not re.match(r'^[a-zA-Z0-9_]+$', v):
            raise ValueError('El nombre de usuario debe ser alfanumérico')
        return v

async def create_user(data: Dict[str, Any]) -> Dict[str, Any]:
    """
    Crear un nuevo usuario en la base de datos.
    
    Args:
        data: Datos del usuario que contienen username, email, etc.
        
    Returns:
        Respuesta con estado e información del usuario
    """
    try:
        # Validar los datos de entrada con Pydantic
        user_data = CreateUserRequest(**data)
        
        if not db_pool:
            logger.error("Pool de base de datos no inicializado")
            return {
                "status": "error",
                "message": "Conexión a base de datos no disponible"
            }
        
        async with db_pool.acquire() as connection:
            # Comprobar si el usuario ya existe
            existing_user = await connection.fetchrow(
                "SELECT id FROM users WHERE username = $1 OR email = $2",
                user_data.username,
                user_data.email
            )
            
            if existing_user:
                return {
                    "status": "error",
                    "message": "Ya existe un usuario con este nombre de usuario o email"
                }
            
            # Insertar el nuevo usuario
            query = """
                INSERT INTO users (username, email, department, role) 
                VALUES ($1, $2, $3, $4)
                RETURNING id;
            """
            
            user_id = await connection.fetchval(
                query,
                user_data.username,
                user_data.email,
                user_data.department,
                user_data.role
            )
            
            logger.info(f"Creado nuevo usuario: {user_data.username} (ID: {user_id})")
            
            return {
                "status": "success",
                "message": f"Usuario {user_data.username} creado exitosamente",
                "user_id": user_id
            }
    except Exception as e:
        logger.error(f"Error al crear usuario: {e}")
        return {
            "status": "error",
            "message": f"Error al crear usuario: {str(e)}"
        }

Registra esta herramienta en server.py:

from tools.users import create_user

# ... código existente ...

# Registrar la herramienta
@server.tool(
    name="create_user",
    description="Crear un nuevo usuario en la base de datos."
)
async def create_user_tool(data: dict):
    return await create_user(data)

Características de seguridad clave añadidas: Validación con Pydantic con restricciones claras, Validación de email usando EmailStr, Validación de formato de nombre de usuario con regex, Comprobación de duplicados antes de la inserción, Mensajes de error claros, Logging completo, Consultas parametrizadas para prevenir inyección SQL.

Paso 7: Gestión de transacciones

Para operaciones que requieren múltiples cambios en la base de datos, usa transacciones:

async def transfer_user_to_department(
    user_id: int,
    new_department: str
) -> Dict[str, Any]:
    """
    Transferir un usuario a un nuevo departamento, registrando el cambio en el log de auditoría.
    
    Args:
        user_id: ID del usuario a transferir
        new_department: Nombre del departamento destino
        
    Returns:
        Estado de la operación
    """
    try:
        if not db_pool:
            return {"error": "Conexión a base de datos no disponible"}
            
        async with db_pool.acquire() as connection:
            # Iniciar una transacción
            async with connection.transaction():
                # Obtener departamento actual
                current_dept = await connection.fetchval(
                    "SELECT department FROM users WHERE id = $1",
                    user_id
                )
                
                if not current_dept:
                    return {"error": "Usuario no encontrado"}
                
                # Actualizar el departamento del usuario
                await connection.execute(
                    "UPDATE users SET department = $1 WHERE id = $2",
                    new_department,
                    user_id
                )
                
                # Registrar el cambio en el log de auditoría
                await connection.execute(
                    """
                    INSERT INTO user_audit_log 
                    (user_id, field_changed, old_value, new_value) 
                    VALUES ($1, $2, $3, $4)
                    """,
                    user_id,
                    "department",
                    current_dept,
                    new_department
                )
                
                logger.info(f"Usuario {user_id} transferido de {current_dept} a {new_department}")
                
                return {
                    "status": "success",
                    "message": f"Usuario transferido de {current_dept} a {new_department}"
                }
    except Exception as e:
        logger.error(f"Error al transferir usuario: {e}")
        return {"error": f"Transferencia fallida: {str(e)}"}

Registra esto como una herramienta:

@server.tool(
    name="transfer_user",
    description="Transferir un usuario a un nuevo departamento."
)
async def transfer_user_tool(data: dict):
    user_id = data.get("user_id")
    new_department = data.get("new_department")
    
    if not user_id or not new_department:
        return {"error": "Falta user_id o new_department"}
        
    return await transfer_user_to_department(user_id, new_department)

Esto asegura que ambas operaciones (actualizar usuario + añadir log de auditoría) tengan éxito o fallen juntas.

Paso 8: Ejemplo completo de código del servidor

Aquí hay un ejemplo completo que reúne todo:

import asyncio
import logging
from mcp_server import MCPServer
from database import init_db, close_db, db_pool
from resources.users import fetch_recent_users, fetch_users_by_criteria
from tools.users import create_user, transfer_user_to_department

# Configurar logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mcp_server")

# Crear servidor MCP
server = MCPServer(
    name="DatabaseMCPServer",
    version="1.0.0",
    description="Servidor MCP con acceso a base de datos PostgreSQL"
)

# Registrar recursos
@server.resource(
    name="recent_users", 
    description="Obtener los usuarios más recientes de la base de datos."
)
async def recent_users_resource():
    return await fetch_recent_users(limit=20)

@server.resource(
    name="users_by_criteria", 
    description="Obtener usuarios que coincidan con criterios específicos como departamento o rol."
)
async def users_by_criteria_resource(data: dict):
    return await fetch_users_by_criteria(
        department=data.get("department"),
        role=data.get("role"),
        active=data.get("active", True),
        limit=data.get("limit", 20)
    )

# Registrar herramientas
@server.tool(
    name="create_user",
    description="Crear un nuevo usuario en la base de datos."
)
async def create_user_tool(data: dict):
    return await create_user(data)

@server.tool(
    name="transfer_user",
    description="Transferir un usuario a un nuevo departamento."
)
async def transfer_user_tool(data: dict):
    user_id = data.get("user_id")
    new_department = data.get("new_department")
    
    if not user_id or not new_department:
        return {"error": "Falta user_id o new_department"}
        
    return await transfer_user_to_department(user_id, new_department)

async def main():
    try:
        # Inicializar la base de datos
        await init_db()
        
        # Iniciar el servidor
        logger.info("Iniciando servidor MCP...")
        await server.start()
    except Exception as e:
        logger.error(f"Error del servidor: {e}")
    finally:
        # Cerrar conexiones de base de datos al apagar
        await close_db()

if __name__ == "__main__":
    asyncio.run(main())

Consideraciones de seguridad

Al conectar tu servidor MCP a una base de datos, sigue estas mejores prácticas de seguridad:

1. Usa un usuario de base de datos dedicado con permisos limitados:

   CREATE USER mcp_user WITH PASSWORD 'secure_password';
   GRANT SELECT, INSERT ON users TO mcp_user;
   -- Otorga solo lo necesario
   

2. Valida todas las entradas usando modelos Pydantic con reglas de validación estrictas.

3. Usa consultas parametrizadas exclusivamente para prevenir inyección SQL.

4. Implementa autenticación para tu servidor MCP:

   @server.middleware
   async def auth_middleware(request, next_handler):
       token = request.headers.get("Authorization")
       if not token or not verify_token(token):
           return {"error": "No autorizado"}
       return await next_handler(request)
   

5. Implementa limitación de velocidad para prevenir abusos:

   # Limitador de velocidad simple en memoria
   request_counts = {}
   
   @server.middleware
   async def rate_limit_middleware(request, next_handler):
       client_id = request.client.host
       current_time = time.time()
       
       # Limpiar entradas antiguas
       request_counts = {k: v for k, v in request_counts.items() 
                        if v["timestamp"] > current_time - 60}
       
       if client_id in request_counts:
           if request_counts[client_id]["count"] > 100:  # 100 solicitudes por minuto
               return {"error": "Límite de velocidad excedido"}
           request_counts[client_id]["count"] += 1
       else:
           request_counts[client_id] = {"count": 1, "timestamp": current_time}
           
       return await next_handler(request)
   

6. Implementa registro de solicitudes para auditoría:

   @server.middleware
   async def logging_middleware(request, next_handler):
       logger.info(f"Solicitud: {request.method} {request.path} desde {request.client.host}")
       response = await next_handler(request)
       return response
   

7. Configura tiempos de espera para consultas de base de datos para evitar que consultas de larga duración afecten el rendimiento del servidor.

Optimización del rendimiento

1. El pooling de conexiones ya está implementado, pero ajusta los parámetros según tu carga de trabajo:

   db_pool = await asyncpg.create_pool(
       **DB_CONFIG,
       min_size=5,       # Establece más alto para escenarios de alta carga
       max_size=20,      # Ajusta según la capacidad de tu servidor de base de datos
       statement_cache_size=100,  # Caché de declaraciones preparadas
       max_inactive_connection_lifetime=300  # Segundos antes de reciclar conexiones inactivas
   )
   

2. Crea índices de base de datos para campos consultados frecuentemente:

   CREATE INDEX idx_users_department ON users(department);
   CREATE INDEX idx_users_created_at ON users(created_at DESC);
   

3. Implementa caché de resultados para datos frecuentemente accedidos y que cambian raramente:

   from functools import lru_cache
   from datetime import datetime, timedelta
   
   # Caché que expira después de 5 minutos
   cache_time = None
   cached_result = None
   
   async def fetch_departments_with_caching():
       global cache_time, cached_result
       
       # Comprobar si la caché es válida
       current_time = datetime.now()
       if cache_time and cached_result and current_time - cache_time < timedelta(minutes=5):
           return cached_result
           
       # Caché no válida - obtener de la base de datos
       async with db_pool.acquire() as connection:
           result = await connection.fetch("SELECT * FROM departments")
           
       # Actualizar caché
       cache_time = current_time
       cached_result = [dict(row) for row in result]
       
       return cached_result
   

4. Usa JSONB para datos complejos que no necesitan ser consultados extensivamente:

   CREATE TABLE user_preferences (
       user_id INTEGER PRIMARY KEY,
       preferences JSONB NOT NULL
   );
   

5. Paginación para conjuntos de resultados grandes:

   async def fetch_paginated_users(page: int = 1, page_size: int = 20):
       offset = (page - 1) * page_size
       
       async with db_pool.acquire() as connection:
           # Obtener conteo total
           total = await connection.fetchval("SELECT COUNT(*) FROM users")
           
           # Obtener resultados paginados
           rows = await connection.fetch(
               "SELECT * FROM users ORDER BY created_at DESC LIMIT $1 OFFSET $2",
               page_size, offset
           )
           
           return {
               "users": [dict(row) for row in rows],
               "pagination": {
                   "total": total,
                   "page": page,
                   "page_size": page_size,
                   "pages": (total + page_size - 1) // page_size
               }
           }
   

Pruebas de interacciones con la base de datos

Crea un directorio tests con un archivo de prueba:

# tests/test_database.py
import asyncio
import pytest
import os
from dotenv import load_dotenv
import asyncpg

# Load test environment variables
load_dotenv(".env.test")

# Testing database configuration
TEST_DB_CONFIG = {
    "user": os.getenv("TEST_DB_USER"),
    "password": os.getenv("TEST_DB_PASSWORD"),
    "database": os.getenv("TEST_DB_NAME"),
    "host": os.getenv("TEST_DB_HOST", "localhost"),
    "port": int(os.getenv("TEST_DB_PORT", "5432")),
}

@pytest.fixture
async def db_pool():
    """Create a test database pool."""
    pool = await asyncpg.create_pool(**TEST_DB_CONFIG)
    yield pool
    await pool.close()

@pytest.fixture
async def setup_test_data(db_pool):
    """Set up test data before tests and clean up after."""
    async with db_pool.acquire() as conn:
        # Create test tables
        await conn.execute("""
            CREATE TEMPORARY TABLE users (
                id SERIAL PRIMARY KEY,
                username TEXT UNIQUE NOT NULL,
                email TEXT UNIQUE NOT NULL,
                department TEXT NOT NULL DEFAULT 'General',
                role TEXT NOT NULL DEFAULT 'User',
                active BOOLEAN NOT NULL DEFAULT TRUE,
                created_at TIMESTAMP NOT NULL DEFAULT NOW()
            )
        """)
        
        # Insert test users
        await conn.executemany(
            """
            INSERT INTO users (username, email, department, role) 
            VALUES ($1, $2, $3, $4)
            """,
            [
                ("test1", "test1@example.com", "Engineering", "Developer"),
                ("test2", "test2@example.com", "Marketing", "Manager"),
                ("test3", "test3@example.com", "Sales", "Representative")
            ]
        )
    
    yield
    
    # Cleanup happens automatically for temporary tables

@pytest.mark.asyncio
async def test_fetch_recent_users(db_pool, setup_test_data):
    """Test that fetching recent users works correctly."""
    from resources.users import fetch_recent_users
    
    # Monkeypatch db_pool
    import database
    database.db_pool = db_pool
    
    # Execute the function
    result = await fetch_recent_users(limit=10)
    
    # Assertions
    assert isinstance(result, list)
    assert len(result) == 3
    assert result[0]["username"] in ["test1", "test2", "test3"]
    assert "email" in result[0]
    assert "department" in result[0]

@pytest.mark.asyncio
async def test_create_user(db_pool, setup_test_data):
    """Test that creating a user works correctly."""
    from tools.users import create_user
    
    # Monkeypatch db_pool
    import database
    database.db_pool = db_pool
    
    # Test data
    user_data = {
        "username": "newuser",
        "email": "newuser@example.com",
        "department": "Finance",
        "role": "Analyst"
    }
    
    # Execute the function
    result = await create_user(user_data)
    
    # Assertions
    assert result["status"] == "success"
    
    # Verify in database
    async with db_pool.acquire() as conn:
        user = await conn.fetchrow(
            "SELECT * FROM users WHERE username = $1",
            "newuser"
        )
        
    assert user is not None
    assert user["email"] == "newuser@example.com"
    assert user["department"] == "Finance"

Ejecuta las pruebas con: pytest -xvs tests/

Consideraciones para el despliegue

1. Usa archivos de configuración específicos para cada entorno: .env.development, .env.staging, .env.production

2. Configura migraciones de base de datos para cambios de esquema:

   pip install alembic
   alembic init migrations

3. Despliega con Docker para mantener consistencia:

   FROM python:3.10-slim
   
   WORKDIR /app
   
   COPY requirements.txt .
   RUN pip install --no-cache-dir -r requirements.txt
   
   COPY . .
   
   CMD ["python", "server.py"]

4. Configura comprobaciones de salud:

   @server.route("/health")
   async def health_check():
       try:
           async with db_pool.acquire() as connection:
               await connection.fetchval("SELECT 1")
           return {"status": "healthy", "database": "connected"}
       except Exception as e:
           return {"status": "unhealthy", "error": str(e)}

5. Monitoriza el rendimiento de las consultas:

   -- En PostgreSQL
   CREATE EXTENSION pg_stat_statements;
   
   -- Para analizar consultas lentas
   SELECT query, calls, total_time, mean_time
   FROM pg_stat_statements
   ORDER BY mean_time DESC
   LIMIT 10;

6. Configura copias de seguridad de la base de datos:

   # Script de copia de seguridad diaria
   pg_dump -U user -d database -F c -f /backups/db_$(date +%Y%m%d).dump

Conclusión

Conectar tu servidor MCP a una base de datos PostgreSQL lo transforma de una simple demostración a un backend de IA listo para producción. Con una atención cuidadosa a la seguridad, el rendimiento y la fiabilidad, puedes exponer de forma segura tu base de datos a modelos de lenguaje, habilitando flujos de trabajo potentes impulsados por IA. Principios clave a recordar: Recursos para operaciones de lectura, Herramientas para operaciones de escritura, Validación para todas las entradas, Transacciones para consistencia, Agrupación de conexiones para rendimiento, Variables de entorno para seguridad, Manejo estructurado de errores para fiabilidad. Siguiendo estas prácticas, permites que los LLMs trabajen junto con tus aplicaciones de base de datos existentes de manera segura y controlada, desbloqueando nuevas posibilidades para flujos de trabajo asistidos por IA y análisis de datos. En la próxima guía, mostraremos cómo conectar tu backend API existente a través de MCP sin reescribir la lógica de negocio.

Preguntas frecuentes