Building Efficient AI Agents — Part 3: Code Reference & Agentic Workflows

Working code — minimal ReAct agents in Python and Node.js, multi-agent composition, and the workflow patterns that connect agents to enterprise platforms.

This is the third installment of a four-part series:

1. Part 1: The Mental Model & Core Architecture
2. Part 2: Technology Stack, Deployment Patterns & Cost
3. Part 3: Code Reference & Agentic Workflows
4. Part 4: Security, Evaluation & Production Operations

Code samples are collapsed by default — expand the ones you need. They print expanded.

A complete, runnable reference implementation — the entire pattern, self-contained:

# requirements: openai litellm tavily-python pydantic python-dotenv
# pip install openai litellm tavily-python pydantic python-dotenv

import json
import asyncio
from dataclasses import dataclass, field
from typing import Any, Optional

from pydantic import BaseModel
from litellm import acompletion
from tavily import TavilyClient
import os
from dotenv import load_dotenv

load_dotenv()

# --- Tool Definitions ---

tavily = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))

def search_web(query: str, max_results: int = 3) -> str:
    """Search the web for current information."""
    try:
        results = tavily.search(query, max_results=max_results)
        return "\n".join(
            f"[{r['title']}] {r['content']}"
            for r in results.get("results", [])
        )
    except Exception as e:
        return f"Search error: {e}"

def calculator(expression: str) -> str:
    """Evaluate a safe mathematical expression."""
    try:
        # Restrict to safe math operations only
        allowed = set("0123456789+-*/()., ")
        if not all(c in allowed for c in expression):
            return "Error: unsafe expression"
        return str(eval(expression))
    except Exception as e:
        return f"Calc error: {e}"

# --- Tool Registry ---

TOOLS = {
    "search_web": search_web,
    "calculator": calculator,
}

TOOL_DEFINITIONS = [
    {
        "type": "function",
        "function": {
            "name": "search_web",
            "description": "Search the web for current information. Use for facts, news, or data you don't know.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Search query"},
                    "max_results": {"type": "integer", "description": "Number of results (default 3)"}
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "Evaluate mathematical expressions. Use for any numeric calculation.",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {"type": "string", "description": "Math expression e.g. '356500 / 21.1'"}
                },
                "required": ["expression"]
            }
        }
    }
]

# --- ExecutionContext ---

@dataclass
class ExecutionContext:
    messages: list[dict] = field(default_factory=list)
    step: int = 0
    max_steps: int = 10

# --- Agent Loop ---

async def run_agent(
    user_input: str,
    system_prompt: str = "You are a helpful research assistant. Use tools when you need current information or calculations.",
    model: str = "gpt-4o",
    max_steps: int = 10
) -> str:
    """
    Core ReAct agent loop.
    Think -> Act (tool call) -> Observe -> Repeat until final answer.
    """
    ctx = ExecutionContext(max_steps=max_steps)
    ctx.messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_input}
    ]

    while ctx.step < ctx.max_steps:
        ctx.step += 1
        print(f"\n[Step {ctx.step}] Calling LLM...")

        # --- Think ---
        response = await acompletion(
            model=model,
            messages=ctx.messages,
            tools=TOOL_DEFINITIONS,
            tool_choice="auto"
        )
        msg = response.choices[0].message

        # --- No tool call -> final answer ---
        if not msg.tool_calls:
            print(f"[Final Answer] {msg.content}")
            return msg.content

        # --- Act: execute each tool the LLM requested ---
        # Add assistant message with tool calls to history
        ctx.messages.append({
            "role": "assistant",
            "content": msg.content,
            "tool_calls": msg.tool_calls
        })

        for tool_call in msg.tool_calls:
            fn_name = tool_call.function.name
            fn_args = json.loads(tool_call.function.arguments)
            print(f"    Tool: {fn_name}({fn_args})")

            if fn_name in TOOLS:
                result = TOOLS[fn_name](**fn_args)
            else:
                result = f"Error: tool '{fn_name}' not found"

            print(f"    Result: {result[:200]}...")

            # --- Observe: feed result back into context ---
            ctx.messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": str(result)
            })

    return "Max steps reached without a final answer."

# --- Entry Point ---

if __name__ == "__main__":
    question = (
        "If Eliud Kipchoge runs at his marathon world record pace, "
        "how many hours would it take him to reach the Moon at its closest approach? "
        "Round to the nearest thousand hours."
    )

    answer = asyncio.run(run_agent(question))
    print(f"\n{'='*60}\nFinal Answer: {answer}")

