Building an MCP Server for IBM Db2 12.1 in Docker

This tutorial uses the Python ibm_db driver to connect directly to Db2 — simple, no intermediaries, and compatible with 12.1.

Architecture

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

No intermediaries. ibm_db includes the IBM clidriver and connects directly to Db2 port 50000.

Prerequisites

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

Part 1 — Docker Environment

1.1 docker-compose.yml

Just one container — Db2. No database is created at startup; db2sampl in the next step creates and populates SAMPLE with the sample tables.

# 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 Start

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

# Wait for initialization (~2-3 min)
docker compose logs -f db2
# When "Setup has completed" appears — ready

1.3 Populate the database

Create the SAMPLE database with all sample tables (EMPLOYEE, DEPARTMENT, etc.):

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

Part 2 — The MCP Server

2.1 Create the project

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

2.2 Create pyproject.toml

ibm_db includes the IBM clidriver — no separate Db2 Client installation needed.

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 Install

ibm_db downloads and installs the IBM clidriver automatically during pip install. There is no need to install the Db2 Client separately.

uv venv
uv pip install -e .

# Verify that the driver installed correctly
uv run python -c "import ibm_db; print('ibm_db OK')"

If you are on a Linux x86_64 system and specifically need the clidriver 12.1 (optional — 11.5 connects fine to a 12.1 server):

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

2.4 Create the credentials file

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

2.5 Create the server — server.py

cat > server.py << 'EOF'
"""
MCP Server — Db2 Schema Explorer
Explores tables, columns, and indexes of a Db2 12.1 database
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()

# MCP stdio transport uses stdout exclusively for JSON-RPC.
# The IBM CLIdriver may emit diagnostic messages to stderr
# that appear in the terminal. Redirecting to a file avoids noise.
_log = open(os.path.expanduser("~/.db2-mcp-server.log"), "a", buffering=1)
sys.stderr = _log

# ── Configuration ─────────────────────────────────────────────────────────────

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 in ibm_db DSN format
# DIAGLEVEL=0 suppresses CLIdriver diagnostic messages in the 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;"
)

# ── Minimalist connection pool (thread-safe) ──────────────────────────────────
# ibm_db is not thread-safe per connection — we use a lock to serialize.
# For production consider a more robust pool.

_conn_lock = threading.Lock()
_conn      = None


def _get_conn():
    """Returns the active connection, reconnecting if necessary."""
    global _conn
    try:
        # Check if the connection is still alive
        if _conn is not None:
            ibm_db.active(_conn)  # raises exception if dead
            return _conn
    except Exception:
        _conn = None

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


def run_sql(sql: str) -> list[dict]:
    """
    Executes a SELECT and returns results as a list of dictionaries.
    Thread-safe via global lock.
    """
    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


# ── MCP Server ────────────────────────────────────────────────────────────────

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


@mcp.tool()
def list_schemas() -> str:
    """Lists all schemas in the database (excluding system schemas)."""
    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 "No user schemas found."

    lines = [f"**Available schemas** ({len(rows)} found)\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:
    """
    Lists all tables in a schema.

    Parameters:
        schema — schema name (e.g. 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"No tables found in schema `{schema}`."

    lines = [f"**Tables in `{schema}`** ({len(rows)} found)\n",
             "| Table | Rows | Pages | Last used |",
             "|-------|------|-------|-----------|"]
    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:
    """
    Describes the structure of a table: columns, types, nullable, and defaults.

    Parameters:
        schema — table schema
        table  — table name
    """
    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"Table `{schema}.{table}` not found or has no columns."

    lines = [f"**`{schema}.{table}`** — {len(rows)} columns\n",
             "| # | Column | Type | 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_  = "Yes" 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:
    """
    Lists the indexes of a table.

    Parameters:
        schema — table schema
        table  — table name
    """
    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"No indexes found on `{schema}.{table}`."

    lines = [f"**Indexes on `{schema}.{table}`** ({len(rows)} found)\n",
             "| Index | Type | Unique | Leaf pages | Levels | Columns |",
             "|-------|------|--------|------------|--------|---------|"]
    unique_map = {"U": "✅ Yes", "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:
    """
    Shows the first N rows of a table.

    Parameters:
        schema — table schema
        table  — table name
        limit  — number of rows to show (maximum 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"Table `{schema}.{table}` is empty or does not exist."

    headers = list(rows[0].keys())
    lines   = [f"**`{schema}.{table}`** — first {len(rows)} rows\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:
    """Overview of a schema: number of tables, rows, and space."""
    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}` not found or empty."
    r = rows[0]
    return (
        f"Schema: {schema}\n"
        f"Tables: {r.get('NUM_TABLES','?')}\n"
        f"Total rows: {r.get('TOTAL_ROWS','?')}\n"
        f"Total pages: {r.get('TOTAL_PAGES','?')}\n"
    )


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

if __name__ == "__main__":
    # Test the connection before starting the server
    try:
        test = run_sql("SELECT 1 AS OK FROM SYSIBM.SYSDUMMY1")
        print(f"✅ Db2 connection OK ({DB2_HOST}:{DB2_PORT}/{DB2_DBNAME})")
    except Exception as e:
        print(f"❌ Db2 connection failed: {e}")
        raise

    mcp.run(transport="stdio")

Part 3 — Connect to Claude Code

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

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

# Start session
claude

Part 4 — Troubleshooting common issues

SQL1598N — db2connect license

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

This error means the Db2 server needs to be activated. In the Db2 Community Edition container, activate with:

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

If that does not work, Db2 Community Edition does not require db2connect for local connections — but for remote TCP connections a license file may be needed. Alternative: connect via Unix socket instead of TCP:

# .env — connection via socket (within the same host as Docker)
DB2_HOST=localhost
DB2_PORT=50000

SQL0805N — CLI packages not bound

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

ibm_db does not install (compilation fails)

# Install build dependencies first
sudo apt-get install -y build-essential python3-dev
# Then
uv pip install ibm-db --no-binary :all:

Test the connection directly

# 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}

Part 5 — Expanding the server

The pattern is simple — just decorate with @mcp.tool():

@mcp.tool()
def search_columns(column_name: str) -> str:
    """
    Finds all tables that have a column with this name.

    Parameters:
        column_name — column name or partial name
    """
    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"No columns with `{name}` in the name."

    lines = [f"**Columns with `{name}`** ({len(rows)} found)\n",
             "| Schema | Table | Column | Type |",
             "|--------|-------|--------|------|"]
    for r in rows:
        lines.append(
            f"| `{r['TABSCHEMA']}` | `{r['TABNAME']}` "
            f"| `{r['COLNAME']}` | {r['TYPENAME']} |"
        )
    return "\n".join(lines)

Summary