Claude Agent Workflow: Migrating Data from One SaaS App to Another

When you hand‑code a one‑off CSV‑to‑SQL migration, a few dozen lines get the job done—until the source SaaS changes its field names, rate‑limits start throwing 429s, or a single record blows up the ETL logic. The whole script crashes, you lose progress, and the only thing left to do is stare at logs and rewind time.

What if you could offload the “think‑about‑the‑schema” part to an LLM that watches the data, writes the transformation code, and retries when the API hiccups—while you stay in full control from a desktop UI? The following guide shows how to build exactly that: a production‑ready Claude “computer use” agent wrapped in a PyQt6 GUI, complete with persistent queues, streaming output, and hardened error handling.

Claude Agent workflow for SaaS migration uses Anthropic’s API within a desktop GUI (like PyQt) to automate data transfer. You build tools for API calls and data parsing, then prompt Claude to orchestrate the Extract, Transform, Validate, Load (ETVL) process. The key is a robust wrapper that handles errors, rate limits, and state, turning the LLM into a reasoning engine inside a reliable, user‑controlled application.

The Problem: Why Generic Migration Scripts Fail

Pitfalls of Simple API‑to‑API Scripts

A naïve script typically loops over source records, builds a payload, POSTs to the destination, and logs success. It assumes a static schema, unlimited request budget, and that every record can be processed in a single pass. Real SaaS migrations break that illusion:

• **Schema drift** – fields are added/renamed mid‑migration, causing `KeyError`s.
• **Rate limiting** – 429 responses stall the entire run; exponential backoff is an after‑thought.
• **Partial failures** – a single malformed row forces a full rollback unless you build idempotent writes.

Handling Mapping Logic, Schema Drift & Rate Limiting

You need a *stateful* orchestrator that can:

1. Detect schema changes on the fly.
2. Re‑map fields using a learned mapping table.
3. Queue failed rows for manual review without aborting the whole batch.

Enter Claude: let the LLM generate the mapping logic dynamically, while your deterministic wrapper enforces safety.

Architecture: Designing a Robust Migration Agent

GUI Layer (PyQt6) for Input & Monitoring

• **Form widgets** – API keys, source/destination URLs, batch size.
• **Log console** – a read‑only `QPlainTextEdit` that receives streamed tokens.
• **Control panel** – Start, Pause, Cancel, and a progress bar.

Agent Core (Anthropic SDK, `claude.computer_use`)

• **System prompt** – defines tools: `http_get`, `http_post`, `parse_csv`, `build_sql`.
• **Tool orchestrator** – Claude decides which tool to invoke, returns code, and the sandbox executes it.

Data & State Management

• **SQLite + SQLAlchemy** – persistent tables for `records_queue`, `processed`, `failed`.
• **AsyncIO queue** – feeds tasks to the agent without blocking the UI thread.
• **Structured logging** – JSON lines for audit and replay.

graph TD
    GUI[GUI (PyQt6)] -->|user actions| Controller[Controller (asyncio)]
    Controller -->|push task| Queue[Async Task Queue]
    Queue -->|pop| Agent[Claude Agent Core]
    Agent -->|invoke| Tools[Tool Sandbox]
    Tools -->|result| DB[SQLite State DB]
    DB -->|updates| GUI

Step 1: Scaffolding the Desktop GUI & Event Loop

Choosing PyQt vs. Tkinter vs. Tauri (2026 Trade‑offs)

• **PyQt6** – native widgets, solid threading model, excellent for heavy‑duty apps.
• **Tkinter** – built‑in but limited styling; okay for quick prototypes.
• **Tauri** – Rust + WebView; great for web‑centric UI but adds a compile step.

For a migration tool that needs native file dialogs and fine‑grained threading, **PyQt6** wins.

Building Input Forms for API Keys & Source/Destination

# main_gui.py – Python 3.11+, PyQt6 6.6+
# ---------------------------------------------------------
# Required packages: PyQt6==6.6.0
# ---------------------------------------------------------
from PyQt6.QtWidgets import (
    QApplication, QWidget, QFormLayout, QLineEdit,
    QPushButton, QProgressBar, QPlainTextEdit, QMessageBox
)
import sys, asyncio

