Chapter 55 — API Integration & Legacy Modernization

Overview

Expose AI capabilities via APIs; modernize legacy systems using strangler patterns and events. Legacy systems contain critical business logic and data but often lack the flexibility for AI integration. This chapter covers strategies for safely integrating AI with legacy systems through well-designed APIs, event-driven architecture, and incremental modernization that minimizes risk while unlocking value.

Why It Matters

APIs make AI reusable and safe. Legacy modernization unlocks data and reduces fragility, but requires careful strangler patterns and event architectures. Organizations that successfully modernize legacy systems achieve:

• Faster AI feature delivery (weeks instead of months) via API-first architecture
• Reduced system fragility by decoupling AI from core legacy logic
• Incremental migration with low risk and clear rollback paths
• Data accessibility for AI models without compromising legacy system stability
• Cost savings by selectively modernizing high-value components
• Team agility through clear API contracts enabling parallel development

Failed approaches lead to unstable systems, extended downtime, blown budgets (often 3-5x over), and abandoned modernization initiatives.

API Design Patterns for AI

API Pattern Comparison

RESTful AI API Design

# FastAPI example for AI model serving
from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel, Field
from typing import Optional, Dict
from datetime import datetime

app = FastAPI(title="AI Prediction API", version="2.1.0")

# Request/Response models with validation
class PredictionRequest(BaseModel):
    input_data: Dict[str, any] = Field(..., example={"age": 35, "income": 75000})
    model_version: Optional[str] = "latest"
    explain: bool = False

class PredictionResponse(BaseModel):
    prediction_id: str
    prediction: float
    confidence: float
    model_version: str
    explanation: Optional[Dict] = None
    timestamp: datetime = Field(default_factory=datetime.utcnow)

# Endpoints
@app.post("/v2/predictions", response_model=PredictionResponse)
async def predict(request: PredictionRequest, credentials = Depends(verify_token)):
    """Get real-time prediction from AI model"""
    user = await verify_token(credentials)
    await check_rate_limit(user.id, limit=100, window=60)

    # Get prediction
    try:
        prediction, confidence, explanation = await model_service.predict(
            input_data=request.input_data,
            model_version=request.model_version,
            explain=request.explain
        )
        return PredictionResponse(
            prediction_id=generate_id(),
            prediction=prediction,
            confidence=confidence,
            model_version=request.model_version,
            explanation=explanation if request.explain else None
        )
    except ModelUnavailableError:
        raise HTTPException(status_code=503, detail="Model unavailable")

@app.post("/v2/predictions/batch")
async def batch_predict(inputs: List[Dict], callback_url: Optional[str], credentials = Depends(verify_token)):
    """Submit batch prediction job (async processing)"""
    user = await verify_token(credentials)
    await check_rate_limit(user.id, limit=10, window=60)

    job_id = create_batch_job(user.id, inputs)
    background_tasks.add_task(process_batch, job_id, inputs, callback_url)

    return {"job_id": job_id, "status": "processing", "status_url": f"/v2/predictions/batch/{job_id}"}

# OpenAPI schema auto-generated at /docs

API Versioning Strategies

graph LR
    Client[API Client] --> Gateway[API Gateway]

    Gateway --> V1{Version?}

    V1 -->|v1| Route1[Route to v1 Service]
    V1 -->|v2| Route2[Route to v2 Service]
    V1 -->|v3| Route3[Route to v3 Service]

    Route1 --> Service1[AI Service v1<br/>Deprecated]
    Route2 --> Service2[AI Service v2<br/>Current]
    Route3 --> Service3[AI Service v3<br/>Beta]

    Service1 --> Adapter[Compatibility Adapter]
    Adapter --> Service2

    style Service1 fill:#ffcccc
    style Service2 fill:#ccffcc
    style Service3 fill:#ccccff

Versioning Approaches:

Recommendation: Use URL path versioning for major versions, headers for minor versions.

Legacy Modernization Strategies

Strangler Fig Pattern

