Construir un Servidor MCP para IBM Db2 12.1 en Docker

Este tutorial usa el driver Python ibm_db para conectarse directamente a Db2 — simple, sin intermediarios, y compatible con 12.1.

Arquitectura

Claude Code
    │
    │  JSON-RPC (stdio)
    ▼
Servidor MCP  (Python, server.py)
    │
    │  TCP 50000  (ibm_db / clidriver)
    ▼
IBM Db2 12.1 Community Edition  (Docker, puerto 50000)

Sin intermediarios. El ibm_db incluye el clidriver de IBM y se conecta directamente al puerto 50000 de Db2.

Requisitos previos

• Docker y Docker Compose
• Python 3.10+
• uv — pip install uv
• Claude Code

Parte 1 — Entorno Docker

1.1 docker-compose.yml

Solo un contenedor — el Db2. No se crea ninguna base de datos al arrancar; db2sampl en el paso siguiente crea y popula SAMPLE con las tablas de ejemplo.

# docker-compose.yml
services:

  db2:
    image: icr.io/db2_community/db2:latest
    container_name: db2
    privileged: true
    environment:
      LICENSE: accept
      DB2INST1_PASSWORD: passw0rd
    ports:
      - "50000:50000"
    volumes:
      - db2_data:/database
    healthcheck:
      test: ["CMD", "su", "-", "db2inst1", "-c", "db2 get instance"]
      interval: 30s
      timeout: 10s
      retries: 10
      start_period: 120s

volumes:
  db2_data:

1.2 Arrancar

mkdir db2-mcp-tutorial && cd db2-mcp-tutorial
docker compose up -d

# Esperar inicialización (~2-3 min)
docker compose logs -f db2
# Cuando aparezca "Setup has completed" — listo

1.3 Popular la base de datos

Crear la base de datos SAMPLE con todas las tablas de ejemplo (EMPLOYEE, DEPARTMENT, etc.):

docker exec -it db2 su - db2inst1 -c "db2sampl"

Parte 2 — El Servidor MCP

2.1 Estructura del proyecto

db2-mcp-tutorial/
├── docker-compose.yml
├── server.py
├── pyproject.toml
└── .env

2.2 Dependencias

# pyproject.toml
[project]
name = "db2-schema-mcp"
version = "0.1.0"
requires-python = ">=3.10"
dependencies = [
    "mcp[cli]>=1.25",
    "ibm-db>=3.2.0",
    "python-dotenv>=1.0",
]

2.3 Instalar

El ibm_db descarga e instala el clidriver de IBM automáticamente durante el pip install. No es necesario instalar el Db2 Client por separado.

uv venv
uv pip install -e .

# Verificar que el driver se instaló correctamente
uv run python -c "import ibm_db; print('ibm_db OK')"

Si estás en un sistema Linux x86_64 y necesitas específicamente el clidriver 12.1 (opcional — el 11.5 se conecta bien a un servidor 12.1):

export CLIDRIVER_VERSION=v12.1.0
uv pip install ibm-db --no-binary :all: --no-cache-dir

2.4 Credenciales

# .env
DB2_HOST=localhost
DB2_PORT=50000
DB2_DBNAME=SAMPLE
DB2_USER=db2inst1
DB2_PASSWORD=passw0rd

2.5 El servidor — server.py (completo)

"""
Servidor MCP — Db2 Schema Explorer
Explora tablas, columnas e índices de una base de datos Db2 12.1
mediante ibm_db.
"""
import os
import sys
import threading
import ibm_db
import ibm_db_dbi
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

load_dotenv()

# El transporte stdio de MCP usa stdout exclusivamente para JSON-RPC.
# El CLIdriver IBM puede emitir mensajes de diagnóstico en stderr
# que aparecen en el terminal. Redirigirlos a un fichero evita el ruido.
_log = open(os.path.expanduser("~/.db2-mcp-server.log"), "a", buffering=1)
sys.stderr = _log

# ── Configuración ──────────────────────────────────────────────────────────────

DB2_HOST   = os.getenv("DB2_HOST",     "localhost")
DB2_PORT   = os.getenv("DB2_PORT",     "50000")
DB2_DBNAME = os.getenv("DB2_DBNAME",   "SAMPLE")
DB2_USER   = os.getenv("DB2_USER",     "db2inst1")
DB2_PASS   = os.getenv("DB2_PASSWORD", "passw0rd")

