Chapter 12: The ITI Workflow Adapter

Last Updated: 2026-04-16

12.2 Architecture

Product (PHP / Swift / Python / Rust / TypeScript)
         │
         ▼
  ITI_Workflow_Adapter
         │
         ├── Try: POST to n8n webhook
         │         │
         │         ├── Success → return n8n result
         │         │
         │         └── Failure (timeout / 5xx / network) →
         │
         └── Fallback: POST to Anthropic Claude API directly
                   │
                   └── Return Claude result

The fallback ensures that if n8n is down for maintenance or an upgrade, products can still respond to users (with potentially reduced functionality — no RAG, no multi-step pipelines).

---

12.3 Webhook Endpoint Registry

Each product registers its webhook paths in the adapter configuration. The adapter uses these to route requests to the correct n8n workflow. The full registry of active webhook paths is documented in Chapter 9, §9.6. Representative examples:

Path	Product
/webhook/career-coach	career-coach
/webhook/travelplanner	my-travelplanner
/webhook/scuba-gpt	scuba-gpt
/webhook/executive-advisor	executive-advisor
/webhook/consulting-pipeline	ITI Consulting Pipeline
/webhook/personal-advisor	Personal Advisor Router

When adding a new product, add its webhook path to the registry and add a corresponding test in test_n8n_webhooks.py. See Chapter 9, §9.6 for the complete list of 24 registered endpoints.

---

12.4 PHP Implementation (WordPress)

The PHP adapter is used in all WordPress plugin products. It wraps wp_remote_post() for n8n calls and the ITI Claude API client for fallback.

Basic usage

// Include the adapter
require_once ITI_SHARED_PATH . '/wordpress/api-clients/class-iti-workflow-adapter.php';

// Instantiate
$adapter = new ITI_Workflow_Adapter([
    'webhook_url' => ITI_N8N_BASE_URL . '/webhook/iti-career-coach-session',
    'timeout'     => 30,                   // seconds
    'fallback'    => true,                 // enable Claude fallback
]);

// Call the workflow
$response = $adapter->request([
    'action'       => 'generate_plan',
    'user_message' => sanitize_text_field($_POST['message']),
    'context'      => $user_context,
]);

if (is_wp_error($response)) {
    // Handle error
    return;
}

$result = $response['content'];

Error handling pattern

$response = $adapter->request($payload);

if (is_wp_error($response)) {
    error_log('ITI Adapter error: ' . $response->get_error_message());
    wp_send_json_error(['message' => 'Service temporarily unavailable.']);
    return;
}

if (isset($response['error'])) {
    // n8n returned an error response
    wp_send_json_error(['message' => $response['error']]);
    return;
}

wp_send_json_success($response);

---

12.5 Python Implementation

Used in Flask and FastAPI services (e.g., Patriot Agent).

import httpx
import anthropic
from typing import Optional

class ITIWorkflowAdapter:
    def __init__(
        self,
        webhook_url: str,
        timeout: float = 30.0,
        fallback: bool = True,
    ):
        self.webhook_url = webhook_url
        self.timeout = timeout
        self.fallback = fallback
        self._claude = anthropic.Anthropic()  # uses ANTHROPIC_API_KEY env var

    async def request(self, payload: dict) -> dict:
        # Try n8n first
        try:
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                response = await client.post(self.webhook_url, json=payload)
                response.raise_for_status()
                return response.json()
        except (httpx.TimeoutException, httpx.HTTPStatusError) as e:
            if not self.fallback:
                raise
            # Fallback to Claude
            return self._claude_fallback(payload, str(e))

    def _claude_fallback(self, payload: dict, error_reason: str) -> dict:
        message = self._claude.messages.create(
            model="claude-opus-4-5",
            max_tokens=2048,
            system=payload.get("system_prompt", "You are a helpful assistant."),
            messages=[{"role": "user", "content": payload.get("user_message", "")}]
        )
        return {
            "content": message.content[0].text,
            "source": "claude_fallback",
            "fallback_reason": error_reason,
        }

---

12.6 Swift Implementation (iOS/macOS)

Used in Swift apps (expat-advisor, personal-assistant).

import Foundation

