Integration Patterns: Connecting to Systems

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