# Cadena de conexión en formato DSN de ibm_db
# DIAGLEVEL=0 suprime los mensajes de diagnóstico del CLIdriver en el terminal.
_DSN = (
    f"DATABASE={DB2_DBNAME};"
    f"HOSTNAME={DB2_HOST};"
    f"PORT={DB2_PORT};"
    f"PROTOCOL=TCPIP;"
    f"UID={DB2_USER};"
    f"PWD={DB2_PASS};"
    "CONNECTTIMEOUT=30;"
    "QUERYTIMEOUT=120;"
    "DIAGLEVEL=0;"
)

# ── Pool de conexión minimalista (thread-safe) ─────────────────────────────────
# ibm_db no es thread-safe por conexión — usamos un lock para serializar.
# Para producción considerar un pool más robusto.

_conn_lock = threading.Lock()
_conn      = None


def _get_conn():
    """Devuelve la conexión activa, reconectando si es necesario."""
    global _conn
    try:
        # Verificar si la conexión sigue activa
        if _conn is not None:
            ibm_db.active(_conn)  # lanza excepción si está caída
            return _conn
    except Exception:
        _conn = None

    _conn = ibm_db.connect(_DSN, "", "")
    return _conn


def run_sql(sql: str) -> list[dict]:
    """
    Ejecuta un SELECT y devuelve los resultados como lista de diccionarios.
    Thread-safe mediante lock global.
    """
    with _conn_lock:
        conn  = _get_conn()
        stmt  = ibm_db.exec_immediate(conn, sql)
        rows  = []
        row   = ibm_db.fetch_assoc(stmt)
        while row:
            rows.append(dict(row))
            row = ibm_db.fetch_assoc(stmt)
        ibm_db.free_result(stmt)
        return rows


# ── Servidor MCP ──────────────────────────────────────────────────────────────

mcp = FastMCP("db2-schema-explorer")


@mcp.tool()
def list_schemas() -> str:
    """Lista todos los schemas de la base de datos (excluyendo schemas de sistema)."""
    rows = run_sql("""
        SELECT SCHEMANAME, OWNER, CREATE_TIME
        FROM SYSCAT.SCHEMATA
        WHERE SCHEMANAME NOT LIKE 'SYS%'
          AND SCHEMANAME NOT IN ('NULLID', 'SQLJ', 'ERRORSCHEMA')
        ORDER BY SCHEMANAME
    """)

    if not rows:
        return "Sin schemas de usuario encontrados."

    lines = [f"**Schemas disponibles** ({len(rows)} encontrados)\n"]
    for r in rows:
        lines.append(f"- `{r['SCHEMANAME']}` (owner: {r.get('OWNER', '?')})")
    return "\n".join(lines)


@mcp.tool()
def list_tables(schema: str) -> str:
    """
    Lista todas las tablas de un schema.

    Parámetros:
        schema — nombre del schema (ej: SAMPLE, DB2INST1)
    """
    schema = schema.upper()
    rows = run_sql(f"""
        SELECT TABNAME, TYPE, CARD, NPAGES, LASTUSED
        FROM SYSCAT.TABLES
        WHERE TABSCHEMA = '{schema}'
          AND TYPE = 'T'
        ORDER BY TABNAME
    """)

    if not rows:
        return f"Ninguna tabla encontrada en el schema `{schema}`."

    lines = [f"**Tablas en `{schema}`** ({len(rows)} encontradas)\n",
             "| Tabla | Filas | Páginas | Último acceso |",
             "|-------|-------|---------|---------------|"]
    for r in rows:
        card     = r.get("CARD",     "?")
        npages   = r.get("NPAGES",   "?")
        lastused = str(r.get("LASTUSED", "—"))[:10]
        lines.append(f"| `{r['TABNAME']}` | {card} | {npages} | {lastused} |")
    return "\n".join(lines)


