Semantic Kernel skills for Mainframe CICS automation

The Bridge: Connecting AI to the Green Screen

Enterprise organizations often run their core transaction processing on IBM Mainframes using CICS (Customer Information Control System). While robust, these systems are “black boxes” to modern AI agents. To enable Microsoft Semantic Kernel to automate CICS workflows—such as checking inventory, processing refunds, or updating customer records—we need a translation layer.

This guide provides a FastMCP server that acts as a 3270 terminal emulator. It wraps the screen-scraping logic into clean “Skills” (Plugins) that Semantic Kernel can invoke natively.

🏗️ Architecture

FastMCP Server (server.py): Hosted in Docker. It uses py3270 to create a virtual terminal session, sending keystrokes and reading screen text.
Semantic Kernel Agent (agent.py): A Python client that bridges the Model Context Protocol (MCP) tools into Semantic Kernel functions.

1. The Bridge Server (`server.py`)

This server exposes tools to connect to the mainframe, send transaction codes, and scrape the resulting screen.

Requirements:

fastmcp
py3270 (Requires s3270 installed in the OS)

import os
import time
from fastmcp import FastMCP
from py3270 import Emulator

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

# Global emulator instance (in production, manage sessions per user)
# Ensure your container has network access (e.g. via NordLayer)
em = Emulator(visible=False)

@mcp.tool()
def connect_mainframe(host: str) -> str:
    """
    Connects to the Mainframe 3270 host.
    Args:
        host: The hostname or IP of the mainframe (e.g., 'frame.corp.local').
    """
    try:
        if not em.is_connected():
            em.connect(host)
        return f"Connected to {host}. Screen status: {em.string_get(1, 1, 80)}"
    except Exception as e:
        return f"Connection failed: {str(e)}"

@mcp.tool()
def execute_cics_command(command: str) -> str:
    """
    Executes a CICS transaction code or command.
    Args:
        command: The transaction string (e.g., 'CESN', 'CEMT').
    """
    try:
        if not em.is_connected():
            return "Error: Not connected to mainframe."

        # Clear screen or home cursor if needed (context dependent)
        em.send_string(command)
        em.send_enter()

        # Wait for host response (latency varies)
        time.sleep(1)

        # Scrape the full screen (24 rows x 80 columns)
        screen_content = []
        for row in range(1, 25):
            screen_content.append(em.string_get(row, 1, 80))

        return "\n".join(screen_content)
    except Exception as e:
        return f"Execution error: {str(e)}"

@mcp.tool()
def disconnect() -> str:
    """Terminates the TN3270 session."""
    em.terminate()
    return "Disconnected."

if __name__ == "__main__":
    mcp.run(transport='sse', host='0.0.0.0', port=8000)

2. Dockerfile

The py3270 library relies on the system-level s3270 binary. We must install this in our Docker image.

# Base image
FROM python:3.11-slim

# Install system dependencies (s3270 is required for py3270)
RUN apt-get update && apt-get install -y \
    s3270 \
    && rm -rf /var/lib/apt/lists/*

# Set working directory
WORKDIR /app

# Install Python dependencies
RUN pip install fastmcp py3270

# Copy application code
COPY server.py .

# Expose the FastMCP port
EXPOSE 8000

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

3. Client: Semantic Kernel Integration (`agent.py`)

This client demonstrates how to consume the MCP server using Semantic Kernel. We define the mcps configuration list and then use a standardized ClientSession bridge to register the tools as kernel functions.

Prerequisites:

pip install semantic-kernel mcp httpx-sse

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.functions import kernel_function
from mcp import ClientSession, StdioServerParameters
from mcp.client.sse import sse_client

# Configuration: List of MCP Servers to connect to
mcps = ["http://localhost:8000/sse"]

class McpBridgePlugin:
    """
    A generic bridge that wraps MCP tools into Semantic Kernel functions.
    """
    def __init__(self, mcp_endpoint):
        self.mcp_endpoint = mcp_endpoint

    @kernel_function(
        description="Executes a tool on the MCP server",
        name="execute_tool",
    )
    async def execute_tool(self, tool_name: str, arguments: dict = {}) -> str:
        # Connect to the MCP server for each execution
        async with sse_client(self.mcp_endpoint) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()

                # List tools to verify existence (optional but good practice)
                tools = await session.list_tools()
                available_tools = [t.name for t in tools.tools]

                if tool_name not in available_tools:
                    return f"Error: Tool '{tool_name}' not found. Available: {available_tools}"

                # Call the requested tool
                result = await session.call_tool(tool_name, arguments=arguments)

                # Return the text content
                if result.content and hasattr(result.content[0], 'text'):
                    return result.content[0].text
                return "No output returned."

async def main():
    # 1. Initialize Semantic Kernel
    kernel = Kernel()

    # 2. Register MCP Plugins
    # We iterate through our configuration list to register a bridge for each server
    print(f"🔌 Connecting to MCP servers: {mcps}")
    for url in mcps:
        plugin = McpBridgePlugin(mcp_endpoint=url)
        kernel.add_plugin(plugin, plugin_name="CicsMcpBridge")

    print("✅ MCP Tools Registered as Kernel Functions.")

    # 3. Execute a workflow (Simulating Agent Behavior)
    # The agent decides to call 'execute_tool' with specific parameters
    cics_plugin = kernel.get_plugin("CicsMcpBridge")

    # Step A: Connect to Mainframe
    print("\n🤖 Agent: Connecting to Mainframe...")
    connect_func = cics_plugin["execute_tool"]
    result = await kernel.invoke(
        connect_func,
        tool_name="connect_mainframe",
        arguments={"host": "mainframe.internal.net"}
    )
    print(f"📠 Output: {result}")

    # Step B: Check System Status
    print("\n🤖 Agent: Checking System Health (CEMT)...")
    result = await kernel.invoke(
        connect_func,
        tool_name="execute_cics_command",
        arguments={"command": "CEMT INQ TASK"}
    )
    print(f"📠 Screen Dump:\n{result}")

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

🧠 How it Works

Configuration: We define mcps = ["http://localhost:8000/sse"] to clearly state our integration points.
Bridge Pattern: The McpBridgePlugin class acts as a generic adapter. It takes an MCP URL, connects via ClientSession, and exposes a generic execute_tool function to the Kernel.
Execution: When the Semantic Kernel invokes execute_tool, the Python code dynamically opens an SSE connection to the Docker container, serializes the arguments, and returns the green-screen data.

⚠️ Production Notes

Concurrency: For high-throughput production environments, maintain a persistent ClientSession object rather than creating a new connection for every single tool call.
Prompt Engineering: You will need to provide the Agent with a system prompt that explains which tools are available via the execute_tool function, or enhance the McpBridgePlugin to dynamically register a distinct Kernel Function for every MCP tool discovered during initialization.

🛡️ Quality Assurance

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

Transparency: This page may contain affiliate links.

Semantic Kernel skills for Mainframe CICS automation

Semantic Kernel skills for Mainframe CICS automation

The Bridge: Connecting AI to the Green Screen

🏗️ Architecture

1. The Bridge Server (server.py)

2. Dockerfile

3. Client: Semantic Kernel Integration (agent.py)

🧠 How it Works

⚠️ Production Notes

🛡️ Quality Assurance

1. The Bridge Server (`server.py`)

3. Client: Semantic Kernel Integration (`agent.py`)