Semantic Kernel skills for Mainframe CICS automation

The Context: Retrofitting the Mainframe for AI

CICS (Customer Information Control System) is the transaction server that runs the world’s economy. It processes billions of transactions daily for banks, insurers, and logistics giants. However, modern AI agents running on Semantic Kernel (SK) do not speak CICS protocols (IPIC, ECI) natively.

This guide provides a “Retrofit Pattern” to bridge this gap. We will build a FastMCP server that acts as a Semantic Kernel “Skill” (or Plugin). This server translates the agent’s natural language intents into structured HTTP requests targeting IBM z/OS Connect or a CICS Web Support interface.

This architecture allows your AI agents to execute legacy transactions (e.g., GETCUST, UPDTBAL) safely without modifying the underlying COBOL application logic.

Architecture Blueprint

Semantic Kernel Agent: Determines it needs to run a CICS transaction.
MCP Protocol: The agent queries the MCP server for available tools.
FastMCP Server: Exposes functions like execute_cics_transaction and check_cics_region.
Legacy Bridge: The Python code uses standard HTTP methods to communicate with the CICS Gateway (z/OS Connect).

The Code: CICS Gateway MCP Server

This server provides two critical skills:

Region Health Check: Verifies if the CICS region is accepting transactions.
Transaction Execution: Sends a payload to a specific CICS program via the z/OS Connect REST wrapper.

server.py

import os
import requests
from fastmcp import FastMCP, Context

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

# Configuration (In production, load these from environment variables)
# Example z/OS Connect Endpoint: https://mainframe.internal:9443/zosConnect/services
ZOS_CONNECT_BASE_URL = os.getenv("ZOS_CONNECT_URL", "https://mainframe.example.com/zosConnect/services")
CICS_AUTH = (os.getenv("CICS_USER", "CICSUSER"), os.getenv("CICS_PASS", "PASSWORD"))

# Ensure your container has network access (e.g. via NordLayer)

@mcp.tool()
def check_cics_region_status(region_name: str) -> str:
    """
    Checks the health status of a specific CICS region via the z/OS Connect discovery endpoint.

    Args:
        region_name: The identifier of the CICS region (e.g., 'CICSPROD').
    """
    url = f"{ZOS_CONNECT_BASE_URL}/{region_name}/health"

    try:
        # Many internal mainframes use self-signed certs; verify=False is common in retrofit dev
        response = requests.get(url, auth=CICS_AUTH, timeout=5, verify=False)

        if response.status_code == 200:
            data = response.json()
            status = data.get("status", "UNKNOWN")
            return f"Region {region_name} is {status}. Load: {data.get('transactionLoad', 'N/A')}"
        elif response.status_code == 404:
            return f"Region {region_name} not found in gateway."
        else:
            return f"Error checking region: HTTP {response.status_code}"

    except requests.exceptions.RequestException as e:
        return f"Network failure connecting to CICS Gateway: {str(e)}"

@mcp.tool()
def execute_cics_transaction(transaction_id: str, payload: dict) -> str:
    """
    Executes a specific CICS transaction program via JSON payload.

    Args:
        transaction_id: The 4-character CICS transaction code (e.g., 'INQ1') or Service Name.
        payload: A dictionary containing the input fields required by the COBOL program.
    """
    url = f"{ZOS_CONNECT_BASE_URL}/{transaction_id}?action=execute"

    try:
        response = requests.post(
            url,
            json=payload,
            auth=CICS_AUTH,
            timeout=10,
            verify=False
        )

        if response.status_code == 200:
            result = response.json()
            # Return the raw JSON response from the mainframe for the Agent to parse
            return str(result)
        elif response.status_code == 401:
            return "Authentication Failed: Check RACF credentials."
        elif response.status_code == 503:
            return "CICS Region Unavailable or Transaction Disabled."
        else:
            return f"Transaction {transaction_id} failed with HTTP {response.status_code}: {response.text}"

    except requests.exceptions.RequestException as e:
        return f"Connectivity Error during transaction execution: {str(e)}"

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

Containerization: Dockerfile

To deploy this skill in a modern orchestrator (Kubernetes, ECS, or Railway), we wrap it in a lightweight container.

Dockerfile

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

# Set working directory
WORKDIR /app

# Install dependencies
# fastmcp: The MCP server framework
# requests: For HTTP communication with z/OS Connect
RUN pip install --no-cache-dir fastmcp requests

# Copy the server code
COPY server.py .

# EXPOSE 8000 for Railway compatibility
EXPOSE 8000

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

Integration Notes for DevOps Architects

Network Tunneling: Mainframes typically reside in isolated private subnets. If this container runs in the cloud (e.g., Railway, AWS), you must establish a secure tunnel (VPN or VPC Peering) back to the data center.
Certificate Handling: The code above uses verify=False to bypass SSL verification, which is common during initial “retrofit” prototyping against internal mainframes with self-signed CAs. In production, mount your corporate CA bundle into the container and point requests to it.
Semantic Kernel Connection: To use this with Semantic Kernel, configure your SK Python or .NET agent to connect to this MCP server via SSE (Server-Sent Events) or Stdio, depending on your hosting model.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.