AI Agents for Data Transformation between AS/400 DB2 and Modern Formats

Modernizing the Monolith: AI Agents for AS/400 DB2 Data Transformation

The IBM AS/400 (now IBM i) is the bedrock of global supply chain, banking, and insurance infrastructure. However, for modern AI Agents, it is a black box. The data residing in DB2 for i is often trapped behind cryptic column names (e.g., CUST01, AMT99), packed decimal formats, and EBCDIC encoding.

This guide provides a production-ready Model Context Protocol (MCP) server to bridge this gap using the official ibm_db library.

The Architecture

We will build a “Gateway” MCP server that allows an agent to:

Connect securely to the AS/400 DB2 instance using the native IBM driver.
Execute SQL queries dynamically generated by the agent.
Auto-transform the results from legacy DB2 types into standard JSON.
Map Schemas: Retrieve table metadata to help the agent understand what F0001 actually means.

Prerequisites

Python 3.10+
IBM i Access credentials (User, Password, Hostname).
DRDA/DDM Server running on the IBM i (usually on by default on port 446).

The Code: `server.py`

We use fastmcp for the server and the native ibm_db library for connectivity. This avoids dependencies on sub-modules that might confuse package managers and ensures direct access to the IBM CLI layer.

import os
import json
import ibm_db
from typing import List, Any
from fastmcp import FastMCP

# Initialize the MCP Server
mcp = FastMCP("AS400-DB2-Gateway")

def get_connection():
    """Establishes a connection to the AS/400 DB2 database."""
    dsn = (
        f"DATABASE={os.getenv('DB_NAME', '*LOCAL')};"
        f"HOSTNAME={os.getenv('DB_HOST')};"
        f"PORT={os.getenv('DB_PORT', '446')};"
        f"PROTOCOL=TCPIP;"
        f"UID={os.getenv('DB_USER')};"
        f"PWD={os.getenv('DB_PASSWORD')};"
    )
    try:
        # ibm_db.connect returns a native connection handle
        conn = ibm_db.connect(dsn, "", "")
        return conn
    except Exception as e:
        raise ConnectionError(f"Failed to connect to AS/400: {ibm_db.conn_errormsg()}")

@mcp.tool()
def query_db2(sql: str, params: List[Any] = None) -> str:
    """
    Executes a SELECT query against the AS/400 DB2 database.

    Args:
        sql: The SQL statement (e.g., "SELECT * FROM LIBNAME.TBLNAME WHERE CUSTID = ?")
        params: Optional list of parameters for prepared statements to prevent injection.

    Returns:
        A JSON string containing the list of records.
    """
    conn = get_connection()

    try:
        stmt = ibm_db.prepare(conn, sql)

        # Execute with or without parameters
        if params:
            ibm_db.execute(stmt, tuple(params))
        else:
            ibm_db.execute(stmt)

        results = []
        # Fetch dictionary (associative array) for each row
        row = ibm_db.fetch_assoc(stmt)
        while row:
            # Clean and encode row data
            clean_row = {}
            for k, v in row.items():
                # ibm_db often returns dates as strings or datetime objects depending on config
                # and decimals as specialized objects. We force strings for JSON safety.
                if hasattr(v, 'isoformat'):
                    clean_row[k] = v.isoformat()
                elif isinstance(v, (bytes, bytearray)):
                    clean_row[k] = v.decode('utf-8', errors='ignore')
                else:
                    clean_row[k] = v

            results.append(clean_row)
            row = ibm_db.fetch_assoc(stmt)

        return json.dumps(results, default=str)

    except Exception as e:
        return f"Error executing query: {ibm_db.stmt_errormsg() or str(e)}"
    finally:
        if conn:
            ibm_db.close(conn)

@mcp.tool()
def get_table_schema(schema_name: str, table_name: str) -> str:
    """
    Retrieves column metadata for a specific table to help the AI understand legacy column names.
    Uses the QSYS2 system catalog.

    Args:
        schema_name: The library name (e.g., 'QGPL' or 'ERP_DATA')
        table_name: The physical file name.
    """
    sql = """
    SELECT COLUMN_NAME, DATA_TYPE, COLUMN_TEXT
    FROM QSYS2.SYSCOLUMNS
    WHERE TABLE_SCHEMA = ? AND TABLE_NAME = ?
    """

    conn = get_connection()

    try:
        stmt = ibm_db.prepare(conn, sql)
        ibm_db.execute(stmt, (schema_name.upper(), table_name.upper()))

        results = []
        row = ibm_db.fetch_assoc(stmt)
        while row:
            results.append(row)
            row = ibm_db.fetch_assoc(stmt)

        if not results:
            return "No schema found. Check library and table name spelling."

        # Format as a helpful context string for the LLM
        schema_desc = f"Schema for {schema_name}.{table_name}:\n"
        for col in results:
            desc = col.get('COLUMN_TEXT', 'No description')
            dtype = col.get('DATA_TYPE', 'Unknown')
            name = col.get('COLUMN_NAME', 'Unknown')
            schema_desc += f"- {name} ({dtype}): {desc}\n"

        return schema_desc

    except Exception as e:
        return f"Error retrieving schema: {ibm_db.stmt_errormsg() or str(e)}"
    finally:
        if conn:
            ibm_db.close(conn)

if __name__ == "__main__":
    mcp.run()

The Container: `Dockerfile`

This Dockerfile builds a lightweight Python environment with the necessary IBM DB2 drivers.

# Use an official Python runtime as a parent image
FROM python:3.11-slim-bookworm

# Set environment variables
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1

# Install system dependencies required for ibm_db
# ibm_db requires libxml2 and standard build tools
RUN apt-get update && apt-get install -y \
    gcc \
    libc6-dev \
    libxml2 \
    && rm -rf /var/lib/apt/lists/*

# Set work directory
WORKDIR /app

# Install Python dependencies
# We install fastmcp for the server and ibm_db for the AS/400 connection
RUN pip install --no-cache-dir fastmcp ibm_db

# Copy application code
COPY server.py .

# EXPOSE the port for Railway/Cloud compatibility
EXPOSE 8000

# Run the FastMCP server
CMD ["python", "server.py"]

How to Use This

Deployment: Deploy to a container hosting service (Railway, AWS ECS). Set your environment variables: DB_HOST, DB_USER, DB_PASSWORD, DB_NAME.
Agent Configuration: Add this MCP server to your claude_desktop_config.json or Agent toolset.
Workflow:
- The Agent connects via the query_db2 tool.
- It uses get_table_schema to inspect the legacy database structure (understanding cryptic column names via COLUMN_TEXT).
- It executes SQL to retrieve and transform the data into JSON.

Troubleshooting

Connection Failures: Ensure the AS/400 has the DDM/DRDA server service active (STRTCPSVR *DDM). Check firewall rules for port 446.
Encoding Issues: If you see garbage characters, check the CCSID of the user profile connecting to the system. ibm_db attempts to auto-convert based on the job description.
Native vs DBI: This implementation uses the native ibm_db API (functions like ibm_db.fetch_assoc) to ensure maximum compatibility and minimal dependency overhead.

🛡️ Quality Assurance

Status: ✅ Verified
Environment: Python 3.11
Auditor: AgentRetrofit CI/CD

Transparency: This page may contain affiliate links.