Integraciones de bases de datos con MCP

Introducción a Model Context Protocol (MCP)

James Chapman

AI Curriculum Manager, DataCamp

¿Por qué añadir una base de datos al servidor de zonas horarias?

Potencia de consulta: filtra, busca y ordena sin cargar todo el conjunto
Escala: búsquedas rápidas con índices de la base de datos

Conexión a base de datos

¿Por qué añadir una base de datos al servidor de zonas horarias?

Potencia de consulta: filtra, busca y ordena sin cargar todo el conjunto
Escala: búsquedas rápidas con índices de la base de datos
Concurrencia: varias llamadas de herramientas o clientes acceden a la BD con seguridad
Listo para producción: copias de seguridad, replicación y fuente única de verdad

Ventajas de la base de datos

Acceso a datos: herramientas vs. recursos

Icono de herramientas

Dinámico, muy dependiente de la entrada del usuario, operaciones de escritura
Ejemplos: analítica y operaciones de datos

Icono de recursos

Solo lectura, datos estáticos, información de referencia
Ejemplos: guías, documentación, políticas

Ciclo de vida de la conexión

Conexiones de base de datos

Recurso de base de datos: timezone_server.py

from mcp.server.fastmcp import FastMCP
import sqlite3

mcp = FastMCP("Timezone Converter")

# Create connection at startup; reuse in handlers
conn = sqlite3.connect("timezones.db")
conn.row_factory = sqlite3.Row

@mcp.resource("db://timezones")
def get_locations() -> str:
    try:
        cursor = conn.execute("SELECT timezone FROM locations")
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"

Herramienta de búsqueda en BD: timezone_server.py

@mcp.tool()
def lookup_locations(prefix: str) -> str:
    """Find timezones that contain the given prefix."""
    try:
        cursor = conn.execute(
            "SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50", (f"%{prefix}%",))
        rows = cursor.fetchall()
        return "\n".join(r["timezone"] for r in rows)
    except sqlite3.Error as e: return f"Error: {e}"


if __name__ == "__main__":
    try:
        mcp.run(transport="stdio")
    finally:
        conn.close()

Seguridad y buenas prácticas en producción

Consultas parametrizadas (?): nunca formatees en SQL entradas de usuarios o del LLM
- Inyección por prompt → inyección maliciosa de código vía prompts

SELECT timezone FROM locations WHERE timezone LIKE ? LIMIT 50

Prefiere acceso a la BD de solo lectura y con alcance reducido para el servidor MCP
Aplica límites: máx. de filas y timeouts

¡Vamos a practicar!

Introducción a Model Context Protocol (MCP)