ai-workflow-orchestrator

name	AI Workflow Orchestrator
description	Expert guidance for building AI-powered workflows with n8n, Zapier, and custom orchestration systems. Use when automating workflows, integrating AI agents, or building no-code/low-code automation.
version	1.0.0
allowed-tools	["Read","Write","Edit","Bash"]

AI Workflow Orchestrator

Enterprise AI workflow automation with n8n, Zapier, and custom orchestration.

n8n Workflow Patterns

AI Agent Workflow

{
  "nodes": [
    {
      "name": "Webhook",
      "type": "n8n-nodes-base.webhook",
      "typeVersion": 1,
      "position": [250, 300],
      "webhookId": "user-query"
    },
    {
      "name": "OpenAI Chat",
      "type": "@n8n/n8n-nodes-langchain.lmChatOpenAi",
      "typeVersion": 1,
      "position": [450, 300],
      "parameters": {
        "model": "gpt-4-turbo",
        "temperature": 0.7,
        "systemMessage": "You are a helpful assistant."
      }
    },
    {
      "name": "Vector Store Query",
      "type": "@n8n/n8n-nodes-langchain.vectorStorePinecone",
      "typeVersion": 1,
      "position": [450, 150],
      "parameters": {
        "operation": "retrieve",
        "topK": 5
      }
    },
    {
      "name": "Response Formatter",
      "type": "n8n-nodes-base.function",
      "typeVersion": 1,
      "position": [650, 300],
      "parameters": {
        "functionCode": "return items.map(item => ({\n  json: {\n    response: item.json.response,\n    sources: item.json.sources,\n    confidence: item.json.confidence\n  }\n}));"
      }
    }
  ],
  "connections": {
    "Webhook": {
      "main": [[{ "node": "Vector Store Query", "type": "main", "index": 0 }]]
    },
    "Vector Store Query": {
      "main": [[{ "node": "OpenAI Chat", "type": "main", "index": 0 }]]
    },
    "OpenAI Chat": {
      "main": [[{ "node": "Response Formatter", "type": "main", "index": 0 }]]
    }
  }
}

Multi-Agent Orchestration

// n8n Custom Node for Agent Orchestration
import { IExecuteFunctions } from 'n8n-core';
import { INodeExecutionData, INodeType, INodeTypeDescription } from 'n8n-workflow';

export class MultiAgentOrchestrator implements INodeType {
  description: INodeTypeDescription = {
    displayName: 'Multi-Agent Orchestrator',
    name: 'multiAgentOrchestrator',
    group: ['transform'],
    version: 1,
    description: 'Orchestrate multiple AI agents',
    defaults: {
      name: 'Multi-Agent Orchestrator',
    },
    inputs: ['main'],
    outputs: ['main'],
    properties: [
      {
        displayName: 'Agents',
        name: 'agents',
        type: 'fixedCollection',
        typeOptions: {
          multipleValues: true,
        },
        default: {},
        options: [
          {
            name: 'agentValues',
            : ,
            : [
              {
                : ,
                : ,
                : ,
                : ,
              },
              {
                : ,
                : ,
                : ,
                : [
                  { : , :  },
                  { : , :  },
                  { : , :  },
                ],
                : ,
              },
            ],
          },
        ],
      },
    ],
  };

   (: ): <[][]> {
     items = .();
     : [] = [];

     ( itemIndex = ; itemIndex < items.; itemIndex++) {
       input = .(, itemIndex, )  ;
       agents = .(, itemIndex, {})  ;

       : [] = [];

       ( agent  agents. || []) {
         result =  .(agent., input);
        results.({ : agent., result });
      }

      returnData.({
        : {
          input,
          results,
          : .(results),
        },
      });
    }

     [returnData];
  }

    (: , : ): <> {
    
     ;
  }

   (: []):  {
     results.( r.).();
  }
}

Zapier Integration Patterns

AI-Powered Email Automation

// Zapier Custom Code Action
const inputData = inputData || {};
const { email_content, sender } = inputData;