graph TB
    subgraph Phase 1 - Initial
        LB1[Load Balancer] --> Legacy1[Legacy System<br/>100% Traffic]
    end

    subgraph Phase 2 - Facade
        LB2[API Gateway/Facade] --> Legacy2[Legacy System<br/>100% Traffic]
        LB2 -.->|No traffic yet| Modern2[Modern Service<br/>0% Traffic]
    end

    subgraph Phase 3 - Incremental
        LB3[API Gateway] --> Route{Routing Logic}
        Route -->|80%| Legacy3[Legacy System]
        Route -->|20%| Modern3[Modern Service]
    end

    subgraph Phase 4 - Completion
        LB4[API Gateway] --> Modern4[Modern Service<br/>100% Traffic]
        Legacy4[Legacy System<br/>Decommissioned]
    end

    Phase 1 --> Phase 2
    Phase 2 --> Phase 3
    Phase 3 --> Phase 4

Implementation Example

# Strangler facade pattern for gradual migration
from fastapi import FastAPI, Request
import httpx
import random

app = FastAPI()

class StranglerFacade:
    def __init__(self):
        self.legacy_client = httpx.AsyncClient(base_url="http://legacy-app:8080")
        self.modern_client = httpx.AsyncClient(base_url="http://modern-service:8080")

        # Gradual rollout configuration
        self.rollout_config = {
            "/api/customers": {"modern_percentage": 0, "shadow_mode": False},  # Not ready
            "/api/orders": {"modern_percentage": 20, "shadow_mode": False},  # 20% canary
            "/api/recommendations": {"modern_percentage": 100, "shadow_mode": False},  # Complete
            "/api/inventory": {"modern_percentage": 50, "shadow_mode": True}  # 50% + shadow testing
        }

    async def route_request(self, path: str, request: Request):
        """Route request to legacy or modern service based on config"""
        config = self.rollout_config.get(path, {"modern_percentage": 0})
        use_modern = random.random() * 100 < config["modern_percentage"]

        if config.get("shadow_mode"):
            return await self.shadow_mode_request(path, request)  # Call both, compare
        elif use_modern:
            return await self.call_modern_service(path, request)
        else:
            return await self.call_legacy_system(path, request)

    async def call_legacy_system(self, path: str, request: Request):
        """Forward request to legacy system"""
        try:
            response = await self.legacy_client.request(
                method=request.method, url=path,
                headers=dict(request.headers), content=await request.body(), timeout=30.0
            )
            return {"source": "legacy", "status_code": response.status_code, "data": response.json()}
        except httpx.TimeoutException:
            return await self.call_modern_service(path, request)  # Fallback on timeout

    async def call_modern_service(self, path: str, request: Request):
        """Forward request to modern service"""
        response = await self.modern_client.request(
            method=request.method, url=path,
            headers=dict(request.headers), content=await request.body(), timeout=30.0
        )
        return {"source": "modern", "status_code": response.status_code, "data": response.json()}

    async def shadow_mode_request(self, path: str, request: Request):
        """Call both systems, log discrepancies, return legacy result"""
        legacy_task = asyncio.create_task(self.call_legacy_system(path, request))
        modern_task = asyncio.create_task(self.call_modern_service(path, request))

        legacy_response, modern_response = await asyncio.gather(legacy_task, modern_task, return_exceptions=True)

        # Compare results and log discrepancies
        if not isinstance(modern_response, Exception) and legacy_response["data"] != modern_response["data"]:
            await log_discrepancy(path, legacy_response["data"], modern_response["data"])

        return legacy_response  # Always return legacy in shadow mode

# FastAPI integration
facade = StranglerFacade()

@app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"])
async def proxy(path: str, request: Request):
    """Proxy all requests through strangler facade"""
    result = await facade.route_request(f"/{path}", request)
    return result["data"]

@app.post("/admin/rollout")
async def update_rollout(path: str, percentage: int):
    """Update rollout percentage for a path"""
    if path in facade.rollout_config:
        facade.rollout_config[path]["modern_percentage"] = percentage
        return {"status": "updated", "path": path, "percentage": percentage}
    return {"status": "error", "message": "Path not found"}

