Los agentes GenAI no existen en aislamiento. Necesitan integrarse con sistemas corporativos existentes: CRM, ERP, bases de datos, APIs legacy, mainframes. El GenAI Integration Architect es quien estandariza la integración entre agentes GenAI y sistemas corporativos, habilitando interoperabilidad segura y controlada.
El Desafío: Connecting the Dots
Una empresa típica tiene:
100+ sistemas corporativos (SAP, Salesforce, ServiceNow, etc.)
Legacy systems con décadas de antigüedad
APIs inconsistentes: REST, SOAP, GraphQL, RPC
Datos en silos: Sin estándares comunes
Protocolos de seguridad diversos
Latencias variables: Algunos sistemas son lentos
El problema: ¿Cómo habilitar que agentes GenAI accedan a estos sistemas de forma segura, eficiente y mantenible, sin crear un spaghetti de integraciones?
El Rol: Arquitecto de Conectividad Inteligente
Un GenAI Integration Architect diseña:
Integration Patterns: Cómo agentes acceden a sistemas
API Gateway/Proxy: Single point of access con control
Tool Abstractions: Wrappers consistentes sobre APIs heterogéneas
Security & Authorization: Quién puede hacer qué, cuándo
Error Handling: Resiliencia ante fallos de sistemas externos
Rate Limiting & Quotas: Prevenir sobrecarga de systems
Monitoring & Tracing: Observabilidad de integraciones
Competencias Técnicas Core
1. Integration Patterns para GenAI
Tool Calling (Function Calling):
El modelo decide qué herramientas usar y con qué argumentos.
# Ejemplo: OpenAI Function Calling
tools = [
{
"type": "function",
"function": {
"name": "get_customer_info",
"description": "Retrieves customer information by ID",
"parameters": {
"type": "object",
"properties": {
"customer_id": {
"type": "string",
"description": "The customer ID"
}
},
"required": ["customer_id"]
}
}
},
{
"type": "function",
"function": {
"name": "create_support_ticket",
"description": "Creates a new support ticket",
"parameters": {
"type": "object",
"properties": {
"title": {"type": "string"},
"description": {"type": "string"},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"]
}
},
"required": ["title", "description"]
}
}
}
]
# LLM decides which tool to call
response = llm.chat(
messages=conversation_history,
tools=tools,
tool_choice="auto"
)
# Execute tool
if response.tool_calls:
for tool_call in response.tool_calls:
result = execute_tool(tool_call.function.name, tool_call.function.arguments)
# Feed result back to LLM
API Orchestration:
El agente no llama APIs directamente. Va a través de orchestrator.
# Agent request
agent_query = "Get customer John Doe's account balance"
# Orchestrator:
1. Parse query → entities: name="John Doe"
2. Lookup customer ID from name (API call to CRM)
3. Get account ID from customer ID (API call to Core Banking)
4. Get balance from account ID (API call to Account Service)
5. Format response para agent
# Agent recibe resultado limpio sin conocer complejidad interna
Event-Driven Integration:
En vez de polling, systems emiten eventos.
# Example: Customer update event
{
"event_type": "customer_updated",
"customer_id": "12345",
"changed_fields": ["address", "phone"],
"timestamp": "2026-03-28T10:15:00Z"
}
# Agent puede suscribirse a eventos relevantes
# Y actualizar su contexto/memoria automáticamente
2. API Gateway para GenAI
Centralized Control:
Agent → API Gateway → Backend Systems
↓
- Authentication
- Authorization
- Rate limiting
- Logging/Monitoring
- Transformation
- Caching
Implementation Example:
# Kong / AWS API Gateway / Azure APIM / Custom
# Configuration:
@app.route("/api/tools/get_customer", methods=["POST"])
@require_auth
@rate_limit(max_calls=100, window=60) # 100 calls/min
@log_request
def get_customer():
# Validate request from agent
request_data = validate_schema(request.json)
# Authorization check
if not can_access_customer(current_agent, request_data['customer_id']):
return {"error": "Unauthorized"}, 403
# Call backend
try:
result = crm_service.get_customer(request_data['customer_id'])
return transform_response(result)
except ServiceUnavailable:
return {"error": "Service temporarily unavailable"}, 503
Challenge: Cada sistema tiene API diferente.
Solution: Abstraction layer con interface consistente.
# Base Tool Interface
class Tool(ABC):
@abstractmethod
def name(self) -> str:
pass
@abstractmethod
def description(self) -> str:
pass
@abstractmethod
def parameters_schema(self) -> dict:
pass
@abstractmethod
def execute(self, **kwargs) -> ToolResult:
pass
# Concrete Tools
class GetCustomerInfoTool(Tool):
def name(self):
return "get_customer_info"
def description(self):
return "Retrieves customer information by ID"
def parameters_schema(self):
return {
"customer_id": {"type": "string", "required": True}
}
def execute(self, customer_id: str) -> ToolResult:
# Internal implementation details hidden
# Puede llamar a Salesforce, SAP, custom DB, etc.
data = self._fetch_from_crm(customer_id)
return ToolResult(success=True, data=data)
class CreateTicketTool(Tool):
def name(self):
return "create_support_ticket"
def execute(self, title: str, description: str, priority: str) -> ToolResult:
# Abstrae llamada a ServiceNow, Jira, Zendesk, etc.
ticket = self._create_in_ticketing_system(title, description, priority)
return ToolResult(success=True, data={"ticket_id": ticket.id})
# Tool Registry
tool_registry = {
"get_customer_info": GetCustomerInfoTool(),
"create_support_ticket": CreateTicketTool(),
# ... 50 more tools
}
# Agent usage (consistent regardless of backend)
result = tool_registry["get_customer_info"].execute(customer_id="12345")
4. Authorization & Security
Challenge: ¿Quién puede acceder qué?
RBAC (Role-Based Access Control):
# Agent tiene roles
agent_context = {
"agent_id": "customer_support_agent_1",
"roles": ["customer_support", "read_customer_data"],
"user_on_behalf": "user_12345" # Agent actúa en nombre de usuario
}
# Tool tiene permissions requeridos
tool_permissions = {
"get_customer_info": ["read_customer_data"],
"delete_customer": ["admin"],
"get_financial_report": ["finance_viewer"]
}
# Check antes de ejecución
def can_execute_tool(agent_context, tool_name):
required_permissions = tool_permissions.get(tool_name, [])
agent_roles = agent_context["roles"]
return any(perm in agent_roles for perm in required_permissions)
ABAC (Attribute-Based Access Control):
Más granular, basado en atributos y condiciones.
# Policy ejemplo
policy = {
"effect": "allow",
"action": "get_customer_info",
"conditions": {
"customer_region": {"matches": agent.region},
"time": {"between": ["08:00", "18:00"]},
"customer_id": {"not_in": blacklist}
}
}
# Dynamic evaluation
def evaluate_policy(agent, tool, parameters, policies):
for policy in policies:
if matches_conditions(policy.conditions, agent, parameters):
return policy.effect == "allow"
return False # Deny by default
Secrets & Credentials:
# Agents NEVER tienen credenciales directamente
# Integration layer maneja auth
class SalesforceIntegration:
def __init__(self):
# Credentials desde secret store
self.client = Salesforce(
username=get_secret("salesforce_user"),
password=get_secret("salesforce_pass"),
security_token=get_secret("salesforce_token")
)
def get_customer(self, customer_id):
return self.client.query(f"SELECT * FROM Contact WHERE Id='{customer_id}'")
# Agent solo llama a tool, no conoce credenciales
5. Error Handling & Resilience
Systems Fail. Agents Must Handle Gracefully.
Retry Logic:
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10)
)
def call_external_api(endpoint, data):
response = requests.post(endpoint, json=data, timeout=5)
response.raise_for_status()
return response.json()
Circuit Breaker:
from circuitbreaker import circuit
@circuit(failure_threshold=5, recovery_timeout=60)
def call_flaky_service(request):
# If service fails 5 times, circuit opens
# All calls fail fast for 60 seconds
# Then half-open state to test recovery
return service.call(request)
Fallback Strategies:
def get_customer_info(customer_id):
try:
# Try primary source (CRM)
return crm.get_customer(customer_id)
except ServiceUnavailable:
# Fallback to cache
cached = cache.get(f"customer:{customer_id}")
if cached:
return {"data": cached, "source": "cache", "warning": "CRM unavailable"}
# Fallback to secondary source
try:
return data_warehouse.get_customer(customer_id)
except:
# Graceful degradation
return {"error": "Customer data temporarily unavailable"}
Partial Success:
# Agent solicita datos de 5 fuentes
# 4 exitosos, 1 falla
# Return partial data + error report
results = {
"customer_info": crm.get_customer(), # Success
"account_balance": core_banking.get_balance(), # Success
"recent_transactions": transaction_service.get_recent(), # Success
"credit_score": credit_bureau.get_score(), # FAILED
"loan_eligibility": loan_service.check() # Success
}
# Agent puede trabajar con data parcial
# E informar al usuario: "Nota: Credit score no disponible temporalmente"
6. Rate Limiting & Throttling
Protect Backend Systems:
# Per-agent rate limits
rate_limits = {
"customer_support_agent": {
"get_customer_info": 100/minute,
"create_ticket": 20/minute
},
"sales_agent": {
"get_customer_info": 200/minute,
"create_opportunity": 50/minute
}
}
# Implementation
from redis import Redis
from time import time
def is_rate_limited(agent_id, tool_name, limit):
key = f"rate_limit:{agent_id}:{tool_name}"
current = redis.get(key) or 0
if int(current) >= limit:
return True # Rate limited
# Increment counter
redis.incr(key)
redis.expire(key, 60) # 1 minute window
return False
Backpressure:
# Si backend está sobrecargado, señalizar a agent para slow down
if backend_queue_length > threshold:
return {
"status": "throttled",
"retry_after": 30, # seconds
"message": "System under heavy load, please retry"
}
Problema: Cada sistema habla diferente "idioma".
# CRM Response (Salesforce)
{
"Id": "0031x000004XXXYYY",
"FirstName": "John",
"LastName": "Doe",
"Email": "john.doe@example.com",
"AccountId": "0011x000004YYYZZZ"
}
# Agent expects standard format
{
"customer_id": "12345",
"name": {"first": "John", "last": "Doe"},
"email": "john.doe@example.com",
"account_id": "67890"
}
# Transformation Layer
def transform_crm_to_standard(crm_data):
return {
"customer_id": crm_data["Id"],
"name": {
"first": crm_data["FirstName"],
"last": crm_data["LastName"]
},
"email": crm_data["Email"],
"account_id": crm_data["AccountId"]
}
Schema Registry:
# Maintain schemas for all integrations
schemas = {
"customer_v1": {
"customer_id": "string",
"name": {"first": "string", "last": "string"},
"email": "string",
...
},
"account_v1": {...},
"transaction_v2": {...}
}
# Validate responses antes de return to agent
def validate_and_transform(data, schema_name):
schema = schemas[schema_name]
validated = validate_against_schema(data, schema)
return validated
8. Async & Long-Running Operations
Problema: Algunos operations tardan minutos/horas.
Pattern: Job Queue
# Agent solicita operación larga (ej: generar reporte complejo)
# Synchronous (bad para operations largas)
result = generate_report(params) # Blocks por 10 min
return result
# Asynchronous (good)
job_id = queue_report_generation(params)
return {
"status": "processing",
"job_id": job_id,
"check_status_url": f"/api/jobs/{job_id}/status"
}
# Agent puede:
# 1. Polling: Check status periodically
# 2. Webhook: system notifica cuando complete
# 3. Informar usuario: "Report en proceso, te notificaré"
Implementation:
# Celery, RQ, AWS SQS, etc.
from celery import Celery
app = Celery('tasks', broker='redis://localhost')
@app.task
def generate_complex_report(customer_id, params):
# Long running
report = complex_computation(customer_id, params)
# Store result
store_result(report)
# Notify
send_webhook_to_agent(report_url)
Strategic Caching:
# Cache frecuencia-based
cache_strategies = {
"customer_info": {
"ttl": 300, # 5 min (cambia poco)
"cache_key": lambda params: f"customer:{params['customer_id']}"
},
"account_balance": {
"ttl": 60, # 1 min (cambia más frecuentemente)
"cache_key": lambda params: f"balance:{params['account_id']}"
},
"real_time_stock_price": {
"ttl": 0, # Sin cache (tiempo real)
"cache_key": None
}
}
def call_with_cache(tool_name, params):
strategy = cache_strategies.get(tool_name)
if strategy["ttl"] == 0:
# No cache
return call_backend(tool_name, params)
cache_key = strategy["cache_key"](params)
cached = redis.get(cache_key)
if cached:
return json.loads(cached)
# Cache miss
result = call_backend(tool_name, params)
redis.setex(cache_key, strategy["ttl"], json.dumps(result))
return result
10. Observability de Integraciones
Distributed Tracing:
Metrics:
# Track integration health
metrics = {
"integration.call_count": Counter,
"integration.latency": Histogram,
"integration.error_rate": Gauge,
"integration.cache_hit_rate": Gauge
}
@metrics["integration.call_count"].labels(service="crm", method="get_customer").inc()
@metrics["integration.latency"].labels(service="crm").observe(duration)
Stack Tecnológico
API Gateway
Kong: Open source, plugin ecosystem
AWS API Gateway: Managed AWS
Azure API Management: Managed Azure
Apigee (Google): Enterprise-grade
Tyk: Open source alternative
Integration Frameworks
LangChain Tools: Python, easy tool abstractions
LlamaIndex Tool Specs: Similar
Semantic Kernel: Microsoft's framework
Custom frameworks: For specific needs
Message Queues
RabbitMQ: Reliable, mature
Apache Kafka: High throughput, streaming
AWS SQS/SNS: Managed queuing
Redis Streams: Lightweight
Workflow Orchestration
Temporal: Durable workflows
Apache Airflow: Batch orchestration
AWS Step Functions: State machines
Prefect: Modern Python workflows
Service Mesh (for microservices)
Arquitectura de Referencia
Casos de Uso en Banca
1. Customer Support Agent
Necesita acceder a:
CRM (customer info)
Core Banking (account balance)
Transaction System (history)
Loan System (loan status)
Ticket System (create ticket)
Integration arquitectura:
API Gateway maneja auth & rate limiting
Tool abstractions para cada sistema
Orchestrator coordina llamadas múltiples
Caching agresivo para customer info
Circuit breakers para resiliencia
2. Sales Agent
Necesita:
CRM (leads, opportunities)
Product Catalog
Pricing Engine
Email System (send proposals)
Calendar (schedule meetings)
Desafíos:
Real-time pricing (no cache)
Email requires async (no esperar envío)
Calendar sync bidireccional (Google Cal, Outlook)
3. Compliance Agent
Necesita:
Security critical:
Métricas de Éxito
Integration Health:
Availability: 99.9%+ uptime
Latency p95: < 500ms per call
Error rate: < 0.1%
Cache hit rate: > 60% (donde aplicable)
Agent Productivity:
Tool call success rate: > 95%
Average tools per query: Benchmark & optimize
End-to-end latency: Agent query → final response
Cost:
Desafíos Únicos
The Legacy Problem
Muchos sistemas legacy no tienen APIs modernas. Requires custom adapters, screen scraping, o mainframe connectors.
Eventual Consistency
Datos pueden estar out of sync entre sistemas. Agent necesita context sobre data freshness.
Un sistema slow puede degradar toda la experience. Timeouts y fallbacks son críticos.
Schema Evolution
APIs cambian versiones. Integration layer debe handle múltiples versiones.
El Futuro: Universal Connectors
AI-Powered Integration:
LLMs que generan integration code automáticamente
Natural language API specs → working connectors
Self-healing integrations que se adaptan a cambios
Standard Protocols:
Industria está convergiendo en APIs estándar (OpenAPI, gRPC)
GenAI agents could become interoperable across companies
Conclusión
El GenAI Integration Architect es el traductor universal entre el mundo de los agentes inteligentes y el laberinto de sistemas corporativos. Sin esta layer, cada agente se convierte en un proyecto de integración monumental. Con arquitectura bien diseñada, agregar un nuevo agente es plug-and-play.
En banca, donde sistemas son legacy, complejos y críticos, la integración no es una afterthought - es el enabling layer que determina si GenAI puede o no agregar valor real.
Conectividad inteligente = Agentes útiles.
¿Cómo integras tus agentes con sistemas corporativos? ¿Qué patrones usas?
#GenAI #Integration #API #EnterpriseArchitecture #Interoperability #Microservices
