OpenAI Operator for automated Mainframe CICS interactions

OpenAI Operator for Automated Mainframe CICS Interactions

Bridging the gap between cutting-edge OpenAI Operators and mission-critical IBM CICS (Customer Information Control System) environments requires a robust translation layer. Most CICS systems in banking and insurance still rely on TN3270 streams, which are not natively understood by modern LLMs.

This guide provides a production-ready Model Context Protocol (MCP) server built with FastMCP. It includes a runnable CrewAI Agent that uses the mcps configuration to connect to the server, allowing it to autonomously execute CICS transactions.

🏗️ Architecture

Server: A FastMCP wrapper around the s3270 terminal emulator.
Agent: A CrewAI Python script configured with the mcps parameter to consume the server’s tools via SSE.
Transport: Server-Sent Events (SSE) over HTTP.

🛠️ Implementation Guide

1. The Server (`server.py`)

This server exposes tools to connect to the mainframe, navigate screens, and extract text. Note: We rely on the py3270 library and the underlying s3270 system binary.

import os
from fastmcp import FastMCP
from py3270 import Emulator

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

# Global emulator instance (simulating a single session for this demo)
em = Emulator(visible=False, timeout=30)

@mcp.tool()
def connect_mainframe(host: str, port: int = 23) -> str:
    """
    Connects to the IBM Mainframe via TN3270.
    Args:
        host: The mainframe IP or hostname.
        port: The TN3270 port (default 23).
    """
    # Ensure your container has network access (e.g. via NordLayer)
    try:
        if not em.is_connected():
            em.connect(f"{host}:{port}")
        return f"Connected to {host}:{port}. Screen:\n{em.string_get()}"
    except Exception as e:
        return f"Connection failed: {str(e)}"

@mcp.tool()
def execute_command(command: str) -> str:
    """
    Executes a CICS transaction code (e.g. CESN) or types text.
    """
    if not em.is_connected():
        return "Error: Not connected. Call connect_mainframe first."

    try:
        em.send_string(command)
        em.send_enter()
        em.wait_for_field()
        return f"Command sent. New Screen:\n{em.string_get()}"
    except Exception as e:
        return f"Execution error: {str(e)}"

@mcp.tool()
def read_screen() -> str:
    """
    Reads the full text content of the current terminal screen.
    """
    if not em.is_connected():
        return "Error: Not connected."
    return em.string_get()

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

2. The Dockerfile

The s3270 binary must be installed at the OS level for py3270 to work.

FROM python:3.11-slim

# Install system dependencies including s3270
RUN apt-get update && apt-get install -y \
    s3270 \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app

# Install Python libraries
RUN pip install --no-cache-dir fastmcp==0.1.1 py3270==0.6.0

# Copy server code
COPY server.py .

# Expose port 8000 for Railway/Cloud compatibility
EXPOSE 8000

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

3. The Agent (`agent.py`)

This script demonstrates how to configure a CrewAI agent to consume the running MCP server using the mcps list.

Prerequisites:

Run the server: docker build -t cics-mcp . && docker run -p 8000:8000 cics-mcp
Install CrewAI: pip install crewai

from crewai import Agent, Task, Crew

# 1. Define the Agent with the MCP Server connection
# The 'mcps' argument tells CrewAI to discover tools from the SSE endpoint.
mainframe_operator = Agent(
    role='Mainframe CICS Operator',
    goal='Check system status on the legacy mainframe',
    backstory=(
        "You are a veteran systems operator. You know how to navigate "
        "IBM 3270 screens (""Green Screens""). You connect to mainframes, "
        "run transaction codes, and interpret the text output."
    ),
    # MANDATORY: Connect to the FastMCP server running on localhost
    mcps=["http://localhost:8000/sse"],
    verbose=True,
    allow_delegation=False
)

# 2. Define the Task
# Note: In a real scenario, you would replace 'mainframe.local' with a real IP.
check_system_task = Task(
    description=(
        "1. Connect to the mainframe at host 'mainframe.local' port 23. "
        "2. Once connected, read the screen to confirm we are at the login prompt. "
        "3. Run the transaction code 'CEMT' to check terminal status."
    ),
    expected_output="A summary of the screen content after running CEMT.",
    agent=mainframe_operator
)

# 3. Create the Crew
crew = Crew(
    agents=[mainframe_operator],
    tasks=[check_system_task],
    verbose=True
)

# 4. Execute
if __name__ == "__main__":
    print("## Starting Mainframe Automation Crew ##")
    try:
        result = crew.kickoff()
        print("\n\n########################")
        print("## AUTOMATION RESULT ##")
        print("########################\n")
        print(result)
    except Exception as e:
        print(f"Error running crew: {e}")
        print("Ensure the Docker container is running on port 8000!")

🚀 Deployment Instructions

Build and Run Server:

docker build -t cics-mcp .
docker run -d -p 8000:8000 cics-mcp

Run Agent:
Terminal window
```
python agent.py
```
Troubleshooting:
- Network: If the agent.py fails to connect, ensure host='0.0.0.0' is set in server.py.
- Mainframe Access: If the connect_mainframe tool times out, verify your Docker container has VPN access (e.g., via NordLayer) to the legacy network.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.