Event-Driven Integration

Event Architecture Patterns

graph TB
    subgraph Legacy System
        Legacy[Legacy Application]
        Legacy_DB[(Legacy Database)]
    end

    subgraph Event Infrastructure
        CDC[Change Data Capture<br/>Debezium]
        Kafka[Event Stream<br/>Kafka/EventBridge]
        Schema[Schema Registry]
    end

    subgraph Modern Services
        Transform[Event Transformer]
        AI_Service[AI Service]
        Analytics[Analytics Service]
        Notification[Notification Service]
    end

    subgraph Data Lake
        Lake[(Data Lake)]
        DW[(Data Warehouse)]
    end

    Legacy --> Legacy_DB
    Legacy_DB --> CDC
    CDC --> Kafka
    Kafka --> Schema

    Kafka --> Transform
    Transform --> AI_Service
    Transform --> Analytics
    Transform --> Notification

    Kafka --> Lake
    Lake --> DW

Event Schema Management

# Event schema with Avro
from confluent_kafka.avro import AvroProducer, AvroConsumer
import avro

# Define Avro schema for events (simplified)
order_event_schema = """
{
  "namespace": "com.company.events",
  "type": "record",
  "name": "OrderEvent",
  "fields": [
    {"name": "event_id", "type": "string"},
    {"name": "event_type", "type": {"type": "enum", "name": "EventType", "symbols": ["CREATED", "UPDATED", "CANCELLED"]}},
    {"name": "timestamp", "type": "long", "logicalType": "timestamp-millis"},
    {"name": "order_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "order_data", "type": {
      "type": "record",
      "name": "OrderData",
      "fields": [
        {"name": "total_amount", "type": "double"},
        {"name": "items", "type": {"type": "array", "items": "string"}}
      ]
    }}
  ]
}
"""

# Event producer
class EventProducer:
    def __init__(self, bootstrap_servers, schema_registry_url):
        self.producer = AvroProducer({
            'bootstrap.servers': bootstrap_servers,
            'schema.registry.url': schema_registry_url
        }, default_value_schema=avro.loads(order_event_schema))

    def publish_order_event(self, event_type, order_id, customer_id, order_data):
        """Publish order event to Kafka"""
        event = {
            'event_id': str(uuid.uuid4()),
            'event_type': event_type,
            'timestamp': int(datetime.utcnow().timestamp() * 1000),
            'order_id': order_id,
            'customer_id': customer_id,
            'order_data': order_data
        }
        # Publish with key for partitioning (ensures ordering per customer)
        self.producer.produce(topic='orders', key=customer_id, value=event)
        self.producer.flush()
        return event['event_id']

# Event consumer for AI processing
class OrderEventConsumer:
    def __init__(self, bootstrap_servers, schema_registry_url, group_id):
        self.consumer = AvroConsumer({
            'bootstrap.servers': bootstrap_servers,
            'group.id': group_id,
            'schema.registry.url': schema_registry_url,
            'enable.auto.commit': False  # Manual commit for at-least-once processing
        })
        self.consumer.subscribe(['orders'])

    async def consume_and_process(self):
        """Consume events and trigger AI processing"""
        while True:
            msg = self.consumer.poll(timeout=1.0)
            if msg and not msg.error():
                event = msg.value()
                await self.process_order_event(event)
                self.consumer.commit(msg)  # Commit after successful processing

    async def process_order_event(self, event):
        """Process order event with AI services"""
        if event['event_type'] == 'CREATED':
            # Trigger fraud detection
            fraud_score = await ai_service.predict_fraud(event['customer_id'], event['order_data'])
            if fraud_score > 0.8:
                await alert_service.send_fraud_alert(event['order_id'], fraud_score)

            # Generate product recommendations
            recommendations = await ai_service.get_recommendations(event['customer_id'])
            await notification_service.send_recommendations(event['customer_id'], recommendations)

        elif event['event_type'] == 'CANCELLED':
            await ai_service.update_churn_risk(event['customer_id'], event='order_cancelled')

