Migrating Agent Integrations: Swapping Third-Party APIs Without Breaking Workflows

Why Agent Integrations Are Hard to Swap

AI agents interact with the world through tool calls. Each tool wraps a third-party API — a CRM, a payment processor, a search engine, a calendar service. When you need to swap Twilio for Vonage, or Stripe for Paddle, or SendGrid for Amazon SES, every agent that uses that tool is affected.

If the tool function is tightly coupled to the vendor SDK, the swap requires changing agent code, rewriting tool definitions, and re-testing every workflow that uses that tool. The adapter pattern eliminates this coupling by putting an abstraction layer between your agents and external APIs.

Step 1: Define a Vendor-Agnostic Interface

Start by defining what your agents actually need from the integration, independent of any specific vendor.

flowchart LR
    CUR(["On Current Vendor"])
    AUDIT["1. Audit current<br/>flows and data"]
    EXPORT["2. Export contacts,<br/>scripts, recordings"]
    BUILD["3. Build CallSphere<br/>agent and integrations"]
    PILOT{"4. Pilot on<br/>10 percent of traffic"}
    CUTOVER["5. Forward all<br/>numbers"]
    LIVE(["Live on<br/>CallSphere"])
    CUR --> AUDIT --> EXPORT --> BUILD --> PILOT
    PILOT -->|Pass| CUTOVER --> LIVE
    PILOT -->|Issues| BUILD
    style CUR fill:#dc2626,stroke:#b91c1c,color:#fff
    style PILOT fill:#f59e0b,stroke:#d97706,color:#1f2937
    style LIVE fill:#059669,stroke:#047857,color:#fff

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Optional

@dataclass
class EmailMessage:
    to: str
    subject: str
    body_html: str
    from_address: str
    reply_to: Optional[str] = None

@dataclass
class EmailResult:
    success: bool
    message_id: Optional[str] = None
    error: Optional[str] = None

class EmailProvider(ABC):
    """Vendor-agnostic email interface."""

    @abstractmethod
    async def send(self, message: EmailMessage) -> EmailResult:
        ...

    @abstractmethod
    async def check_delivery_status(self, message_id: str) -> str:
        ...

Step 2: Implement Adapters for Each Vendor

Each vendor gets its own adapter that implements the interface.

import httpx

class SendGridAdapter(EmailProvider):
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.sendgrid.com/v3"

    async def send(self, message: EmailMessage) -> EmailResult:
        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"{self.base_url}/mail/send",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={
                    "personalizations": [{"to": [{"email": message.to}]}],
                    "from": {"email": message.from_address},
                    "subject": message.subject,
                    "content": [{
                        "type": "text/html",
                        "value": message.body_html,
                    }],
                },
            )
            if response.status_code == 202:
                msg_id = response.headers.get("X-Message-Id", "")
                return EmailResult(success=True, message_id=msg_id)
            return EmailResult(success=False, error=response.text)

    async def check_delivery_status(self, message_id: str) -> str:
        # SendGrid status check implementation
        return "delivered"

class SESAdapter(EmailProvider):
    def __init__(self, region: str = "us-east-1"):
        import boto3
        self.client = boto3.client("ses", region_name=region)

    async def send(self, message: EmailMessage) -> EmailResult:
        try:
            import asyncio
            response = await asyncio.to_thread(
                self.client.send_email,
                Source=message.from_address,
                Destination={"ToAddresses": [message.to]},
                Message={
                    "Subject": {"Data": message.subject},
                    "Body": {"Html": {"Data": message.body_html}},
                },
            )
            return EmailResult(
                success=True,
                message_id=response["MessageId"],
            )
        except Exception as e:
            return EmailResult(success=False, error=str(e))

    async def check_delivery_status(self, message_id: str) -> str:
        return "sent"

Step 3: Wire the Adapter Into Agent Tools

The agent tool uses the interface, not the concrete implementation.

from agents import Agent, function_tool, RunContextWrapper
from dataclasses import dataclass

@dataclass
class AppContext:
    email_provider: EmailProvider
    user_id: str

@function_tool
async def send_email(
    wrapper: RunContextWrapper[AppContext],
    to: str,
    subject: str,
    body: str,
) -> str:
    """Send an email to a customer."""
    provider = wrapper.context.email_provider
    result = await provider.send(EmailMessage(
        to=to,
        subject=subject,
        body_html=body,
        from_address="support@example.com",
    ))
    if result.success:
        return f"Email sent successfully (ID: {result.message_id})"
    return f"Failed to send email: {result.error}"

agent = Agent(
    name="Support Agent",
    instructions="You help customers with support requests.",
    model="gpt-4o",
    tools=[send_email],
)

Step 4: Parallel Testing Before Cutover

Run both providers simultaneously to verify the new one works before switching.

class ParallelEmailProvider(EmailProvider):
    """Sends through both providers, returns primary result."""

    def __init__(
        self,
        primary: EmailProvider,
        shadow: EmailProvider,
    ):
        self.primary = primary
        self.shadow = shadow

    async def send(self, message: EmailMessage) -> EmailResult:
        import asyncio

        primary_result, shadow_result = await asyncio.gather(
            self.primary.send(message),
            self.shadow.send(message),
            return_exceptions=True,
        )

        # Log shadow result for comparison
        if isinstance(shadow_result, Exception):
            print(f"Shadow provider error: {shadow_result}")
        else:
            print(f"Shadow result: {shadow_result.success}")

        return primary_result  # Always return primary

    async def check_delivery_status(self, message_id: str) -> str:
        return await self.primary.check_delivery_status(message_id)

# During migration testing:
provider = ParallelEmailProvider(
    primary=SendGridAdapter(api_key="sg-key"),
    shadow=SESAdapter(region="us-east-1"),
)

Step 5: Cut Over with a Config Change

The actual cutover is a configuration change, not a code change.

import os

def get_email_provider() -> EmailProvider:
    provider_name = os.getenv("EMAIL_PROVIDER", "sendgrid")
    if provider_name == "ses":
        return SESAdapter(region=os.getenv("AWS_REGION", "us-east-1"))
    elif provider_name == "sendgrid":
        return SendGridAdapter(api_key=os.environ["SENDGRID_API_KEY"])
    else:
        raise ValueError(f"Unknown email provider: {provider_name}")

FAQ

How do I handle vendor-specific features that do not map to the common interface?

Add optional methods or metadata fields to the interface. For example, if SendGrid supports email scheduling but SES does not, add a schedule_at optional parameter to EmailMessage. The SES adapter ignores it. Document which features are vendor-specific so the team knows what will be lost during migration.

Should I use the adapter pattern for every integration?

Use it for integrations you might realistically swap: email providers, payment processors, SMS services, and search APIs. Do not over-abstract integrations that are deeply embedded and unlikely to change, like your primary database. The adapter pattern adds indirection — only add it where the flexibility pays off.

How do I test the shadow provider without sending duplicate emails?

For email specifically, use a sandbox mode or test recipient domain. SendGrid and SES both support sandbox endpoints that validate the request without delivering. Set the shadow provider to sandbox mode so you verify API compatibility without spamming users.

---

#APIMigration #AdapterPattern #Integration #Python #AgentTools #AgenticAI #LearnAI #AIEngineering