Extends the minimal ReAct agent from 5.1 with two additions: session persistence and typed structured output. A SessionStore keeps an in-memory dict of message histories keyed by session_id — explicitly noted in the code as a stand-in for Redis/Postgres in production — and run_session_agent loads that history, prepends it to the new user turn, and saves the updated transcript back once the turn completes. Rather than returning free-text when the agent decides it is done, the loop adds a final_answer tool to the tool list; when the model calls it, the arguments are used to construct a ResearchOutput Pydantic model (answer, confidence, sources_used, steps_taken) which is returned directly, guaranteeing a typed contract at the call site instead of parsed prose. A short session_demo() shows two sequential turns where the second turn relies on memory of context established in the first.

# Core pattern: the model's "done" signal is a validated structured-output tool call,
# not free text — and the running transcript is persisted to the session store.
if fn_name == "final_answer":
    result = ResearchOutput(
        answer=fn_args["answer"],
        confidence=fn_args["confidence"],
        sources_used=fn_args.get("sources_used", []),
        steps_taken=steps
    )
    session_store.save(session_id, messages[1:])  # strip system msg
    return result

This implements the agent-as-tool pattern — one orchestrator delegates to specialized sub-agents. researcher_agent runs its own bounded ReAct loop (up to 8 steps) with access to search_web/calculator and returns a raw-text research brief; analyst_agent takes that brief plus an analysis goal and calls the LLM once with no tools at all, since it should reason over the context it’s given rather than gather new information. The top-level orchestrator_agent exposes run_researcher and run_analyst as its own tool definitions (ORCHESTRATOR_TOOLS); when the model calls one, the orchestrator dispatches to the matching specialist coroutine, caches researcher results by topic to avoid redundant lookups, and feeds each specialist’s output back into the orchestrator’s own message history as a tool response. This lets the orchestrator decompose a request, delegate pieces to the right specialist, and synthesize a final answer once no further delegation is needed.

# Core pattern: specialist agents are exposed to the orchestrator as ordinary tools —
# the orchestrator never sees their internal ReAct loops, only their final output.
for tool_call in msg.tool_calls:
    fn_name = tool_call.function.name
    fn_args = json.loads(tool_call.function.arguments)

    if fn_name == "run_researcher":
        topic = fn_args["topic"]
        if topic not in research_cache:
            research_cache[topic] = await researcher_agent(topic)
        result = research_cache[topic]
    elif fn_name == "run_analyst":
        result = await analyst_agent(
            fn_args["research_brief"],
            fn_args["analysis_goal"]
        )
    else:
        result = f"Unknown agent: {fn_name}"

    messages.append({
        "role": "tool",
        "tool_call_id": tool_call.id,
        "content": result
    })

// Node.js ReAct agent using OpenAI SDK
// npm install openai dotenv

import OpenAI from "openai";
import dotenv from "dotenv";

dotenv.config();

const client = new OpenAI({ apiKey: process.env.OPENAI_API_KEY });

// --- Tool Implementations ---

async function searchWeb(query, maxResults = 3) {
  /**
   * Web search via Tavily REST API.
   * Replace with any search provider — interface stays the same.
   */
  try {
    const response = await fetch("https://api.tavily.com/search", {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        Authorization: `Bearer ${process.env.TAVILY_API_KEY}`,
      },
      body: JSON.stringify({ query, max_results: maxResults }),
    });
    const data = await response.json();
    return data.results
      .map((r) => `[${r.title}] ${r.content}`)
      .join("\n");
  } catch (err) {
    return `Search error: ${err.message}`;
  }
}

function calculator(expression) {
  /**
   * Safe arithmetic evaluator.
   * In production replace with a proper math parser (mathjs etc.)
   */
  try {
    if (!/^[0-9+\-*/()., ]+$/.test(expression)) {
      return "Error: unsafe expression";
    }
    // eslint-disable-next-line no-eval
    return String(eval(expression));
  } catch (err) {
    return `Calc error: ${err.message}`;
  }
}

// --- Tool Registry ---

const TOOLS = {
  search_web: ({ query, max_results }) => searchWeb(query, max_results),
  calculator: ({ expression }) => calculator(expression),
};

const TOOL_DEFINITIONS = [
  {
    type: "function",
    function: {
      name: "search_web",
      description:
        "Search the web for current information. Use for facts, news, or data you don't know.",
      parameters: {
        type: "object",
        properties: {
          query: { type: "string", description: "Search query" },
          max_results: {
            type: "integer",
            description: "Number of results to return (default 3)",
          },
        },
        required: ["query"],
      },
    },
  },
  {
    type: "function",
    function: {
      name: "calculator",
      description: "Evaluate a mathematical expression. Use for any numeric calculation.",
      parameters: {
        type: "object",
        properties: {
          expression: {
            type: "string",
            description: "Math expression e.g. '356500 / 21.1'",
          },
        },
        required: ["expression"],
      },
    },
  },
];