@mcp.tool()
def describe_table(schema: str, table: str) -> str:
    """
    Describe la estructura de una tabla: columnas, tipos, nullable y defaults.

    Parámetros:
        schema — schema de la tabla
        table  — nombre de la tabla
    """
    schema = schema.upper()
    table  = table.upper()

    rows = run_sql(f"""
        SELECT COLNAME, TYPENAME, LENGTH, SCALE,
               NULLS, DEFAULT, REMARKS
        FROM SYSCAT.COLUMNS
        WHERE TABSCHEMA = '{schema}'
          AND TABNAME   = '{table}'
        ORDER BY COLNO
    """)

    if not rows:
        return f"Tabla `{schema}.{table}` no encontrada o sin columnas."

    lines = [f"**`{schema}.{table}`** — {len(rows)} columnas\n",
             "| # | Columna | Tipo | Nullable | Default |",
             "|---|---------|------|----------|---------|"]
    for i, r in enumerate(rows, 1):
        length = f"({r['LENGTH']})" if r.get("LENGTH") else ""
        scale  = f",{r['SCALE']}"   if r.get("SCALE")  else ""
        tipo   = f"{r['TYPENAME']}{length}{scale}"
        null_  = "Sí" if r.get("NULLS") == "Y" else "**No**"
        dflt   = r.get("DEFAULT") or "—"
        lines.append(f"| {i} | `{r['COLNAME']}` | `{tipo}` | {null_} | {dflt} |")
    return "\n".join(lines)


@mcp.tool()
def list_indexes(schema: str, table: str) -> str:
    """
    Lista los índices de una tabla.

    Parámetros:
        schema — schema de la tabla
        table  — nombre de la tabla
    """
    schema = schema.upper()
    table  = table.upper()

    rows = run_sql(f"""
        SELECT I.INDNAME, I.UNIQUERULE, I.INDEXTYPE,
               I.NLEAF, I.NLEVELS,
               LISTAGG(K.COLNAME, ', ') WITHIN GROUP (ORDER BY K.COLSEQ) AS COLUMNS
        FROM SYSCAT.INDEXES I
        JOIN SYSCAT.INDEXCOLUSE K
          ON I.INDSCHEMA = K.INDSCHEMA
         AND I.INDNAME   = K.INDNAME
        WHERE I.TABSCHEMA = '{schema}'
          AND I.TABNAME   = '{table}'
        GROUP BY I.INDNAME, I.UNIQUERULE, I.INDEXTYPE, I.NLEAF, I.NLEVELS
        ORDER BY I.INDNAME
    """)

    if not rows:
        return f"Ningún índice encontrado en `{schema}.{table}`."

    lines = [f"**Índices en `{schema}.{table}`** ({len(rows)} encontrados)\n",
             "| Índice | Tipo | Único | Hojas | Niveles | Columnas |",
             "|--------|------|-------|-------|---------|----------|"]
    unique_map = {"U": "✅ Sí", "P": "✅ PK", "D": "No"}
    for r in rows:
        unique = unique_map.get(r.get("UNIQUERULE", "D"), "?")
        lines.append(
            f"| `{r['INDNAME']}` | {r.get('INDEXTYPE','?')} "
            f"| {unique} | {r.get('NLEAF','?')} | {r.get('NLEVELS','?')} "
            f"| `{r.get('COLUMNS','?')}` |"
        )
    return "\n".join(lines)


@mcp.tool()
def sample_rows(schema: str, table: str, limit: int = 5) -> str:
    """
    Muestra las primeras N filas de una tabla.

    Parámetros:
        schema — schema de la tabla
        table  — nombre de la tabla
        limit  — número de filas a mostrar (máximo 20)
    """
    schema = schema.upper()
    table  = table.upper()
    limit  = min(limit, 20)

    rows = run_sql(f"""
        SELECT * FROM {schema}.{table}
        FETCH FIRST {limit} ROWS ONLY
    """)

    if not rows:
        return f"La tabla `{schema}.{table}` está vacía o no existe."

    headers = list(rows[0].keys())
    lines   = [f"**`{schema}.{table}`** — primeras {len(rows)} filas\n",
               "| " + " | ".join(headers) + " |",
               "| " + " | ".join(["---"] * len(headers)) + " |"]
    for r in rows:
        cells = [str(r.get(h, "")) for h in headers]
        lines.append("| " + " | ".join(cells) + " |")
    return "\n".join(lines)


# ── Resource ──────────────────────────────────────────────────────────────────