class ITIWorkflowAdapter {
    private let webhookURL: URL
    private let timeout: TimeInterval
    private let fallbackEnabled: Bool

    init(webhookURL: URL, timeout: TimeInterval = 30, fallback: Bool = true) {
        self.webhookURL = webhookURL
        self.timeout = timeout
        self.fallbackEnabled = fallback
    }

    func request(payload: [String: Any]) async throws -> [String: Any] {
        do {
            return try await callN8N(payload: payload)
        } catch {
            guard fallbackEnabled else { throw error }
            return try await callClaude(payload: payload)
        }
    }

    private func callN8N(payload: [String: Any]) async throws -> [String: Any] {
        var request = URLRequest(url: webhookURL)
        request.httpMethod = "POST"
        request.setValue("application/json", forHTTPHeaderField: "Content-Type")
        request.timeoutInterval = timeout
        request.httpBody = try JSONSerialization.data(withJSONObject: payload)

        let (data, response) = try await URLSession.shared.data(for: request)
        guard let http = response as? HTTPURLResponse, http.statusCode < 500 else {
            throw URLError(.badServerResponse)
        }
        return try JSONSerialization.jsonObject(with: data) as! [String: Any]
    }

    private func callClaude(payload: [String: Any]) async throws -> [String: Any] {
        // Use ClaudeService from ITI shared library
        return try await ClaudeService.shared.complete(
            system: payload["system_prompt"] as? String ?? "",
            user: payload["user_message"] as? String ?? ""
        )
    }
}

---

12.7 Rust Implementation (Tauri)

Used in Tauri desktop apps (estate-manager, personal-assistant desktop).

use reqwest::Client;
use serde_json::Value;
use std::time::Duration;

pub struct ITIWorkflowAdapter {
    webhook_url: String,
    timeout: Duration,
    fallback: bool,
    client: Client,
}

impl ITIWorkflowAdapter {
    pub fn new(webhook_url: &str, timeout_secs: u64, fallback: bool) -> Self {
        Self {
            webhook_url: webhook_url.to_string(),
            timeout: Duration::from_secs(timeout_secs),
            fallback,
            client: Client::new(),
        }
    }

    pub async fn request(&self, payload: &Value) -> Result<Value, String> {
        match self.call_n8n(payload).await {
            Ok(response) => Ok(response),
            Err(e) if self.fallback => self.call_claude_fallback(payload, &e).await,
            Err(e) => Err(e),
        }
    }

    async fn call_n8n(&self, payload: &Value) -> Result<Value, String> {
        self.client
            .post(&self.webhook_url)
            .timeout(self.timeout)
            .json(payload)
            .send()
            .await
            .map_err(|e| e.to_string())?
            .json::<Value>()
            .await
            .map_err(|e| e.to_string())
    }

    async fn call_claude_fallback(&self, payload: &Value, reason: &str) -> Result<Value, String> {
        // Use ITI claude_client from shared Rust library
        // ...
        Ok(serde_json::json!({"source": "claude_fallback", "fallback_reason": reason}))
    }
}

---

12.8 Adding a New Product Webhook

When a new product needs to integrate with the adapter:

1. Create the n8n workflow using the SDK lifecycle (Chapter 9).
2. Add the webhook path to the WEBHOOK_ENDPOINTS constant/registry in the adapter for that platform.
3. Add a webhook test to ITI/infrastructure/n8n-dify/tests/test_n8n_webhooks.py:

@pytest.mark.smoke
def test_new_product_webhook(n8n_base_url):
    response = requests.post(
        f"{n8n_base_url}/webhook/iti-new-product-action",
        json={"test": True, "user_message": "Hello"},
        timeout=30,
    )
    assert response.status_code == 200
    data = response.json()
    assert "content" in data

1. Update the migration appendix in ITI/operations/documentation/MIGRATION-GUIDE.md to document the new endpoint.
2. Test the fallback by temporarily stopping n8n:

docker compose stop iti-n8n
# Run the product's integration test — it should succeed via Claude fallback
docker compose start iti-n8n

---

Previous: Chapter 11 — Dify Knowledge Bases & RAG | Next: Chapter 13 — The ITI Shared Library