OpenAI Operator for reading and writing COBOL Flat Files

OpenAI Operator: Reading and Writing COBOL Flat Files

This guide implements a full “Retrofit” architecture. We provide a Dockerized FastMCP server that parses legacy COBOL flat files (fixed-width) and a CrewAI agent configured to operate that server using the Model Context Protocol (MCP).

🏗️ Architecture

Server (server.py): A FastMCP server running inside Docker. It parses fixed-width text files using a “Copybook” schema.
Client (agent.py): A CrewAI agent that connects to the server via SSE using the mcps parameter to read/write data.

1. The Server (`server.py`)

This server exposes tools to read and append to a flat file (CLIENTS.DAT). It handles the translation between JSON (LLM) and Fixed-Width String (Mainframe).

import os
from fastmcp import FastMCP
from typing import List, Dict, Optional

# Initialize FastMCP
mcp = FastMCP("CobolFlatFileAdapter")

# --- CONFIGURATION: The "Copybook" Definition ---
# Schema: ID(10) | NAME(30) | BALANCE(12) | STATUS(1)
COPYBOOK_SCHEMA = [
    {"name": "CLIENT_ID",   "length": 10, "type": "str"},
    {"name": "FULL_NAME",   "length": 30, "type": "str"},
    {"name": "BALANCE",     "length": 12, "type": "money"},
    {"name": "STATUS",      "length": 1,  "type": "char"}
]

DATA_FILE = "data/CLIENTS.DAT"

def _ensure_data_file():
    """Helper to create a dummy mainframe file if it doesn't exist."""
    if not os.path.exists("data"):
        os.makedirs("data")

    if not os.path.exists(DATA_FILE):
        # Create dummy legacy data
        records = [
            "0000000001JOHN DOE                      00000500.00A",
            "0000000002JANE SMITH                    00012000.50A",
        ]
        with open(DATA_FILE, "w") as f:
            f.write("\n".join(records))

def _parse_line(line: str) -> Dict:
    """Parses a single fixed-width line into a JSON dict."""
    record = {}
    cursor = 0
    for field in COPYBOOK_SCHEMA:
        length = field["length"]
        chunk = line[cursor:cursor+length]
        value = chunk.strip()

        if field["type"] == "money":
            try:
                record[field["name"]] = float(value)
            except ValueError:
                record[field["name"]] = 0.0
        else:
            record[field["name"]] = value

        cursor += length
    return record

def _format_record(data: Dict) -> str:
    """Formats a JSON dict back into a fixed-width string."""
    line = ""
    for field in COPYBOOK_SCHEMA:
        length = field["length"]
        val = str(data.get(field["name"], ""))

        if field["type"] == "money":
            # Right align, zero pad
            val = val.rjust(length, "0")
        else:
            # Left align, space pad
            val = val.ljust(length, " ")

        line += val[:length] # Hard truncate

    return line

@mcp.tool()
def find_client_by_id(client_id: str) -> Optional[Dict]:
    """Searches for a specific client ID in the flat file."""
    _ensure_data_file()
    target_id = client_id.strip().zfill(10)

    try:
        with open(DATA_FILE, "r") as f:
            for line in f:
                if line.startswith(target_id):
                    return _parse_line(line.rstrip("\n"))
    except FileNotFoundError:
        return None
    return None

@mcp.tool()
def append_new_client(client_id: str, full_name: str, balance: str, status: str = "A") -> str:
    """
    Appends a new client record to the flat file.
    Args:
        client_id: Numeric ID (will be padded to 10 chars)
        full_name: Name (max 30 chars)
        balance: Amount (e.g. '100.50')
        status: 'A' for Active, 'X' for Closed
    """
    _ensure_data_file()

    new_record = {
        "CLIENT_ID": client_id,
        "FULL_NAME": full_name,
        "BALANCE": balance,
        "STATUS": status
    }

    formatted_line = _format_record(new_record)

    with open(DATA_FILE, "a") as f:
        f.write("\n" + formatted_line)

    return f"Success: Appended line '{formatted_line}'"

if __name__ == "__main__":
    # MANDATORY: Bind to 0.0.0.0 for Docker compatibility
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

2. The Dockerfile

This configuration ensures the server runs reliably in any container environment (Railway, Kubernetes, etc.) and exposes the necessary port.

FROM python:3.11-slim

WORKDIR /app

# Install FastMCP and Uvicorn
RUN pip install --no-cache-dir fastmcp uvicorn

# Copy the server code
COPY server.py .

# Create the data directory for persistence
RUN mkdir -p data

# MANDATORY: Expose port 8000 for external access
EXPOSE 8000

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

3. The Agent (`agent.py`)

This file demonstrates the Client Connectivity pattern using CrewAI. The agent acts as an “Operator” that can autonomously query and update the legacy file system.

Prerequisites:

pip install crewai
Docker container running on localhost:8000

import os
from crewai import Agent, Task, Crew

# 1. Configuration
# Ensure you have your OpenAI API Key set in your environment
# os.environ["OPENAI_API_KEY"] = "sk-..."

MCP_SERVER_URL = "http://localhost:8000/sse"

# 2. Define the Agent
# We use the 'mcps' parameter to connect the agent to our Dockerized FastMCP server.
cobol_operator = Agent(
    role="Legacy Systems Architect",
    goal="Manage client records in the legacy COBOL flat file system",
    backstory="""You are a specialized AI interface capable of reading and writing
    to 1990s-era mainframe datasets. You handle fixed-width data conversion effortlessly.""",

    # MANDATORY PATTERN: Connect to the MCP Server
    mcps=[MCP_SERVER_URL],

    verbose=True,
    allow_delegation=False
)

# 3. Define Tasks
# Task A: Read a record to verify it exists
task_audit_client = Task(
    description="Find the client with ID '2'. Report their Name and Balance.",
    expected_output="A summary of Client 2's details.",
    agent=cobol_operator
)

# Task B: Add a new record
task_onboard_client = Task(
    description="""
    Create a new client with the following details:
    - ID: 99
    - Name: 'AI RECRUIT'
    - Balance: 50.00
    - Status: A

    After creating, confirm the exact line that was written.
    """,
    expected_output="Confirmation of the new record creation.",
    agent=cobol_operator
)

# 4. Execute the Crew
crew = Crew(
    agents=[cobol_operator],
    tasks=[task_audit_client, task_onboard_client],
    verbose=True
)

if __name__ == "__main__":
    print("Starting Mainframe Operations...")
    result = crew.kickoff()
    print("\n\n########################")
    print("## OPERATION COMPLETE ##")
    print("########################\n")
    print(result)

🚀 Deployment Instructions

Build & Run Server:

docker build -t cobol-mcp .
docker run -p 8000:8000 -v $(pwd)/data:/app/data cobol-mcp

Run Agent:

export OPENAI_API_KEY=sk-your-key-here
python agent.py

The Agent will connect to http://localhost:8000/sse, discover the find_client_by_id and append_new_client tools, and execute the tasks autonomously.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.