// --- ReAct Agent Loop ---

async function runAgent({
  userInput,
  systemPrompt = "You are a helpful research assistant. Use tools when needed.",
  model = "gpt-4o",
  maxSteps = 10,
}) {
  const messages = [
    { role: "system", content: systemPrompt },
    { role: "user", content: userInput },
  ];

  for (let step = 1; step <= maxSteps; step++) {
    console.log(`\n[Step ${step}] Calling LLM...`);

    const response = await client.chat.completions.create({
      model,
      messages,
      tools: TOOL_DEFINITIONS,
      tool_choice: "auto",
    });
    const msg = response.choices[0].message;

    // --- No tool call -> final answer ---
    if (!msg.tool_calls || msg.tool_calls.length === 0) {
      console.log(`[Final Answer] ${msg.content}`);
      return msg.content;
    }

    // --- Add assistant message with tool calls ---
    messages.push({
      role: "assistant",
      content: msg.content ?? null,
      tool_calls: msg.tool_calls,
    });

    // --- Execute each tool call and feed results back ---
    for (const toolCall of msg.tool_calls) {
      const fnName = toolCall.function.name;
      const fnArgs = JSON.parse(toolCall.function.arguments);
      console.log(`    Tool: ${fnName}(${JSON.stringify(fnArgs)})`);

      let result;
      if (TOOLS[fnName]) {
        result = await TOOLS[fnName](fnArgs);
      } else {
        result = `Error: tool '${fnName}' not found`;
      }
      console.log(`    Result: ${String(result).slice(0, 200)}...`);

      messages.push({
        role: "tool",
        tool_call_id: toolCall.id,
        content: String(result),
      });
    }
  }

  return "Max steps reached without a final answer.";
}

// --- Entry Point ---

const answer = await runAgent({
  userInput:
    "What is the current market cap of NVIDIA and how does it compare to AMD? " +
    "Calculate the ratio between them.",
  maxSteps: 8,
});

console.log(`\n${"=".repeat(60)}\nFinal Answer: ${answer}`);

Extends the multi-agent pattern into a three-stage sequential pipeline — Researcher → Analyst → Writer — where each stage’s output is a Zod-validated structured object consumed as typed input by the next stage. A generic runStructuredAgent({ role, task, context, tools, outputSchema, ... }) factory runs the ReAct loop for any specialist: it builds a submit_output tool from the given Zod schema, and when the model calls it, the arguments are checked with schema.safeParse; on failure, the validation error is fed back to the model as a tool result so it can retry rather than the pipeline silently accepting malformed data. runMultiAgentPipeline wires the three stages together (runResearcherAgent → runAnalystAgent → runWriterAgent), passing each stage’s typed output forward as the next stage’s input/context, and logs timing and record counts at each handoff. The core idea demonstrated is that type safety — not shared memory or a shared prompt — is the contract between agents in the pipeline.

// Core pattern: the "done" signal is a schema-validated tool call; a failed validation
// is turned into feedback for the model instead of a thrown error or silent bad data.
for (const toolCall of msg.tool_calls) {
  const fnName = toolCall.function.name;
  const fnArgs = JSON.parse(toolCall.function.arguments);

  if (fnName === "submit_output") {
    const parsed = outputSchema.safeParse(fnArgs);
    if (parsed.success) return parsed.data;
    messages.push({
      role: "tool",
      tool_call_id: toolCall.id,
      content: `Validation error: ${JSON.stringify(parsed.error.issues)}. Please fix and resubmit.`,
    });
    continue;
  }
  // ...regular (non-terminal) tool execution omitted
}

A single ReAct loop handles tasks where steps cannot be predicted in advance. An agentic workflow goes one level higher: it composes multiple agents — or agents and deterministic steps — into a directed process where different segments are handled by different components, often running on different platforms.

The defining characteristic is who owns each segment:

• Developer-owned segments — deterministic steps: validation, formatting, routing, storage. Fixed cost, unit-testable, predictable.
• Agent-owned segments — open-ended steps: research, synthesis, judgment. Variable cost, bounded by step limits, evaluated statistically.

The key insight: not every stage needs to be an agent. Validation, formatting, storage, routing — these are cheaper and more reliable as deterministic code. Agents belong at the stages where the number of steps genuinely cannot be predetermined.

Three collaboration patterns cover most multi-agent systems; a fourth — cross-platform handoff — is the one that matters most in enterprise deployments:

