How CrewAI agents can interact with Mainframe CICS applications

Slug: crewai-mainframe-cics-integration

The Knowledge Gap

IBM CICS (Customer Information Control System) handles billions of transactions daily, forming the backbone of banking and insurance infrastructure. However, for a modern AI agent like CrewAI, a CICS mainframe is an alien world.

Traditionally, interacting with CICS meant:

3270 Emulation: “Screen scraping” green-screen terminals (fragile and slow).
EBCDIC Encoding: Dealing with non-ASCII character sets.
Binary COBOL Structures: Parsing COMP-3 packed decimals.

The Retrofit Solution

Modernized mainframes expose CICS programs via CICS Web Support (CWS) or IBM z/OS Connect. These layers wrap legacy COBOL programs (COMMARREA or Channels) into standard REST/JSON endpoints.

This guide provides the “glue code” to connect CrewAI to these HTTP-enabled CICS endpoints using the Model Context Protocol (MCP). This allows your agent to trigger transactions (e.g., “Get Account Balance”, “Update Policy”) as if they were standard API calls.

🛠️ The FastMCP Server

We will build an MCP server that acts as a CICS Gateway. It translates the agent’s natural language intent into structured HTTP requests compatible with z/OS Connect or CICS Web Services.

Prerequisites

Python 3.11+
Network Access: The container running this code must have VPN/Tunnel access to the Mainframe’s internal IP.
Service Credentials: A RACF ID and Password for authentication.

`server.py`

from fastmcp import FastMCP
import requests
import os
import json
from typing import Dict, Any

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

# Configuration (In production, load these from environment variables)
CICS_BASE_URL = os.getenv("CICS_BASE_URL", "https://mainframe.internal.corp:9443/cics/api")
# NOTE: Verify SSL certificates in production. False is used here for internal/self-signed legacy certs.
VERIFY_SSL = os.getenv("VERIFY_SSL", "False").lower() == "true"

@mcp.tool()
def execute_cics_transaction(transaction_id: str, program_name: str, payload: Dict[str, Any]) -> str:
    """
    Executes a CICS transaction via z/OS Connect or CICS Web Support.

    Args:
        transaction_id: The 4-character CICS Transaction ID (TRANSID), e.g., 'ACCT'.
        program_name: The name of the target COBOL program exposed via the API.
        payload: A dictionary containing the input fields expected by the COBOL program.
                 These will be mapped to the COMMAREA or Channel by the z/OS Connect layer.

    Returns:
        A JSON string containing the response from the mainframe or error details.
    """

    # Ensure your container has network access (e.g. via NordLayer)
    # The Mainframe is likely behind a corporate firewall.

    endpoint = f"{CICS_BASE_URL}/{program_name}"

    headers = {
        "Content-Type": "application/json",
        "X-CICS-TransactionID": transaction_id,  # Custom header often used for tracking
        "Accept": "application/json"
    }

    # Basic Auth is common for Mainframe HTTP services (RACF ID/Password)
    auth = (os.getenv("CICS_USER"), os.getenv("CICS_PASSWORD"))

    try:
        response = requests.post(
            endpoint,
            json=payload,
            headers=headers,
            auth=auth,
            verify=VERIFY_SSL,
            timeout=30 # Mainframes can be fast, but network hops add latency
        )

        response.raise_for_status()

        # Return the parsed JSON response from the mainframe
        return json.dumps({
            "status": "success",
            "cics_response_code": response.status_code,
            "data": response.json()
        })

    except requests.exceptions.HTTPError as e:
        # Handle specific Mainframe HTTP return codes (e.g., 500 for ABEND)
        error_msg = f"CICS HTTP Error: {e.response.status_code} - {e.response.text}"
        return json.dumps({"status": "error", "message": error_msg})

    except requests.exceptions.ConnectionError:
        return json.dumps({
            "status": "error",
            "message": "Failed to connect to Mainframe. Check VPN/Tunnel configuration."
        })

    except Exception as e:
        return json.dumps({"status": "error", "message": str(e)})

@mcp.tool()
def check_cics_region_status() -> str:
    """
    Checks if the CICS Region is reachable and accepting web requests.
    """
    try:
        # Often a generic 'ping' or 'version' endpoint exists
        response = requests.get(
            f"{CICS_BASE_URL}/health",
            timeout=5,
            verify=VERIFY_SSL
        )
        if response.status_code == 200:
            return "CICS Region Online: Web Interface Active"
        else:
            return f"CICS Region Warning: Responded with {response.status_code}"
    except Exception as e:
        return f"CICS Region Offline: {str(e)}"

if __name__ == "__main__":
    mcp.run()

🐳 Dockerfile

To deploy this in a modern cloud environment (like Railway, AWS ECS, or Kubernetes) while talking to a legacy on-premise system, you need a clean container.

Important: This Dockerfile exposes port 8000, which is required for the MCP server to communicate with the AgentRetrofit platform or your local CrewAI instance.

# Use a lightweight Python base image
FROM python:3.11-slim

# Set working directory
WORKDIR /app

# Install dependencies
# fastmcp: The library to run the MCP server
# requests: Standard library for HTTP calls
RUN pip install --no-cache-dir fastmcp requests

# Copy the server code
COPY server.py .

# Environment variables (Defaults - Override these in deployment!)
ENV CICS_BASE_URL="https://mainframe.internal.corp:9443/cics/api"
ENV CICS_USER="USERID"
ENV CICS_PASSWORD="PASSWORD"
ENV VERIFY_SSL="False"

# Expose the port for the MCP server
EXPOSE 8000

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

🔌 Connecting to CrewAI

Once your Docker container is running (and has VPN access to the mainframe), you can connect it to your CrewAI agent.

`agents.py` Configuration

from crewai import Agent, Task, Crew
# Assuming you are using a generic MCP client tool wrapper
from mcp_client import RemoteMCPTool

# Initialize the tool pointing to your Docker container
cics_tool = RemoteMCPTool(
    url="http://localhost:8000", # Or your deployed URL
    tool_name="execute_cics_transaction"
)

mainframe_agent = Agent(
    role='Mainframe Integration Specialist',
    goal='Retrieve and update customer records in the legacy CICS system',
    backstory="""You are a veteran systems architect capable of bridging the gap
    between modern AI and 30-year-old COBOL applications. You understand that
    transactions must be precise.""",
    tools=[cics_tool],
    verbose=True
)

cics_task = Task(
    description="""
    Look up customer ID '88421' using the 'CUSTINQ' program.
    If the account status is 'Active', update the last_contact_date to today
    using the 'CUSTUPD' program.
    """,
    agent=mainframe_agent
)

crew = Crew(agents=[mainframe_agent], tasks=[cics_task])
result = crew.kickoff()

Troubleshooting Common Legacy Errors

HTTP 500 (ABEND): This often means the COBOL program crashed (“Abnormal End”). Check if your payload matches the exact field lengths expected by the COBOL LINKAGE SECTION.
Connection Refused: Ensure your Docker container is running inside a VPC or has a VPN sidecar (like NordLayer or Tailscale) allowing access to the Mainframe’s private IP.
RACF Authentication Failed: Mainframe passwords expire frequently. Ensure your service account is active.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.

How CrewAI agents can interact with Mainframe CICS applications

How CrewAI agents can interact with Mainframe CICS applications

The Knowledge Gap

The Retrofit Solution

🛠️ The FastMCP Server

Prerequisites

server.py

🐳 Dockerfile

🔌 Connecting to CrewAI

agents.py Configuration

Troubleshooting Common Legacy Errors

🛡️ Quality Assurance

`server.py`

`agents.py` Configuration