AI Agent Integration Patterns: Connecting AI to Everything

The patterns for integrating AI agents with your existing systems and tools.

AI agents need to connect to databases, APIs, file systems, and other services. The integration pattern you choose affects reliability, security, and maintainability. This guide covers every major pattern.

Integration Pattern Catalog

1. Direct API Integration

Agent → API Call → External Service

class DirectIntegration:
    def __init__(self):
        self.tools = {
            "get_customer": self.get_customer,
            "create_order": self.create_order,
        }

    async def get_customer(self, customer_id: str):
        response = await httpx.get(f"https://api.acme.com/customers/{customer_id}")
        return response.json()

Pros: Simple, direct control Cons: Tight coupling, auth complexity

2. MCP Server Integration

Agent → MCP Protocol → MCP Server → External Service

# MCP server wraps external API
server = McpServer({"name": "crm-integration"})

@server.tool("get_customer")
async def get_customer(customer_id: str):
    customer = await crm_api.get_customer(customer_id)
    return {"content": [{"type": "text", "text": json.dumps(customer)}]}

# Agent uses via standard MCP discovery

Pros: Universal compatibility, discovery, standardized Cons: Extra layer, protocol overhead

3. Webhook Integration

External Service → Webhook → Agent → Process

@app.post("/webhook/github")
async def github_webhook(event):
    if event["action"] == "opened":
        # Agent processes new PR
        await code_review_agent.process(event["pull_request"])

Pros: Real-time, event-driven Cons: Need public endpoint, security concerns

4. Message Queue Integration

Agent → Queue → Worker → External Service
                    ↓
Agent ← Result ← Worker

class QueueIntegration:
    async def execute_async(self, task):
        # Publish to queue
        job_id = await self.queue.publish({
            "task": task.type,
            "params": task.params,
        })
        return job_id

    async def check_result(self, job_id):
        return await self.queue.get_result(job_id)

Pros: Async, scalable, resilient Cons: Added complexity, eventual consistency

5. Database Integration

Agent → SQL → Database

@server.tool("query_sales")
async def query_sales(start_date: str, end_date: str):
    results = await db.execute("""
        SELECT product, SUM(quantity), SUM(revenue)
        FROM sales
        WHERE date BETWEEN %s AND %s
        GROUP BY product
    """, start_date, end_date)
    return results

6. File System Integration

Agent → File Operations → Local/Cloud Storage

@server.tool("read_document")
async def read_document(path: str):
    content = await s3.read_file(path)
    return {"content": content, "mime_type": detect_mime(path)}

7. SaaS Integration (via Connectors)

Agent → Connector → SaaS API (Salesforce, Slack, etc.)

class SalesforceConnector:
    def __init__(self):
        self.api = SalesforceAPI(
            client_id=env("SF_CLIENT_ID"),
            client_secret=env("SF_CLIENT_SECRET"),
        )

    @tool("create_lead")
    async def create_lead(self, name, email, company):
        lead = await self.api.create_lead({
            "Name": name,
            "Email": email,
            "Company": company,
        })
        return {"id": lead.id, "status": "created"}

    @tool("update_opportunity")
    async def update_opportunity(self, opp_id, stage, amount):
        return await self.api.update(opp_id, {"StageName": stage, "Amount": amount})

Choosing the Right Pattern

Common Integration Challenges

Authentication Management

class AuthManager:
    """Manage auth for multiple integrations."""
    def __init__(self):
        self.providers = {
            "salesforce": SalesforceAuth(),
            "slack": SlackAuth(),
            "github": GitHubAuth(),
            "internal": InternalAuth(),
        }

    async def get_auth_header(self, service):
        provider = self.providers[service]
        token = await provider.get_valid_token()  # Handles refresh
        return {"Authorization": f"Bearer {token}"}

Error Handling Across Services

class ResilientIntegration:
    async def call_with_retry(self, service, operation, params, max_retries=3):
        for attempt in range(max_retries):
            try:
                return await self.services[service][operation](**params)
            except AuthError:
                await self.refresh_auth(service)
            except RateLimitError as e:
                await asyncio.sleep(e.retry_after)
            except ServiceUnavailableError:
                await asyncio.sleep(2 ** attempt)
            except ValidationError as e:
                return {"error": str(e), "recoverable": False}

        return {"error": "Max retries exceeded", "recoverable": False}

Integration as MCP Skills

Publish your integrations as MCP skills for reuse:

# Build a reusable integration skill
@mcp_tool(name="salesforce-tools", pricing={"per_use": 0.50})
class SalesforceTools:
    @tool("create_lead")
    async def create_lead(self, name, email, company):
        ...

    @tool("search_contacts")
    async def search_contacts(self, query):
        ...

    @tool("update_opportunity")
    async def update_opportunity(self, opp_id, data):
        ...

# Publish on SkillExchange
# Any agent can now use your Salesforce integration

Monitoring Integrations

class IntegrationMonitor:
    async def health_check_all(self):
        results = {}
        for name, integration in self.integrations.items():
            try:
                start = time.time()
                await integration.ping()
                latency = (time.time() - start) * 1000

                results[name] = {
                    "status": "healthy",
                    "latency_ms": latency,
                }
            except Exception as e:
                results[name] = {
                    "status": "unhealthy",
                    "error": str(e),
                }

        return results

Conclusion

Choosing the right integration pattern depends on your specific needs: MCP for universal compatibility, webhooks for real-time, queues for async processing, and direct API for simplicity. Most production systems use a combination of patterns.

Learn More

• MCP Protocol Explained
• Building AI Agents with TypeScript
• AI Agent Deployment Best Practices
• Skill-Based AI Architecture

Find integration tools on SkillExchange.