@mcp.resource("db2://schema/{schema}/overview")
def schema_overview(schema: str) -> str:
    """Visión general de un schema: número de tablas, filas y espacio."""
    schema = schema.upper()
    rows = run_sql(f"""
        SELECT COUNT(*)    AS NUM_TABLES,
               SUM(CARD)   AS TOTAL_ROWS,
               SUM(NPAGES) AS TOTAL_PAGES
        FROM SYSCAT.TABLES
        WHERE TABSCHEMA = '{schema}'
          AND TYPE = 'T'
    """)
    if not rows or not rows[0].get("NUM_TABLES"):
        return f"Schema `{schema}` no encontrado o vacío."
    r = rows[0]
    return (
        f"Schema: {schema}\n"
        f"Tablas: {r.get('NUM_TABLES','?')}\n"
        f"Total filas: {r.get('TOTAL_ROWS','?')}\n"
        f"Total páginas: {r.get('TOTAL_PAGES','?')}\n"
    )


# ── Entry point ───────────────────────────────────────────────────────────────

if __name__ == "__main__":
    # Probar la conexión antes de arrancar el servidor
    try:
        test = run_sql("SELECT 1 AS OK FROM SYSIBM.SYSDUMMY1")
        print(f"✅ Conexión a Db2 OK ({DB2_HOST}:{DB2_PORT}/{DB2_DBNAME})")
    except Exception as e:
        print(f"❌ Fallo en la conexión a Db2: {e}")
        raise

    mcp.run(transport="stdio")

Parte 3 — Conectar a Claude Code

# Registrar
claude mcp add db2-schema \
  --command uv \
  --args run server.py

# Verificar
claude mcp list
# db2-schema   stdio   uv run server.py

# Iniciar sesión
claude

Parte 4 — Resolución de problemas comunes

SQL1598N — db2connect license

SQL1598N  An attempt to connect to the database server failed because
          of a licensing problem.

Este error indica que el servidor Db2 necesita ser activado. En el contenedor Db2 Community Edition, activar:

docker exec -it db2 su - db2inst1 -c "db2connectactivate -u db2inst1 -p passw0rd"

Si no funciona, Db2 Community Edition no requiere db2connect para conexiones locales — pero para conexiones TCP remotas puede ser necesario el archivo de licencia. Alternativa: conectar por socket Unix en lugar de TCP:

# .env — conexión mediante socket (dentro del mismo host que Docker)
DB2_HOST=localhost
DB2_PORT=50000

SQL0805N — CLI packages sin bind

docker exec -it db2 su - db2inst1 -c \
  "db2 bind /home/db2inst1/sqllib/bnd/@db2cli.lst blocking all grant public"

ibm_db no instala (falla la compilación)

# Instalar dependencias de compilación primero
sudo apt-get install -y build-essential python3-dev
# Después
uv pip install ibm-db --no-binary :all:

Probar la conexión directamente

# test_conn.py
import ibm_db
conn = ibm_db.connect(
    "DATABASE=SAMPLE;HOSTNAME=localhost;PORT=50000;"
    "PROTOCOL=TCPIP;UID=db2inst1;PWD=passw0rd;",
    "", ""
)
stmt = ibm_db.exec_immediate(conn, "SELECT 1 FROM SYSIBM.SYSDUMMY1")
row  = ibm_db.fetch_assoc(stmt)
print("OK:", row)
ibm_db.close(conn)

uv run python test_conn.py
# OK: {'1': 1}

Parte 5 — Ampliar el servidor

El patrón es simple — basta con decorar con @mcp.tool():

@mcp.tool()
def search_columns(column_name: str) -> str:
    """
    Encuentra todas las tablas que tienen una columna con este nombre.

    Parámetros:
        column_name — nombre o parte del nombre de la columna
    """
    name = column_name.upper()
    rows = run_sql(f"""
        SELECT TABSCHEMA, TABNAME, COLNAME, TYPENAME
        FROM SYSCAT.COLUMNS
        WHERE COLNAME LIKE '%{name}%'
          AND TABSCHEMA NOT LIKE 'SYS%'
        ORDER BY TABSCHEMA, TABNAME, COLNAME
        FETCH FIRST 50 ROWS ONLY
    """)
    if not rows:
        return f"Ninguna columna con `{name}` en el nombre."

    lines = [f"**Columnas con `{name}`** ({len(rows)} encontradas)\n",
             "| Schema | Tabla | Columna | Tipo |",
             "|--------|-------|---------|------|"]
    for r in rows:
        lines.append(
            f"| `{r['TABSCHEMA']}` | `{r['TABNAME']}` "
            f"| `{r['COLNAME']}` | {r['TYPENAME']} |"
        )
    return "\n".join(lines)

Resumen