AutoGen for processing COBOL Flat Files in multi-agent workflows

Deployment Tips

• Infrastructure: Deploy this MCP server on Railway. Read our Step-by-Step Railway Deployment Guide.

AutoGen for processing COBOL Flat Files in multi-agent workflows

This guide provides a “Retrofit Pattern” to bridge modern AutoGen agents with legacy fixed-width COBOL data files using the Model Context Protocol (MCP).

We will build:

1. FastMCP Server: Acts as the “Driver” that knows how to parse bytes based on a Copybook schema.
2. AutoGen Client: A workflow where agents can “plug in” to this server using a configuration list (mcps), allowing them to query legacy data without understanding file offsets.
3. Docker Environment: A production-ready container for the server.

---

1. The Server (server.py)

The server uses fastmcp to expose a specific tool read_flat_file. It handles the low-level string slicing so the LLM doesn’t have to hallucinate character counts.

import os
from fastmcp import FastMCP

# Initialize FastMCP server
mcp = FastMCP("cobol-flat-file-server")

# Sample Data Generation (Simulating a mounted legacy volume)
SAMPLE_FILE_PATH = "legacy_payroll.dat"
SAMPLE_DATA = [
    "001JOHN DOE  050000HR      ",  # ID(3), Name(10), Salary(6), Dept(8)
    "002JANE SMITH075000IT      ",
    "003BOB JONES 042000SALES   ",
    "004ALICE WONG088000IT      "
]

def ensure_sample_file():
    """Creates a dummy flat file if it doesn't exist."""
    if not os.path.exists(SAMPLE_FILE_PATH):
        with open(SAMPLE_FILE_PATH, "w") as f:
            f.write("\n".join(SAMPLE_DATA))

@mcp.tool()
def read_flat_file(filepath: str, schema: str) -> str:
    """
    Reads a fixed-width COBOL flat file and converts it to JSON based on a schema.

    Args:
        filepath: Path to the .dat file (e.g., 'legacy_payroll.dat').
        schema: A string representing field definitions in format 'NAME:LENGTH,NAME:LENGTH'.
                Example: 'ID:3,NAME:10,SALARY:6,DEPT:8'
    """
    ensure_sample_file()

    if not os.path.exists(filepath):
        return f"Error: File {filepath} not found."

    try:
        # Parse the schema string into a list of (name, length) tuples
        fields = []
        for item in schema.split(','):
            parts = item.split(':')
            if len(parts) != 2:
                continue
            fields.append((parts[0].strip(), int(parts[1].strip())))

        results = []
        with open(filepath, 'r') as f:
            for line in f:
                if not line.strip():
                    continue

                record = {}
                current_pos = 0

                # Slice the line based on fixed widths
                for name, length in fields:
                    val = line[current_pos : current_pos + length].strip()
                    record[name] = val
                    current_pos += length

                results.append(record)

        return str(results)

    except Exception as e:
        return f"Error parsing file: {str(e)}"

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. Dockerfile

We expose port 8000 and use a slim Python image.

# Use an official Python runtime
FROM python:3.10-slim

WORKDIR /app

# Install dependencies
RUN pip install --no-cache-dir "mcp[cli]" fastmcp uvicorn

# Copy server code
COPY server.py .

# Expose the SSE port
EXPOSE 8000

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

---

3. The AutoGen Client (client.py)

This client demonstrates the “Retrofit” pattern: we define a list of MCP servers (mcps) and programmatically register their tools to the AutoGen agents. This decouples the agent logic from the tool implementation.

import asyncio
import os
from autogen import AssistantAgent, UserProxyAgent
from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client

# --- Configuration ---
# List of MCP Servers to connect to (Retrofit Pattern)
mcps = ["http://localhost:8000/sse"]

config_list = [{"model": "gpt-4o", "api_key": os.environ.get("OPENAI_API_KEY")}]

async def run_agent_workflow():
    print(f"Connecting to MCP Servers: {mcps}...")

    # We use a context manager to handle the connection lifecycle
    # In a real app, you might iterate over 'mcps' to connect to multiple servers.
    # Here we demonstrate connecting to the first defined server.
    server_url = mcps[0]

    async with sse_client(server_url) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Dynamically discover tools
            tools = await session.list_tools()
            print(f"Connected. Found {len(tools.tools)} tools.")

            # --- Wrapper Logic ---
            # AutoGen requires a callable function to be registered.
            # We create a wrapper that delegates to the MCP session.

            async def read_flat_file(filepath: str, schema: str) -> str:
                result = await session.call_tool(
                    "read_flat_file",
                    arguments={"filepath": filepath, "schema": schema}
                )
                return result.content[0].text

            # --- Agent Setup ---

            assistant = AssistantAgent(
                name="Legacy_Analyst",
                llm_config={
                    "config_list": config_list,
                    "temperature": 0,
                },
                system_message="""You are a Legacy Data Analyst.
                You have access to a tool 'read_flat_file' to read COBOL data.

                The file 'legacy_payroll.dat' has this schema:
                - ID: 3 chars
                - NAME: 10 chars
                - SALARY: 6 chars
                - DEPT: 8 chars

                Construct the schema string 'ID:3,NAME:10,SALARY:6,DEPT:8' and use it to answer questions.
                """
            )

            user_proxy = UserProxyAgent(
                name="User_Proxy",
                human_input_mode="NEVER",
                max_consecutive_auto_reply=5,
                code_execution_config={"work_dir": "coding", "use_docker": False},
            )

            # Register the tool
            # Assistant knows the tool exists
            assistant.register_for_llm(name="read_flat_file", description="Reads fixed-width COBOL files")(read_flat_file)

            # User Proxy executes the tool
            user_proxy.register_for_execution(name="read_flat_file")(read_flat_file)

            # --- Execution ---

            print(">>> Starting Chat...")
            await user_proxy.a_initiate_chat(
                assistant,
                message="Please calculate the total salary for the 'IT' department in legacy_payroll.dat."
            )

if __name__ == "__main__":
    asyncio.run(run_agent_workflow())

How to Run

1. Start the Docker Container:

   docker build -t cobol-mcp .
   docker run -p 8000:8000 cobol-mcp

2. Run the Client:

   export OPENAI_API_KEY=sk-...
   python client.py

Expected Output

The User_Proxy will execute the read_flat_file tool. The MCP server inside Docker will parse the fixed-width file and return JSON. The Legacy_Analyst will then sum the salaries for the ‘IT’ department (75000 + 88000) and respond with 163000.

---

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.