Idempotency in Event Processing

# Idempotent event consumer
class IdempotentEventProcessor:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.processed_ttl = 86400  # 24 hours

    async def process_event(self, event):
        """Process event with idempotency guarantee"""
        event_id = event['event_id']

        # Check if already processed
        if await self.redis.exists(f"processed:{event_id}"):
            return {"status": "duplicate", "event_id": event_id}

        try:
            result = await self.do_processing(event)  # Actual business logic

            # Mark as processed with TTL
            await self.redis.setex(f"processed:{event_id}", self.processed_ttl,
                                  json.dumps({"result": result, "timestamp": datetime.utcnow().isoformat()}))

            return {"status": "processed", "event_id": event_id, "result": result}
        except Exception as e:
            logger.error(f"Error processing event {event_id}: {e}")
            raise  # Don't mark as processed on error - allow retry

Data Migration Strategies

Migration Patterns

Data Synchronization During Migration

# Bi-directional sync during migration
class DataSyncManager:
    def __init__(self, legacy_db, modern_db):
        self.legacy_db = legacy_db
        self.modern_db = modern_db

    async def sync_legacy_to_modern(self, table_name, record_id):
        """Sync single record from legacy to modern"""
        legacy_record = await self.legacy_db.get(table_name, record_id)
        modern_record = self.transform_to_modern(legacy_record)
        await self.modern_db.upsert(table_name, record_id, modern_record)
        await sync_log.record("legacy_to_modern", table_name, record_id)

    async def sync_modern_to_legacy(self, table_name, record_id):
        """Sync single record from modern to legacy"""
        modern_record = await self.modern_db.get(table_name, record_id)
        legacy_record = self.transform_to_legacy(modern_record)
        await self.legacy_db.update(table_name, record_id, legacy_record)
        await sync_log.record("modern_to_legacy", table_name, record_id)

    async def bulk_migration(self, table_name, batch_size=1000):
        """Bulk migrate table from legacy to modern"""
        offset = 0
        total_migrated = 0

        while True:
            records = await self.legacy_db.fetch_batch(table_name, offset=offset, limit=batch_size)
            if not records:
                break

            # Transform and insert
            for record in records:
                modern_record = self.transform_to_modern(record)
                await self.modern_db.upsert(table_name, record['id'], modern_record)

            total_migrated += len(records)
            offset += batch_size
            logger.info(f"Migrated {total_migrated} records from {table_name}")
            await asyncio.sleep(1)  # Rate limit

        return total_migrated

    def transform_to_modern(self, legacy_record):
        """Transform legacy schema to modern schema"""
        return {
            'id': legacy_record['LEGACY_ID'],
            'created_at': parse_legacy_date(legacy_record['CREATE_DT']),
            'customer': {
                'name': f"{legacy_record['FIRST_NM']} {legacy_record['LAST_NM']}",
                'email': legacy_record['EMAIL_ADDR']
            },
            'embedding': generate_embedding(legacy_record)  # AI enhancement
        }

    def transform_to_legacy(self, modern_record):
        """Transform modern schema to legacy schema (lossy)"""
        name_parts = modern_record['customer']['name'].split(' ', 1)
        return {
            'LEGACY_ID': modern_record['id'],
            'CREATE_DT': format_legacy_date(modern_record['created_at']),
            'FIRST_NM': name_parts[0],
            'LAST_NM': name_parts[1] if len(name_parts) > 1 else '',
            'EMAIL_ADDR': modern_record['customer']['email']
            # Note: embedding field dropped (not in legacy schema)
        }

API Gateway Configuration

Kong Gateway Example

# Kong declarative configuration for AI API
_format_version: "3.0"

