Construir um Servidor MCP para IBM Db2 12.1 em Docker

Este tutorial usa o driver Python ibm_db para ligar directamente ao Db2 — mais simples, sem intermediários, e compatível com 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, porta 50000)

Sem intermediários. O ibm_db inclui o clidriver IBM e liga directamente ao porto 50000 do Db2.

Pré-requisitos

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

Parte 1 — Ambiente Docker

1.1 docker-compose.yml

Apenas um container — o Db2. Não criamos a base de dados no arranque; o db2sampl no passo seguinte cria e popula a SAMPLE com as tabelas de exemplo.

# 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

# Aguardar inicialização (~2-3 min)
docker compose logs -f db2
# Quando aparecer "Setup has completed" — pronto

1.3 Popular a base de dados

Criar a base de dados SAMPLE com todas as tabelas de exemplo (EMPLOYEE, DEPARTMENT, etc.):

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

Parte 2 — O Servidor MCP

2.1 Criar o projecto

mkdir db2-mcp-tutorial && cd db2-mcp-tutorial

2.2 Criar o pyproject.toml

O ibm_db inclui o clidriver IBM — não é necessário instalar o Db2 Client separadamente.

cat > pyproject.toml << 'EOF'
[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",
]
EOF

2.3 Instalar dependências

uv venv
uv pip install -e .
uv run python -c "import ibm_db; print('ibm_db OK')"

2.4 Criar o ficheiro de credenciais

cat > .env << 'EOF'
DB2_HOST=localhost
DB2_PORT=50000
DB2_DBNAME=SAMPLE
DB2_USER=db2inst1
DB2_PASSWORD=passw0rd
EOF

2.5 Criar o servidor — server.py

cat > server.py << 'EOF'
"""
Servidor MCP — Db2 Schema Explorer
Explora tabelas, colunas e índices de uma base de dados Db2 12.1
via 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()

# O transporte stdio do MCP usa stdout exclusivamente para JSON-RPC.
# O CLIdriver IBM pode emitir mensagens de diagnóstico para stderr
# que aparecem no terminal. Redireccionar para ficheiro evita ruído.
_log = open(os.path.expanduser("~/.db2-mcp-server.log"), "a", buffering=1)
sys.stderr = _log

# ── Configuração ──────────────────────────────────────────────────────────────

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")

# Connection string no formato DSN do ibm_db
# DIAGLEVEL=0 suprime mensagens de diagnóstico do CLIdriver no 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 conexão minimalista (thread-safe) ─────────────────────────────────
# ibm_db não é thread-safe por conexão — usamos um lock para serializar.
# Para produção considerar um pool mais robusto.

_conn_lock = threading.Lock()
_conn      = None


def _get_conn():
    """Devolve a conexão activa, reconectando se necessário."""
    global _conn
    try:
        # Verificar se a conexão ainda está viva
        if _conn is not None:
            ibm_db.active(_conn)  # lança excepção se morta
            return _conn
    except Exception:
        _conn = None

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


def run_sql(sql: str) -> list[dict]:
    """
    Executa um SELECT e devolve os resultados como lista de dicionários.
    Thread-safe via 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 os schemas da base de dados (excluindo 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 "Sem schemas de utilizador encontrados."

    lines = [f"**Schemas disponíveis** ({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 as tabelas de um schema.

    Parâmetros:
        schema — nome do schema (ex: 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"Nenhuma tabela encontrada no schema `{schema}`."

    lines = [f"**Tabelas em `{schema}`** ({len(rows)} encontradas)\n",
             "| Tabela | Linhas | Páginas | Último acesso |",
             "|--------|--------|---------|---------------|"]
    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:
    """
    Descreve a estrutura de uma tabela: colunas, tipos, nullable e defaults.

    Parâmetros:
        schema — schema da tabela
        table  — nome da tabela
    """
    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"Tabela `{schema}.{table}` não encontrada ou sem colunas."

    lines = [f"**`{schema}.{table}`** — {len(rows)} colunas\n",
             "| # | Coluna | 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_  = "Sim" if r.get("NULLS") == "Y" else "**Não**"
        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 os índices de uma tabela.

    Parâmetros:
        schema — schema da tabela
        table  — nome da tabela
    """
    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"Nenhum índice encontrado em `{schema}.{table}`."

    lines = [f"**Índices em `{schema}.{table}`** ({len(rows)} encontrados)\n",
             "| Índice | Tipo | Único | Folhas | Níveis | Colunas |",
             "|--------|------|-------|--------|--------|---------|"]
    unique_map = {"U": "✅ Sim", "P": "✅ PK", "D": "Não"}
    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:
    """
    Mostra as primeiras N linhas de uma tabela.

    Parâmetros:
        schema — schema da tabela
        table  — nome da tabela
        limit  — número de linhas 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"A tabela `{schema}.{table}` está vazia ou não existe."

    headers = list(rows[0].keys())
    lines   = [f"**`{schema}.{table}`** — primeiras {len(rows)} linhas\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:
    """Visão geral de um schema: número de tabelas, linhas e espaço."""
    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}` não encontrado ou vazio."
    r = rows[0]
    return (
        f"Schema: {schema}\n"
        f"Tabelas: {r.get('NUM_TABLES','?')}\n"
        f"Total linhas: {r.get('TOTAL_ROWS','?')}\n"
        f"Total páginas: {r.get('TOTAL_PAGES','?')}\n"
    )


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

if __name__ == "__main__":
    # Testar a ligação antes de arrancar o servidor
    try:
        test = run_sql("SELECT 1 AS OK FROM SYSIBM.SYSDUMMY1")
        print(f"✅ Ligação ao Db2 OK ({DB2_HOST}:{DB2_PORT}/{DB2_DBNAME})")
    except Exception as e:
        print(f"❌ Falha na ligação ao Db2: {e}")
        raise

    mcp.run(transport="stdio")
EOF

Parte 3 — Ligar ao Claude Code

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

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

# Iniciar sessão
claude

Parte 4 — Resolução de problemas comuns

SQL1598N — db2connect license

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

Este erro significa que o servidor Db2 precisa de ser activado. No container Db2 Community Edition, activar:

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

Se não funcionar, o Db2 Community Edition não requer db2connect para ligações locais — mas para ligações TCP remotas pode ser necessário o ficheiro de licença. Alternativa: ligar pelo socket Unix em vez de TCP:

# .env — ligação via socket (dentro do mesmo host que o Docker)
DB2_HOST=localhost
DB2_PORT=50000

SQL0805N — CLI packages não feitos bind

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

ibm_db não instala (compilação falha)

# Instalar dependências de compilação primeiro
sudo apt-get install -y build-essential python3-dev
# Depois
uv pip install ibm-db --no-binary :all:

Testar a ligação 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 — Expandir o servidor

O padrão é simples — basta decorar com @mcp.tool():

@mcp.tool()
def search_columns(column_name: str) -> str:
    """
    Encontra todas as tabelas que têm uma coluna com este nome.

    Parâmetros:
        column_name — nome ou parte do nome da coluna
    """
    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"Nenhuma coluna com `{name}` no nome."

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

Resumo