Building Custom MCP Connectors for Legacy Systems
Source: mortalapps.com- An MCP connector is a specialized server that translates the Model Context Protocol into legacy-specific protocols like SQL, COBOL, or proprietary RPCs.
- It solves the 'dark data' problem by providing AI agents with a standardized, type-safe interface to interact with systems of record that lack modern APIs.
- In production, these connectors serve as critical security boundaries, implementing query sandboxing and identity mapping to prevent unauthorized data access.
- Successfully deploying a connector enables agents to perform real-time reasoning across legacy databases without manual data exports or fragile ETL pipelines.
- A key tradeoff involves the complexity of maintaining schema alignment between the legacy system and the MCP tool definitions as the database evolves.
Why This Matters
Enterprises frequently struggle with 'dark data' - critical business logic and historical records trapped within legacy SQL databases and mainframe systems that lack modern REST or GraphQL interfaces. While modern AI agents excel at reasoning, they cannot natively interact with a 30-year-old DB2 instance or a complex SQL Server schema without a translation layer. The mcp connector legacy database sql approach was developed to bridge this gap by wrapping these rigid systems in the Model Context Protocol (MCP), providing a standardized way for agents to discover capabilities and execute controlled actions. Ignoring this architectural pattern leads to two primary production risks: security vulnerabilities and system instability. Allowing an LLM to generate and execute raw SQL directly against a production database is a recipe for SQL injection and catastrophic performance degradation due to unoptimized queries. By building a custom MCP connector, engineers can implement a 'Contract-First' approach where the agent only sees validated tools with strict input schemas. This approach is superior to traditional RPA (Robotic Process Automation) because it allows for dynamic, context-aware data retrieval rather than brittle, recorded scripts. It is also more efficient than building bespoke microservices for every agentic use case, as MCP provides a unified discovery mechanism that works across diverse LLM providers and agent frameworks. For teams operating in regulated industries like finance or healthcare, these connectors are the only viable way to expose sensitive legacy data to AI while maintaining strict auditability and access control.
Core Concepts
The architecture of a legacy MCP connector relies on several fundamental concepts that ensure safety and interoperability:
- MCP Tool Server: A service that implements the MCP specification, exposing legacy functions as 'tools' that an agent can call.
- Schema Introspection: The process of programmatically querying a legacy database's metadata (tables, columns, types) to generate MCP tool definitions.
- Query Sandboxing: A security layer that validates, sanitizes, and restricts the scope of queries before they reach the legacy engine.
- Identity Mapping: The translation of an agent's session identity into a legacy credential (e.g., a specific DB user or mainframe terminal ID).
- Read-Only Replicas: A production strategy where the MCP connector targets a mirrored instance of the legacy data to prevent impact on transactional workloads.
- Protocol Translation: The logic that converts MCP JSON-RPC requests into legacy-specific formats like T-SQL, PL/SQL, or fixed-width mainframe buffers.
How It Works
The lifecycle of an MCP connector for legacy systems follows a structured sequence from discovery to execution.
Phase 1: Discovery and Introspection
When the MCP server starts, it connects to the legacy system using administrative credentials to perform introspection. It queries system tables (e.g., INFORMATION_SCHEMA in SQL) to identify available entities. It then maps these entities to MCP Tool definitions, including descriptions and Pydantic-style input schemas. This ensures the agent knows exactly what parameters are required for a query.
Phase 2: Tool Registration
The server broadcasts its available tools to the MCP Host (the agent client). Each tool represents a specific legacy capability, such as get_customer_by_id or query_inventory_levels. Crucially, the server does not expose raw SQL execution unless specifically designed with extreme sandboxing.
Phase 3: Request Validation
When an agent invokes a tool, the MCP server receives a JSON-RPC request. The server first validates the input against the registered schema. If the input contains suspicious patterns (e.g., semicolons or 'DROP TABLE' strings), the server rejects the request immediately, returning a standardized error to the agent.
Phase 4: Execution and Translation
The server translates the validated parameters into a parameterized legacy query. It acquires a connection from a managed pool, executes the query, and monitors for timeouts. If the legacy system is a mainframe, this phase may involve EBCDIC encoding and handling fixed-width response buffers.
Phase 5: Response Formatting
The raw result set is converted into a structured JSON format or a Markdown table, depending on the agent's requirements. The server applies pagination if the result set exceeds the configured token limit for the agent's context window. Finally, the connection is returned to the pool, and the response is sent back via the MCP transport.
Architecture
The architecture consists of four primary layers.
- The MCP Host (Agent): This is the entry point, typically an LLM-powered application that consumes the MCP protocol.
- The MCP Connector (Server): A Python or TypeScript service that acts as the mediator. It contains the MCP SDK, the Tool Registry, and the Security Sandbox.
- The Translation Layer: A component within the connector that uses libraries like SQLAlchemy or specialized mainframe drivers to communicate with the backend.
- The Legacy System: The system of record (e.g., SQL Server, Oracle, or an IBM iSeries). Data flows from the Host as a JSON-RPC call, is transformed by the Connector into a native protocol call, and returns as a structured result set. Execution starts with a user prompt in the Host and ends with a formatted response being injected into the agent's context window.
Automated Schema Introspection
Building a connector manually for a database with hundreds of tables is unsustainable. Production-grade connectors use automated introspection to build the MCP toolset. In a SQL environment, this involves querying sys.tables and sys.columns. The connector maps SQL data types (e.g., VARCHAR, INT, DATETIME) to JSON Schema types (string, integer, string with format). For mainframes, this often requires parsing COBOL Copybooks to understand the structure of data files or transaction buffers. The resulting metadata is used to dynamically generate the list_tools response, ensuring the agent always has an up-to-date view of the available data.
Query Validation and Sandboxing
To prevent the agent from causing a Denial of Service (DoS) or performing unauthorized data exfiltration, the connector must implement a multi-layered sandbox. First, use Parameterized Queries: never concatenate strings to build a query. Second, implement Query Cost Estimation: before execution, use the legacy system's 'EXPLAIN' plan functionality to estimate the resource impact. If a query scans a million rows without an index, the connector should preemptively fail. Third, enforce Row Limits: hard-code a maximum number of rows (e.g., 500) that can be returned in a single call to prevent context window overflow and memory exhaustion.
Identity Mapping and Credential Pass-through
In an enterprise environment, the agent should not use a single 'super-user' account. Instead, the MCP connector should implement identity mapping. When the agent makes a request, it includes a session token or user ID. The connector validates this token (often via an integration with Entra ID or Auth0) and then 'impersonates' that user on the legacy system. In SQL Server, this might involve EXECUTE AS USER. In mainframe systems, this may require mapping the modern identity to a RACF (Resource Access Control Facility) ID. This ensures that the legacy system's existing security policies and audit logs remain the source of truth for data access.
Handling Mainframe EBCDIC and Fixed-Width Protocols
Connecting to a mainframe (e.g., z/OS) presents unique challenges compared to SQL. Data is often stored in EBCDIC encoding rather than UTF-8. The MCP connector must handle the character set translation during the request/response cycle. Furthermore, mainframe transactions often use fixed-width buffers where field positions are critical. The connector's translation layer must act as a 'buffer manager,' padding inputs with spaces and truncating outputs based on the COBOL definition. Using a library like ebcdic in Python allows the connector to bridge these two worlds seamlessly, presenting the agent with a clean, modern JSON interface while the backend remains unchanged.
Handling Schema Drift and Versioning
Legacy systems are rarely static. When a DBA adds a column or changes a data type, the MCP connector's tool definitions may become invalid. To handle this, implement a versioning strategy in the MCP server. The server should include a schema_version in its metadata. If the introspection logic detects a breaking change, it can either auto-update the tool definitions (if the change is additive) or flag an alert for manual intervention. For high-availability systems, running two versions of the connector during a migration ensures that agents aren't broken by backend updates.
Code Example
import sqlite3
from mcp.server.fastmcp import FastMCP
# Initialize FastMCP server for a legacy inventory system
mcp = FastMCP("LegacyInventoryConnector")
DB_PATH = "legacy_data.db"
@mcp.tool()
def get_product_stock(product_name: str) -> str:
"""Query the legacy DB for current stock levels of a specific product."""
try:
# Use a context manager for the connection
with sqlite3.connect(DB_PATH) as conn:
cursor = conn.cursor()
# ALWAYS use parameterized queries to prevent SQL injection
query = "SELECT stock_count FROM inventory WHERE item_name = ?"
cursor.execute(query, (product_name,))
result = cursor.fetchone()
if result:
return f"Stock for {product_name}: {result[0]}"
return f"Product {product_name} not found."
except Exception as e:
return f"Error accessing legacy system: {str(e)}"
if __name__ == "__main__":
mcp.run()The server starts and registers the 'get_product_stock' tool. When called with {'product_name': 'Widget-A'}, it returns 'Stock for Widget-A: 42'.
import os
import pyodbc
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("EnterpriseSQLConnector")
# Connection string from environment variables for security
CONN_STR = os.getenv("LEGACY_DB_CONN")
@mcp.tool()
def search_customers(zip_code: str, limit: int = 10) -> list:
"""Search customers by zip code with strict safety limits."""
# Enforce a hard limit to prevent resource exhaustion
safe_limit = min(limit, 50)
try:
conn = pyodbc.connect(CONN_STR)
cursor = conn.cursor()
# Parameterized query with TOP clause for safety
sql = "SELECT TOP (?) CustomerName, Email FROM Customers WHERE ZipCode = ?"
cursor.execute(sql, (safe_limit, zip_code))
rows = cursor.fetchall()
# Convert to list of dicts for clean JSON-RPC response
return [{"name": r[0], "email": r[1]} for r in rows]
except Exception as e:
# In production, log the full error but return a sanitized message
logging.error(f"Database Error: {e}")
return [{"error": "Failed to retrieve customer data."}]
finally:
if 'conn' in locals(): conn.close()Returns a list of up to 10 customer objects for the provided zip code, or a sanitized error message if the connection fails.