Exposing IBM i (AS/400) DB2 Data via Model Context Protocol (MCP)

This guide provides a production-ready blueprint for creating a Model Context Protocol (MCP) server that acts as a gateway to IBM i (AS/400) DB2 databases.

By wrapping the legacy ibm_db driver in a FastMCP server, you enable modern AI agents (CrewAI, LangGraph, OpenAI Operator) to query legacy tables, inspect schemas, and retrieve business data using natural language, without needing to know specific DB2 SQL syntax or EBCDIC encoding rules.

The Architecture

We use the FastMCP framework to create a lightweight Python server. This server runs inside a Docker container, exposing an SSE (Server-Sent Events) endpoint that agents can connect to.

Protocol: MCP (Model Context Protocol)
Transport: SSE (HTTP) on Port 8000
Driver: ibm_db (Official IBM DB2 Python driver)
Infrastructure: Docker (Debian-based)

1. The Server Code (`server.py`)

This script defines the MCP server. It exposes two key tools to the AI agent:

list_tables: To explore the database schema.
query_db2: To execute SQL queries safely.

Security Warning: This example exposes a raw SQL execution tool. In production, you should replace query_db2 with specific, parameterized function calls (e.g., get_order_status(order_id)) to prevent unauthorized access.

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

# Initialize FastMCP
mcp = FastMCP("ibm-as400-db2-service")

# Configuration from Environment Variables
DB_NAME = os.getenv("DB2_NAME", "BLUDB")
DB_HOST = os.getenv("DB2_HOST", "192.168.1.100")
DB_PORT = os.getenv("DB2_PORT", "50000")
DB_USER = os.getenv("DB2_USER", "db2admin")
DB_PASS = os.getenv("DB2_PASS", "password")

# Connection String Construction
CONN_STR = (
    f"DATABASE={DB_NAME};"
    f"HOSTNAME={DB_HOST};"
    f"PORT={DB_PORT};"
    f"PROTOCOL=TCPIP;"
    f"UID={DB_USER};"
    f"PWD={DB_PASS};"
)

def get_connection():
    """Establishes a fresh connection to the DB2 database."""
    try:
        # ibm_db.connect returns a connection object or throws an exception
        conn = ibm_db.connect(CONN_STR, "", "")
        return conn
    except Exception as e:
        error_msg = ibm_db.conn_errormsg()
        raise ConnectionError(f"Failed to connect to DB2: {e}, Detail: {error_msg}")

@mcp.tool()
def list_tables(schema: str = "Administrator") -> List[str]:
    """
    Lists all tables in a specific DB2 schema.
    Useful for the agent to understand the database structure.
    """
    conn = get_connection()
    sql = f"SELECT TABNAME FROM SYSCAT.TABLES WHERE TABSCHEMA = '{schema}'"

    try:
        stmt = ibm_db.exec_immediate(conn, sql)
        tables = []
        dictionary = ibm_db.fetch_assoc(stmt)
        while dictionary:
            tables.append(dictionary["TABNAME"])
            dictionary = ibm_db.fetch_assoc(stmt)
        return tables
    except Exception as e:
        return [f"Error listing tables: {str(e)}"]
    finally:
        ibm_db.close(conn)

@mcp.tool()
def query_db2(sql_query: str) -> str:
    """
    Executes a read-only SQL query against the IBM i DB2 database.
    Use this to retrieve data. Enforce LIMIT on large queries.
    """
    # basic safety check for demo purposes
    if "DROP" in sql_query.upper() or "DELETE" in sql_query.upper():
        return "Error: Destructive queries are not allowed via this MCP."

    conn = get_connection()
    try:
        stmt = ibm_db.exec_immediate(conn, sql_query)
        results = []
        dictionary = ibm_db.fetch_assoc(stmt)

        # Fetch up to 100 rows to prevent context window overflow
        row_count = 0
        while dictionary and row_count < 100:
            results.append(dictionary)
            dictionary = ibm_db.fetch_assoc(stmt)
            row_count += 1

        return json.dumps(results, default=str)
    except Exception as e:
        db_err = ibm_db.stmt_errormsg()
        return f"SQL Error: {str(e)} - {db_err}"
    finally:
        ibm_db.close(conn)

if __name__ == "__main__":
    # HOST must be 0.0.0.0 to allow Docker networking
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

2. The Dockerfile

The ibm_db driver has system dependencies that are best handled in a Linux environment. We use a slim Python image and install necessary build tools.

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

# Install system dependencies required for ibm_db
# gcc and libxml2 are often required for building/linking the driver
RUN apt-get update && apt-get install -y \
    gcc \
    libxml2 \
    && rm -rf /var/lib/apt/lists/*

# Set the working directory
WORKDIR /app

# Copy requirement file first to leverage Docker cache
COPY requirements.txt .

# Install Python dependencies
RUN pip install --no-cache-dir -r requirements.txt

# Copy the server code
COPY server.py .

# Expose the port for the MCP server
EXPOSE 8000

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

`requirements.txt`

fastmcp
ibm_db
uvicorn

3. Running the Server

Build the Image:
Terminal window
```
docker build -t ibm-db2-mcp .
```

Run the Container: Replace the environment variables with your actual AS/400 credentials.

docker run -d \
  -p 8000:8000 \
  -e DB2_HOST="192.168.1.50" \
  -e DB2_USER="QSECOFR" \
  -e DB2_PASS="password123" \
  --name db2-mcp \
  ibm-db2-mcp

4. Connecting Your Agent

Once the Docker container is running, the MCP server is available at http://localhost:8000/sse. Below is how you connect a CrewAI agent to this legacy bridge.

CrewAI Configuration

from crewai import Agent, Task, Crew
# No special import needed for MCP if using the string syntax in `mcps`

# 1. Define the DB2 Specialist Agent
# The 'mcps' argument tells the agent where to find the tools.
db2_specialist = Agent(
    role='Legacy Data Specialist',
    goal='Retrieve inventory data from the AS/400 system',
    backstory='You are an expert in legacy DB2 databases. You query tables to find answers.',
    mcps=["http://localhost:8000/sse"],
    verbose=True
)

# 2. Define the Task
# The agent will automatically discover the 'list_tables' and 'query_db2' tools.
check_inventory_task = Task(
    description=(
        "Check the 'INVENTORY' table for item 'A100'. "
        "First, list the tables to confirm the schema name, "
        "then write a SQL query to get the stock count."
    ),
    agent=db2_specialist,
    expected_output="The current stock count for item A100."
)

# 3. Run the Crew
crew = Crew(
    agents=[db2_specialist],
    tasks=[check_inventory_task]
)

result = crew.kickoff()
print(result)

Troubleshooting Connection

“Connection refused”: Ensure your Docker container is running and mapped to port 8000 (docker ps).
“SQL30081N”: This is a common DB2 error indicating a communication failure. Check if the DB2_HOST IP is reachable from the Docker container and if the firewall on the IBM i side allows traffic on port 50000 (default DRDA port).
Authentication Errors: AS/400 passwords are often case-sensitive and limited to 10 characters on older systems. Ensure your environment variables match exactly.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.

Exposing IBM i (AS/400) DB2 Data via Model Context Protocol (MCP)

Exposing IBM i (AS/400) DB2 Data via Model Context Protocol (MCP)

The Architecture

1. The Server Code (server.py)

2. The Dockerfile

requirements.txt

3. Running the Server

4. Connecting Your Agent

CrewAI Configuration

Troubleshooting Connection

🛡️ Quality Assurance

1. The Server Code (`server.py`)

`requirements.txt`