// Call OpenAI API
const response = await fetch('https://api.openai.com/v1/chat/completions', {
  method: 'POST',
  headers: {
    'Authorization': `Bearer ${process.env.OPENAI_API_KEY}`,
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({
    model: 'gpt-4-turbo',
    messages: [
      {
        role: 'system',
        content: 'You are an email assistant. Categorize and draft responses.',
      },
      {
        role: 'user',
        content: `Categorize and draft a response to this email:\n\n${email_content}`,
      },
    ],
  }),
});

const data = await response.json();
const ai_response = data.choices[0].message.content;

// Parse AI response
const category = ai_response.match(/Category: (.*)/)?.[1] || 'General';
const draft_response = ai_response.match()?.[]?.() || ;

output = {
  category,
  draft_response,
  : category ===  ?  : ,
};

Custom Orchestration System

Workflow Engine

from typing import Dict, List, Callable, Any
from pydantic import BaseModel
import asyncio

class WorkflowStep(BaseModel):
    name: str
    function: str
    inputs: Dict[str, str] = {}
    condition: str | None = None

class Workflow(BaseModel):
    name: str
    steps: List[WorkflowStep]

class WorkflowEngine:
    def __init__(self):
        self.functions: Dict[str, Callable] = {}
        self.context: Dict[str, Any] = {}

    def register_function(self, name: str, func: Callable):
        """Register a function that can be used in workflows."""
        self.functions[name] = func

    async def execute_workflow(self, workflow: Workflow, initial_context: Dict[str, ]):
        
        .context = initial_context.copy()

         step  workflow.steps:
            
             step.condition   ._evaluate_condition(step.condition):
                

            
            inputs = {
                key: ._resolve_value(value)
                 key, value  step.inputs.items()
            }

            
            func = .functions.get(step.function)
              func:
                 ValueError()

            result =  func(**inputs)  asyncio.iscoroutinefunction(func)  func(**inputs)

            
            .context[step.name] = result

         .context

     () -> :
        
         value.startswith():
             .context.get(value[:])
         value

     () -> :
        
         (condition, {}, .context)


engine = WorkflowEngine()


  () -> :
    
     

  () -> :
    
     

engine.register_function(, analyze_sentiment)
engine.register_function(, generate_response)


workflow = Workflow(
    name=,
    steps=[
        WorkflowStep(
            name=,
            function=,
            inputs={: }
        ),
        WorkflowStep(
            name=,
            function=,
            inputs={: , : },
            condition=
        ),
    ]
)


result =  engine.execute_workflow(
    workflow,
    {: }
)

Agent Chain Pattern

interface Agent {
  name: string;
  execute: (input: any) => Promise<any>;
}

class AgentChain {
  private agents: Agent[] = [];

  add(agent: Agent): this {
    this.agents.push(agent);
    return this;
  }

  async execute(initialInput: any): Promise<any> {
    let result = initialInput;

    for (const agent of this.agents) {
      console.log(`Executing ${agent.name}...`);
      result = await agent.execute(result);
    }

    return result;
  }
}

// Usage
const chain = new AgentChain()
  .add({
    name: 'Researcher',
    execute:  (: ) => {
      
       { query, : [, ] };
    },
  })
  .({
    : ,
    :  (: ) => {
      
       { ...data, :  };
    },
  })
  .({
    : ,
    :  (: ) => {
      
       {
        : data.,
        : data.,
        : data.,
      };
    },
  });

 result =  chain.();

Integration Patterns

Webhook Handlers

from fastapi import FastAPI, Request
from pydantic import BaseModel

app = FastAPI()

class ZapierWebhook(BaseModel):
    event: str
    data: dict

@app.post("/zapier/webhook")
async def handle_zapier_webhook(webhook: ZapierWebhook):
    """Handle incoming Zapier webhooks."""
    if webhook.event == "new_lead":
        await process_lead(webhook.data)
    elif webhook.event == "support_ticket":
        await process_ticket(webhook.data)

    return {"status": "processed"}

@app.post("/n8n/webhook/{workflow_id}")
async def handle_n8n_webhook(workflow_id: str, request: Request):
    """Handle incoming n8n webhooks."""
    data = await request.json()
    result = await execute_workflow(workflow_id, data)
    return result

Task Queues

from celery import Celery
from typing import Dict

celery = Celery('tasks', broker='redis://localhost:6379')

@celery.task
def process_with_ai(data: Dict) -> Dict:
    """Process data with AI agent in background."""
    result = ai_agent.process(data)
    notify_completion(result)
    return result

# Trigger from workflow
process_with_ai.delay({"text": "Process this"})

Best Practices

✅ Use idempotent operations ✅ Implement error handling and retries ✅ Add logging and monitoring ✅ Use task queues for long-running operations ✅ Implement rate limiting ✅ Version your workflows ✅ Test workflows thoroughly ✅ Use environment variables for secrets ✅ Implement rollback mechanisms ✅ Monitor workflow performance

---

When to Use: AI workflow automation, n8n/Zapier integrations, multi-agent orchestration, business process automation.