Integration Patterns

Overview

Agents are most powerful when integrated with existing systems—databases, APIs, tools, workflows.

Integration patterns determine deployment success.

-

API & Webhook Integration

RESTful Agent Endpoints

from fastapi import FastAPI
from pydantic import BaseModel

class AgentAPI:
 """REST API for agent"""

 def __init__(self):
 self.app = FastAPI()
 self.agent = Agent()

 @self.app.post("/agent/process")
 async def process(request: ProcessRequest):
 """Run agent on input"""

 try:
 # Validate input
 if not self.validate(request):
 return {"error": "Invalid input"}

 # Execute agent
 result = self.agent.execute(request.input)

 # Validate output
 if not self.validate_output(result):
 return {"error": "Invalid output"}

 return {
 'status': 'success',
 'result': result,
 'execution_time': result.latency
 }

 except Exception as e:
 return {
 'status': 'error',
 'error': str(e)
 }

 def validate(self, request):
 """Validate input"""

 # Length check
 if len(request.input) > 10000:
 return False

 # Required fields
 if not request.task_id:
 return False

 return True

-

Database Integration

Reading/Writing Data

class DatabaseAgent:
 """Agent with database access"""

 def __init__(self, db_connection):
 self.db = db_connection

 def query_database(self, query_text):
 """Agent-written SQL queries"""

 # Agent generates SQL
 generated_sql = self.agent.generate_sql(query_text)

 # Validate SQL (safety!)
 if not self.validate_sql(generated_sql):
 return {"error": "Unsafe SQL"}

 # Execute
 try:
 results = self.db.execute(generated_sql)
 return results
 except Exception as e:
 return {"error": str(e)}

 def validate_sql(self, sql):
 """Prevent SQL injection"""

 # Check for dangerous operations
 dangerous = ['DROP', 'DELETE', 'TRUNCATE']

 for word in dangerous:
 if word in sql.upper():
 return False

 # Parse SQL to check structure
 try:
 parse_sql(sql)
 return True
 except:
 return False

-

Workflow Automation Integration

Zapier/Make Integration

class WorkflowAgent:
 """Agent as part of workflow"""

 def handle_webhook(self, webhook_data):
 """Process webhook from automation tool"""

 # Extract data
 trigger = webhook_data['trigger']
 data = webhook_data['data']

 # Process with agent
 result = self.agent.execute(data)

 # Trigger next steps
 if result.success:
 self.trigger_webhook(
 webhook_data['next_step'],
 {
 'input': data,
 'output': result.output
 }
)
 else:
 self.trigger_error_handling(result.error)

-

Slack/Teams Integration

Chat Bot Integration

class ChatBotAgent:
 """Agent in chat application"""

 def handle_slack_message(self, message):
 """Process Slack message"""

 # Parse message
 user_id = message['user']
 text = message['text']
 channel = message['channel']

 # Check permissions
 if not self.user_has_permission(user_id):
 return self.send_message(
 channel,
 "You don't have permission to use this agent"
)

 # Process
 try:
 result = self.agent.execute(text)

 # Send response
 self.send_message(channel, result.output)

 # Log for audit
 self.log_interaction(user_id, text, result.output)

 except Exception as e:
 self.send_message(channel, f"Error: {str(e)}")

---

Email Integration

Email-Triggered Agent

class EmailAgent:
 """Agent triggered by email"""

 def handle_incoming_email(self, email):
 """Process incoming email"""

 sender = email['from']
 subject = email['subject']
 body = email['body']

 # Check if sender is whitelisted
 if not self.is_whitelisted_sender(sender):
 return self.send_reply(
 sender,
 "Unknown sender"
)

 # Process with agent
 result = self.agent.execute(body)

 # Send reply
 self.send_reply(sender, result.output)

-

3 Warnings

Warning 1: No Rate Limiting

# WRONG
@app.post("/agent")
async def process(request):
 result = agent.execute(request)
 return result

# No rate limiting!
# Attacker hammers endpoint
# System overwhelmed

# RIGHT
@app.post("/agent")
@rate_limit(requests=100, period=3600)
async def process(request):
 result = agent.execute(request)
 return result

# Protected against abuse

Warning 2: No Input Validation

# WRONG
result = agent.execute(request.input)
# No validation!

# User passes dangerous input
# Agent does harmful action

# RIGHT
if not validate_input(request.input):
 return error()

result = agent.execute(request.input)

Warning 3: Leaking Sensitive Data

# WRONG
result = agent.execute(request)
return result # Full output

# Exposes database contents
# Reveals internal structure

# RIGHT
result = agent.execute(request)
sanitized = sanitize_output(result)
return sanitized # Safe output

-

Last Updated: August 9, 2026