AI Agent-driven data retrieval from Mainframe CICS screens

Deployment Tips

• Infrastructure: Deploy this MCP server on Railway. Read our Step-by-Step Railway Deployment Guide.
• Security: This legacy system requires a VPN. Use NordLayer to bridge your cloud container to the internal network.
• Observability: Wrap this server with Helicone for production tracing.

AI Agent-driven data retrieval from Mainframe CICS screens

Older “Green Screen” applications (CICS) running on IBM Mainframes are notoriously difficult for modern AI Agents to access. They don’t speak JSON, REST, or SQL. They speak TN3270—a distinct telnet-like protocol that renders text at specific coordinates (rows and columns).

This guide provides a Model Context Protocol (MCP) server that acts as a “Screen Scraper Bridge.” It allows an AI Agent (like CrewAI, LangGraph, or OpenAI Operator) to “view” the mainframe screen, “type” commands, and extract data, treating the legacy terminal as if it were a web browser.

🏗️ Architecture

We use FastMCP to create the API server and py3270 (a wrapper for the s3270 emulator) to handle the low-level mainframe communication.

The Stack

• Server: FastMCP (Python)
• Protocol: SSE (Server-Sent Events) on Port 8000
• Driver: s3270 (Linux-based IBM 3270 emulator)
• Library: py3270

🚀 Deployment

1. The Server Code (server.py)

This server exposes tools that allow the agent to connect, navigate, and read text from the screen.

Note on State: This example uses a single global emulator instance for simplicity. In a production environment with multiple concurrent agents, you would implement a Session Manager that maps a session_id to a specific Emulator instance.

import os
import logging
from fastmcp import FastMCP
from py3270 import Emulator

# Initialize FastMCP
mcp = FastMCP("CICS-Bridge")

# Configure Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Global Emulator Instance
# In production, use a dictionary {session_id: emulator_instance}
_em = None

def get_emulator():
    global _em
    if _em is None:
        # s3270 must be installed in the Docker container
        if not os.path.exists("/usr/bin/s3270") and not os.path.exists("/usr/local/bin/s3270"):
             logger.warning("s3270 binary not found. Ensure it is installed via apt-get.")
        _em = Emulator(visible=False, timeout=30)
    return _em

@mcp.tool()
def connect_mainframe(host: str, port: int = 23) -> str:
    """
    Connects to a Mainframe TN3270 host.
    Use this before sending any keys.
    """
    em = get_emulator()
    try:
        # Connect format is usually host:port
        target = f"{host}:{port}"
        em.connect(target)
        # Wait for the screen to settle
        em.wait_for_field()
        return f"Connected to {target}. Current Screen:\n\n" + "\n".join(em.string_get())
    except Exception as e:
        return f"Connection Failed: {str(e)}"

@mcp.tool()
def send_keys(keys: str, wait_for_settle: bool = True) -> str:
    """
    Sends keystrokes to the current screen.
    Special keys: 'Enter', 'PF1', 'PF3', 'Tab', 'Clear'.
    Example: 'username' then 'Tab' then 'password' then 'Enter'.
    """
    em = get_emulator()
    if not em.is_connected():
        return "Error: Not connected. Use connect_mainframe first."

    try:
        # Simple parser for common keys
        # In a real scenario, you might parse complex sequences
        if keys.lower() == 'enter':
            em.send_enter()
        elif keys.lower().startswith('pf'):
            em.send_pf(int(keys[2:]))
        elif keys.lower() == 'clear':
            em.send_clear()
        else:
            # Type text
            em.send_string(keys)

        if wait_for_settle:
            em.wait_for_field()

        return "Keys sent. New Screen:\n\n" + "\n".join(em.string_get())
    except Exception as e:
        return f"Error sending keys: {str(e)}"

@mcp.tool()
def read_screen() -> str:
    """
    Reads the full text content of the current green screen.
    Useful for validating where you are or reading data.
    """
    em = get_emulator()
    if not em.is_connected():
        return "Error: Not connected."

    # string_get returns a list of strings (lines)
    lines = em.string_get()
    return "\n".join(lines)

@mcp.tool()
def fill_field(y_row: int, x_col: int, text: str) -> str:
    """
    Types text at a specific coordinate (Row, Col).
    Coordinates are 0-indexed or 1-indexed depending on system; py3270 is 0-indexed.
    """
    em = get_emulator()
    try:
        em.fill_field(y_row, x_col, text, len(text))
        return f"Filled '{text}' at {y_row},{x_col}."
    except Exception as e:
        return f"Error filling field: {str(e)}"

@mcp.tool()
def disconnect_mainframe() -> str:
    """
    Terminates the TN3270 session.
    """
    em = get_emulator()
    if em.is_connected():
        em.terminate()
        return "Disconnected."
    return "Already disconnected."

if __name__ == "__main__":
    # Ensure your container has network access (e.g. via NordLayer)
    # Binding to 0.0.0.0 is MANDATORY for Docker
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

2. The Dockerfile

The critical part here is installing s3270. Without this system binary, the Python library py3270 will fail.

FROM python:3.11-slim

# Install system dependencies
# s3270 is the backend terminal emulator engine
RUN apt-get update && apt-get install -y \
    s3270 \
    && rm -rf /var/lib/apt/lists/*

WORKDIR /app

# Install Python dependencies
# fastmcp for the server, py3270 for the mainframe driver
RUN pip install --no-cache-dir fastmcp py3270 uvicorn

# Copy application code
COPY server.py .

# Expose port 8000 for Railway/Docker networking
EXPOSE 8000

# Start the MCP server
CMD ["python", "server.py"]

🔌 Connecting the Agent

CrewAI Example

We connect to the MCP server using the mcps parameter. This allows CrewAI to automatically discover and use the Mainframe tools (connect_mainframe, send_keys, etc.) defined in server.py.

from crewai import Agent, Task, Crew

# 1. Define the Agent with MCP connectivity
# The agent will automatically load tools from the SSE endpoint
cics_operator = Agent(
    role="CICS Mainframe Specialist",
    goal="Retrieve Account Status for ID 889900",
    backstory="You are an expert at navigating legacy IBM 3270 green screens.",
    # Connect directly to the MCP server
    mcps=["http://localhost:8000/sse"],
    verbose=True
)

# 2. Define the Task
# The Agent will sequence: Connect -> Send Keys -> Read Screen -> Disconnect
task = Task(
    description="""
    1. Connect to the mainframe at 192.168.1.50.
    2. Enter 'CICS' to enter the subsystem.
    3. Type 'ACCT' and hit Enter to go to the Account Menu.
    4. Type '889900' in the Account ID field (approx row 10, col 20) and hit Enter.
    5. Read the 'Status' field from the screen and report it.
    """,
    expected_output="The status of account 889900.",
    agent=cics_operator
)

# 3. Execute
crew = Crew(agents=[cics_operator], tasks=[task])
result = crew.kickoff()
print("Final Mainframe Data:", result)

Troubleshooting Legacy Connections

1. VPN Requirement: Mainframes are almost never on the public internet. Your Docker container must be inside the corporate VPN. We recommend running a “Sidecar” container (like NordLayer or Tailscale) in your Kubernetes/Docker setup to bridge the network.
2. Coordinates: Mainframe fields are coordinate-based (Row, Column). If the Agent struggles to find the field, use the read_screen tool to dump the text. LLMs are surprisingly good at inferring “Account ID is on the 3rd line down” from raw text dumps.
3. Keyboard Locks: If the screen freezes (X SYSTEM status), the emulator usually handles the reset, but sometimes you may need to add a send_keys('Reset') command if the app gets stuck.

---

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.