Pattern 1: Sequential (Pipeline)

Each stage feeds directly into the next. The cleanest pattern — maximum traceability, easiest to debug.

Best for: Report generation, document processing, content pipelines. Weakness: Total latency = sum of all stage latencies. One failure blocks everything downstream.

Pattern 2: Parallel (Fan-Out / Fan-In)

Multiple agents work concurrently on independent sub-tasks. Results are aggregated before proceeding.

Best for: Competitive analysis, batch document processing, parallel data enrichment. Weakness: Aggregation logic can be complex. Total cost = sum of all parallel agent costs.

Pattern 3: Loop (Iterative Refinement)

An agent produces output, a critic evaluates it, and the loop continues until quality criteria are met or a step limit is reached.

Best for: Code generation with testing, content that must meet quality standards, self-healing data pipelines. Weakness: Unbounded cost if convergence is not guaranteed. Always set a hard iteration limit.

Pattern 4: Cross-Platform Handoff

The most complex and most powerful enterprise pattern. Different workflow segments run on different platforms — a no-code automation tool triggers an agent, the agent calls an enterprise API, results land in a database that feeds a BI dashboard.

flowchart TD
    classDef platform fill:#e3f2fd,stroke:#1565c0
    classDef agent fill:#e8f5e9,stroke:#2e7d32
    classDef data fill:#fff3e0,stroke:#e65100
    classDef human fill:#f3e5f5,stroke:#6a1b9a
    subgraph TRIGGER["Trigger Layer — Zapier / Make / n8n"]
        T1["CRM Event<br/>New enterprise lead"]
        T2["Email / Webhook<br/>Document received"]
        T3["Schedule<br/>Daily batch"]
    end
    subgraph AGENT["Agent Layer — Python / Node.js Service"]
        A1["Intake Agent<br/>Classify + enrich input"]
        A2["Research Agent<br/>Web search + internal KB"]
        A3["Analyst Agent<br/>Score + recommend"]
        HITL["Human Gate<br/>Approval required"]
    end
    subgraph INTEGRATION["Integration Layer — MCP / REST / GraphQL"]
        I1["Salesforce MCP Server<br/>Read/write CRM records"]
        I2["Slack MCP Server<br/>Send notifications"]
        I3["Internal API<br/>Custom MCP server"]
    end
    subgraph STORAGE["Storage Layer — Cloud"]
        S1["Postgres + pgvector<br/>Session + memory store"]
        S2["S3 / R2<br/>Document archive"]
        S3["Data Warehouse<br/>Snowflake / BigQuery"]
    end
    subgraph OUTPUT["Output Layer"]
        O1["BI Dashboard<br/>Tableau / Looker"]
        O2["Slack Notification<br/>Summary to team"]
        O3["CRM Update<br/>Enriched lead record"]
    end
    T1 --> A1
    T2 --> A1
    T3 --> A2
    A1 --> A2
    A2 --> A3
    A3 --> HITL
    HITL -->|"approved"| I1
    HITL -->|"approved"| I2
    HITL -->|"revision"| A3
    A2 --> I3
    I3 --> A2
    A1 --> S1
    A2 --> S2
    A3 --> S3
    I1 --> O3
    I2 --> O2
    S3 --> O1
    class T1,T2,T3 platform
    class A1,A2,A3 agent
    class HITL human
    class I1,I2,I3 data
    class S1,S2,S3,O1,O2,O3 platform

Best for: Enterprise automation, sales intelligence pipelines, compliance workflows, any process that crosses organizational system boundaries.

Understanding the internal mechanics of how one agent hands off to another — and how state flows through the chain — is essential for building reliable multi-platform workflows.

Four principles that make this work reliably:

1. Typed contracts between agents. Every handoff carries a validated schema — not a raw string. If the Researcher returns a ResearchBrief with a required keyFacts array, the Analyst knows exactly what it receives. Validation errors surface at the boundary, not silently downstream.
2. Checkpointing after every stage. The orchestrator saves state to the session store after each agent completes. If the Analyst fails, the pipeline resumes from the saved ResearchBrief — it does not re-run the Researcher.
3. Idempotent tool calls. Tools that write data (CRM updates, database writes, file creation) must be idempotent — calling them twice with the same arguments should produce the same result. This makes retry logic safe.
4. Explicit failure boundaries. Each agent has a hard maxSteps limit. When exceeded, it raises a typed error — not a silent timeout. The orchestrator catches this and decides whether to retry, skip, escalate to a human, or abort.