class MigrationWindow(QWidget):
    def __init__(self):
        super().__init__()
        self.setWindowTitle("Claude Migration Agent")
        layout = QFormLayout(self)

        self.api_key_edit = QLineEdit()
        self.src_url_edit = QLineEdit()
        self.dst_url_edit = QLineEdit()
        self.batch_edit   = QLineEdit("100")   # default batch size

        layout.addRow("Claude API Key:", self.api_key_edit)
        layout.addRow("Source API:", self.src_url_edit)
        layout.addRow("Destination API:", self.dst_url_edit)
        layout.addRow("Batch size:", self.batch_edit)

        self.start_btn = QPushButton("Start Migration")
        self.pause_btn = QPushButton("Pause")
        self.cancel_btn = QPushButton("Cancel")
        self.progress = QProgressBar()
        self.log_view = QPlainTextEdit(readOnly=True)

        layout.addRow(self.start_btn, self.pause_btn)
        layout.addRow(self.cancel_btn)
        layout.addRow("Progress:", self.progress)
        layout.addRow("Log:", self.log_view)

        self.start_btn.clicked.connect(self.start_migration)
        self.pause_btn.clicked.connect(self.pause_migration)
        self.cancel_btn.clicked.connect(self.cancel_migration)

        self.loop = asyncio.get_event_loop()
        self.task = None

Creating a Non‑Blocking Console/Log Widget for Agent Output

    def append_log(self, text: str):
        """Thread‑safe UI append."""
        def _update():
            self.log_view.appendPlainText(text)
        QApplication.instance().postEvent(
            self.log_view, QEvent(QEvent.Type.User), _update()
        )

    async def run_agent(self):
        from agent import ClaudeMigrationAgent
        agent = ClaudeMigrationAgent(
            api_key=self.api_key_edit.text(),
            src=self.src_url_edit.text(),
            dst=self.dst_url_edit.text(),
            batch=int(self.batch_edit.text()),
            logger=self.append_log
        )
        await agent.run()

**My take:** Keep the UI thin; delegate everything else to `asyncio` tasks. The GUI thread only paints and forwards events, preventing the dreaded “Application has stopped responding” dialog.

Step 2: Implementing the Claude Agent Core with Computer Use

Configuring the Anthropic Python SDK (v0.16+)

# Install the SDK and required extras
pip install "anthropic[computer_use]"==0.16.1

# agent.py
# ---------------------------------------------------------
# Required packages: anthropic==0.16.1, tenacity==8.2.0
# ---------------------------------------------------------
import asyncio, json, os
from anthropic import Anthropic, AsyncClient
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type

class ClaudeMigrationAgent:
    def __init__(self, api_key: str, src: str, dst: str, batch: int, logger):
        self.client = AsyncClient(api_key=api_key)
        self.src = src
        self.dst = dst
        self.batch = batch
        self.logger = logger
        self.state_db = self._init_state()

Structuring the System Prompt for Tool Orchestration

    def system_prompt(self) -> str:
        return """
You are a data‑migration specialist. Use the following tools only:

- `http_get(url) -> JSON` fetches a page of source records.
- `http_post(url, payload) -> JSON` sends transformed rows.
- `parse_csv(text) -> pandas.DataFrame` parses raw CSV strings.
- `build_sql(df) -> str` returns a parametrized INSERT statement.

Never fabricate URLs or credentials. Always return a short JSON summary of what you did.
If a transformation fails, output a JSON with "error" field and let the wrapper retry.
"""

Defining Tools: GET/POST requests, CSV/JSON Parsing, SQL INSERT Builder

    async def http_get(self, url: str) -> dict:
        @retry(
            retry=retry_if_exception_type(Exception),
            wait=wait_exponential(multiplier=1, min=2, max=30),
            stop=stop_after_attempt(5)
        )
        async def _call():
            async with self.client.messages.create(
                model="claude-3-5-sonnet-202406",
                max_tokens=1024,
                system=self.system_prompt(),
                messages=[{"role": "user", "content": f"GET {url}"}],
                tools=[...]  # omitted for brevity
            ) as stream:
                # Stream tokens to logger
                async for event in stream:
                    if event.type == "content_block_delta":
                        self.logger(event.delta)
            return json.loads(event.content)  # simplified
        return await _call()

**Tip:** The `tenacity` decorator lives **outside** the async function body so the retry loop respects the coroutine’s event loop.

Step 3: Engineering the Critical Workflow Loop

State Machine: Extract → Transform → Validate → Load (ETVL)

    async def run(self):
        from models import RecordQueue, Processed, Failed
        async for batch in self._fetch_batches():
            for record in batch:
                await self._process_record(record)

    async def _process_record(self, record: dict):
        # 1️⃣ Extract – already in `record`
        try:
            transformed = await self._transform(record)
            self._validate(transformed)
            await self._load(transformed)
            await Processed.create(record_id=record["id"])
            self.logger(f"✅ Record {record['id']} migrated")
        except Exception as exc:
            await Failed.create(record_id=record["id"], error=str(exc))
            self.logger(f"⚠️ Record {record['id']} failed: {exc}")

Queue Management & Retry Logic for Failed Records

    @retry(
        wait=wait_exponential(multiplier=2, min=5, max=60),
        stop=stop_after_attempt(3),
        reraise=True
    )
    async def _load(self, payload: dict):
        response = await self.http_post(self.dst, payload)
        if response.get("status") != "ok":
            raise RuntimeError(f"Destination rejected: {response}")