services:
  - name: ai-prediction-service
    url: http://ai-service.internal:8080
    protocol: http
    connect_timeout: 60000
    write_timeout: 60000
    read_timeout: 60000

    routes:
      - name: predictions-v2
        paths:
          - /v2/predictions
        methods:
          - POST
        strip_path: false

    plugins:
      # Rate limiting
      - name: rate-limiting
        config:
          minute: 100
          hour: 1000
          policy: redis
          redis_host: redis.internal
          redis_port: 6379

      # Authentication
      - name: jwt
        config:
          claims_to_verify:
            - exp
          key_claim_name: iss
          secret_is_base64: false

      # Request validation
      - name: request-validator
        config:
          body_schema: |
            {
              "type": "object",
              "required": ["input_data"],
              "properties": {
                "input_data": {"type": "object"},
                "model_version": {"type": "string"},
                "explain": {"type": "boolean"}
              }
            }

      # Response transformer (add headers)
      - name: response-transformer
        config:
          add:
            headers:
              - X-API-Version: v2
              - X-RateLimit-Remaining: $(ratelimit.remaining)

      # CORS
      - name: cors
        config:
          origins:
            - https://app.company.com
          methods:
            - GET
            - POST
          headers:
            - Authorization
            - Content-Type
          exposed_headers:
            - X-API-Version
          credentials: true
          max_age: 3600

      # Logging
      - name: file-log
        config:
          path: /var/log/kong/ai-api.log
          reopen: true

      # Prometheus metrics
      - name: prometheus

  # Legacy system proxy (for strangler pattern)
  - name: legacy-service
    url: http://legacy-app.internal:8080
    routes:
      - name: legacy-orders
        paths:
          - /api/orders
        strip_path: false

    plugins:
      - name: request-transformer
        config:
          add:
            headers:
              - X-Legacy-Routing: true

# Global plugins
plugins:
  - name: correlation-id
    config:
      header_name: X-Correlation-ID
      generator: uuid
      echo_downstream: true

Evaluation Metrics

API Performance

Migration Progress

Case Study: Logistics Route Optimization

Background

A logistics provider has a 20-year-old mainframe system for route planning. Manual routing leads to 15% inefficiency. Modern AI optimizer can improve by 20-25% but requires careful integration.

Challenge

• Legacy system: COBOL on mainframe, batch processing overnight
• Fragile codebase: 500K lines, limited documentation
• High availability requirement: 99.95% uptime SLA
• Complex data: 50+ tables with arcane schemas
• Risk aversion: Any outage costs $100K+/hour

Implementation Strategy

Phase 1: API Facade (Months 1-3)

• Wrapped legacy system with REST API
• No changes to legacy logic
• Read-only endpoints for order, route, vehicle data
• Authentication and rate limiting via API gateway

Phase 2: Shadow Mode AI (Months 4-6)

• Deployed AI optimizer in parallel
• Compared AI routes vs. legacy routes (shadow mode)
• Measured AI improvement: 23% fewer miles, 18% faster delivery
• No changes to production routing yet

Phase 3: Hybrid Routing (Months 7-10)

• Introduced routing logic in API gateway
• 10% of routes: AI optimizer
• 90% of routes: Legacy system
• Gradual increase: 10% → 25% → 50% → 75%
• Rollback capability via config change

Phase 4: Event-Driven Updates (Months 11-14)

• Implemented CDC on legacy database
• Real-time event stream for order updates
• AI optimizer subscribed to events
• Enabled dynamic re-routing based on traffic, weather

Phase 5: Full Migration (Months 15-18)

• 100% traffic to AI optimizer
• Legacy system on standby for 6 months
• Data migration to modern PostgreSQL
• Decommissioned mainframe

Architecture

graph TB
    subgraph Phase 3 - Hybrid Routing
        API[API Gateway<br/>Kong]
        Router{Routing Logic<br/>10% AI, 90% Legacy}

        Legacy[Legacy Mainframe<br/>Route Planning]
        AI[AI Optimizer<br/>ML Service]

        DB_Legacy[(Legacy DB<br/>DB2)]
        DB_Modern[(Modern DB<br/>PostgreSQL)]

        CDC[Change Data Capture<br/>Debezium]
        Kafka[Event Stream<br/>Kafka]
    end

    API --> Router
    Router -->|90%| Legacy
    Router -->|10%| AI

    Legacy --> DB_Legacy
    AI --> DB_Modern

    DB_Legacy --> CDC
    CDC --> Kafka
    Kafka --> AI

    AI --> Metrics[Metrics<br/>Prometheus]
    Legacy --> Metrics