This section implements the cross-platform pattern in Python (FastAPI + Pydantic + LiteLLM): a webhook trigger kicks off a pipeline that spans Intake → Research → Analyst agents, then writes out to a CRM and Slack. Typed Pydantic models (WebhookPayload, EnrichedLead, ResearchPackage, SalesAnalysis, PipelineResult) form the contract at every handoff. A generic run_structured_agent() helper drives the tool-calling loop for any stage, forcing the model to call a submit_output tool shaped by the target Pydantic schema so the stage cannot “finish” without producing valid structured output. A CheckpointStore (in-memory here, Redis/Postgres in production) persists each stage’s result so a failed pipeline resumes from the last completed stage rather than re-running everything. The webhook endpoint (POST /webhook/lead) accepts the payload, returns a pipeline_id immediately, and runs the full pipeline in a background task; the caller polls GET /pipeline/{id} for the result. The final stage writes to CRM and sends a Slack notification concurrently via asyncio.gather.

async def run_sales_pipeline(payload: WebhookPayload, pipeline_id: str) -> PipelineResult:
    """Webhook -> Intake -> Research -> Analysis -> CRM/Slack (parallel)."""
    result = PipelineResult(pipeline_id=pipeline_id, entity_id=payload.entity_id, status="running")

    lead = await run_intake_agent(payload)
    checkpoint_store.save(pipeline_id, "intake", lead)

    research = await run_research_agent(lead)
    checkpoint_store.save(pipeline_id, "research", research)

    analysis = await run_analyst_agent(lead, research)
    checkpoint_store.save(pipeline_id, "analysis", analysis)

    crm_ok, slack_ok = await asyncio.gather(
        update_crm(payload.entity_id, analysis),
        send_slack_notification("sales-alerts", analysis, research),
    )
    result.crm_updated, result.notification_sent = crm_ok, slack_ok
    result.status = "complete"
    return result

@app.post("/webhook/lead", status_code=202)
async def receive_lead_webhook(payload: WebhookPayload, background_tasks: BackgroundTasks):
    """Webhook intake: accept immediately, run pipeline in background, caller polls for result."""
    pipeline_id = str(uuid.uuid4())
    background_tasks.add_task(run_sales_pipeline, payload, pipeline_id)
    return {"pipeline_id": pipeline_id, "poll_url": f"/pipeline/{pipeline_id}"}

This section shows the Node.js side of the same pattern: a lightweight Express server acting as the agent execution backend for any no-code automation platform. A reusable runAgent() function drives the OpenAI tool-calling loop with a maxSteps ceiling, executing whatever tools (web search, calculator) are registered in a handler lookup table. Three stage-runner wrappers (runSummaryAgent, runResearchAgent, runClassificationAgent) call runAgent() with fixed system prompts and tool sets. Endpoints split into synchronous routes (/agent/summarize, /agent/classify) that are fast enough to return inline, and asynchronous, job-queued routes (/agent/research, /workflow/pipeline) that return a jobId immediately, run in the background, and are either polled via GET /jobs/:jobId or pushed to a caller-supplied webhook_url — the two integration modes n8n, Zapier, and Make each need.

app.post("/agent/research", async (req, res) => {
  const { topic, webhook_url } = req.body;
  if (!topic) return res.status(400).json({ error: "topic is required" });

  const jobId = createJob({ type: "research", topic });
  res.status(202).json({ jobId, status: "queued", poll_url: `/jobs/${jobId}` });

  (async () => {
    try {
      updateJob(jobId, { status: "running" });
      const result = await runResearchAgent(topic);
      updateJob(jobId, { status: "complete", result, completedAt: new Date().toISOString() });

      // n8n / Zapier "resume on webhook" pattern
      if (webhook_url) {
        await fetch(webhook_url, {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({ jobId, result }),
        }).catch((e) => console.error("Webhook callback failed:", e.message));
      }
    } catch (err) {
      updateJob(jobId, { status: "failed", error: err.message });
    }
  })();
});

The Node.js server above is designed to be a drop-in agent execution backend for any no-code automation platform. Here is exactly how each platform calls it:

Integration pattern summary:

Key design decision — sync vs. async endpoints:

• Use synchronous endpoints (/agent/summarize, /agent/classify) for Make and simple Zapier zaps — these platforms handle responses inline.
• Use async + polling (/agent/research, /workflow/pipeline) for long-running tasks — n8n’s Wait node and Zapier’s Delay + Webhook combination handle this cleanly.
• Use webhook callbacks (webhook_url parameter) when the platform supports “resume on webhook” — n8n’s Webhook trigger in wait mode is the best implementation of this.

← Building Efficient AI Agents — Part 2: Technology Stack, Deployment Patterns & Cost · Building Efficient AI Agents — Part 4: Security, Evaluation & Production Operations →