Streaming Token Responses into the GUI Without Thread Freeze

    async def _transform(self, record: dict) -> dict:
        # Prompt Claude to generate pandas code
        prompt = f"""
You have a DataFrame `df` containing a single row:
{json.dumps(record, indent=2)}
Write a pandas transformation that:
1. Renames fields to match the destination schema.
2. Casts types (e.g., strings to dates).
Return only the transformed row as a JSON string.
"""
        async for chunk in self.client.messages.stream(
            model="claude-3-5-sonnet-202406",
            max_tokens=1024,
            system=self.system_prompt(),
            messages=[{"role": "user", "content": prompt}]
        ):
            if chunk.type == "content_block_delta":
                self.logger(chunk.delta)   # goes to GUI log instantly
        # Assume the final message contains the transformed JSON
        transformed = json.loads(chunk.content)
        return transformed

**My take:** The streaming loop above feeds every token straight to the UI. If you wrap the whole call in `await asyncio.to_thread(…)` you’d lose that granularity. Keeping it in the async context preserves responsiveness.

Step 4: Solving Real‑World Production Gotchas

API Rate Limit Handling & Exponential Backoff

Tenacity’s `wait_exponential` already spreads retries. Pair it with a shared `RateLimiter` object (leaky bucket) to stay under SaaS quotas.

class RateLimiter:
    def __init__(self, rate_per_sec: float):
        self.interval = 1.0 / rate_per_sec
        self._last = asyncio.get_event_loop().time()

    async def throttle(self):
        now = asyncio.get_event_loop().time()
        wait = self._last + self.interval - now
        if wait > 0:
            await asyncio.sleep(wait)
        self._last = asyncio.get_event_loop().time()

Inject `await self.limiter.throttle()` before every `http_get`/`http_post`.

Schema Validation & Handling Missing/Extra Fields

Use **Pydantic** models to assert incoming record shape. If validation fails, push the record into a “schema‑mismatch” queue for manual mapping.

from pydantic import BaseModel, ValidationError

class SourceRecord(BaseModel):
    id: int
    name: str
    created_at: str   # ISO‑8601

def validate_source(raw: dict) -> SourceRecord:
    try:
        return SourceRecord(**raw)
    except ValidationError as ve:
        raise RuntimeError(f"Schema error: {ve}") from ve

Timeout Recovery & Partial Migration Resumption

Wrap each network call with `asyncio.wait_for(…, timeout=30)`. On timeout, mark the record as `failed` and continue; the state DB lets you resume later with a simple `SELECT * FROM failed`.

Secure Credential Management (Never Hardcode!)

Store API keys in OS keyrings (`keyring` library) or encrypted `.env` files. Never write them to logs.

import keyring
api_key = keyring.get_password("anthropic", "claude_api_key")

Step 5: Testing, Monitoring & Deployment

Simulating API Errors with Mock Servers

Spin up a FastAPI mock that randomly returns 429 or malformed JSON. Use `pytest-asyncio` to assert that the retry logic caps at the configured attempts.

pip install fastapi uvicorn pytest-asyncio

Logging Every Agent Decision for Audit Trails

Create a JSON‑lines logger:

import json, datetime

def audit_log(event: dict):
    line = json.dumps({**event, "ts": datetime.datetime.utcnow().isoformat()})
    with open("migration_audit.log", "a", encoding="utf-8") as f:
        f.write(line + "\n")

Call `audit_log({“type”: “tool_call”, “tool”: “http_post”, “payload”: payload})` everywhere.

Packaging as a Standalone Executable (PyInstaller/Briefcase)

pip install pyinstaller
pyinstaller --onefile --name ClaudeMigrate main_gui.py

The resulting binary can be shipped to non‑technical ops teams. Test on Windows 11, macOS 14, and Ubuntu 24.04 to verify that the bundled `anthropic` wheels load correctly.

From Prototype to Production: Advanced Considerations

Scaling to Large Datasets with Chunking & Batches

Instead of loading all 10k+ rows into memory, fetch in paginated batches (`?page=…&size=500`). Each batch is persisted to the queue before processing, letting you resume from the last successful page.

Supporting Additional SaaS Apps via Plugin Architecture

Define an abstract `ToolPlugin` class that each SaaS integration implements (`fetch`, `push`, `map_schema`). Load plugins from a `plugins/` directory using `importlib`. This turns your migration agent into a framework rather than a single‑purpose script.

Performance Benchmarks: Agent vs. Hand‑Coded Migration

In our internal test, moving 5 000 contacts from **HubSpot** to **Salesforce**:

Implementation	Avg. time per record	Total runtime	Developer time