Results

Cost-Benefit:

• Migration cost: $1.8M
• Annual savings: $5.4M (fuel) +$1.2M (mainframe license)
• Payback: 3.3 months
• 3-year NPV: $17.2M

Lessons Learned

1. Shadow Mode is Critical: 2 months of shadow testing caught edge cases that would have caused production issues
2. Gradual Rollout Works: 10% increments allowed validation at each step
3. Rollback Saves Projects: Two incidents rolled back in <5 min prevented major outages
4. Event Streams Enable Real-Time: CDC + Kafka unlocked dynamic re-routing capability
5. API-First Pays Off: API facade enabled parallel development of modern services

Implementation Checklist

API Design

• Define API contracts with OpenAPI specs
• Establish versioning strategy
• Design request/response schemas with validation
• Plan for authentication and authorization
• Set rate limits and quotas
• Document API for developers

API Gateway Setup

• Deploy API gateway (Kong, Apigee, AWS API Gateway)
• Configure routing rules
• Implement authentication plugins
• Set up rate limiting
• Enable request/response validation
• Configure CORS policies
• Set up logging and monitoring

Legacy Integration

• Map legacy system capabilities and data
• Identify strangler pattern boundaries
• Build adapters for legacy system
• Implement data transformation logic
• Set up bi-directional sync if needed
• Create rollback procedures

Event Infrastructure

• Deploy event streaming platform (Kafka, EventBridge)
• Set up schema registry
• Implement CDC for legacy database
• Design event schemas
• Build event producers and consumers
• Ensure idempotent event processing

Migration Execution

• Start with read-only shadow mode
• Validate modern service matches legacy
• Implement gradual traffic shifting (10%, 25%, 50%, 75%, 100%)
• Monitor metrics at each stage
• Test rollback procedures
• Plan for data migration
• Schedule legacy decommissioning

Monitoring & Operations

• Set up API performance dashboards
• Create alerts for errors and latency
• Monitor migration progress metrics
• Implement distributed tracing
• Log all requests for audit
• Establish on-call procedures

Best Practices

Do's

1. Design API-First: Define contracts before implementation
2. Version from Day 1: Plan for backward compatibility
3. Test in Shadow Mode: Validate before switching traffic
4. Incremental Rollout: Gradual traffic shifting with rollback capability
5. Monitor Everything: Metrics, logs, traces for both legacy and modern
6. Document Thoroughly: API docs, runbooks, architecture diagrams
7. Plan for Rollback: Always have a quick path back to legacy

Don'ts

1. Don't Big Bang Migrate: Incremental is safer
2. Don't Skip Validation: Shadow mode catches issues
3. Don't Neglect Legacy: Keep legacy stable during migration
4. Don't Hardcode Routing: Use configuration for flexibility
5. Don't Ignore Performance: Set SLOs and monitor
6. Don't Forget Data Sync: Bi-directional sync during transition

Common Pitfalls

Technology Recommendations

Deliverables

1. API Contracts

• OpenAPI/Swagger specifications
• Request/response schemas
• Authentication requirements
• Rate limits and quotas
• Versioning policy

2. Migration Plan

• Phased rollout schedule
• Traffic shifting strategy
• Rollback procedures
• Data migration approach
• Risk assessment and mitigation

3. Integration Architecture

• System architecture diagrams
• Event flow diagrams
• Data flow diagrams
• Strangler pattern boundaries
• API gateway configuration

4. Operational Runbooks

• Deployment procedures
• Rollback procedures
• Monitoring and alerting setup
• Incident response playbooks
• Troubleshooting guides