Streaming function calls in Node.js with async iterators

Wiring up streaming function calls nodejs async iterators changes how your backend handles tool-using LLMs: instead of blocking on a full response, you process argument fragments and dispatch side effects as they arrive. This pattern cuts latency and lets you cancel runaway calls mid-stream. The following steps build a complete TypeScript pipeline against any OpenAI-compatible chat completions API.

Step 1: Scaffold the Node.js project

Create a directory and initialize an ESM Node project. We use ESM because top-level await and clean import syntax keep the async iterator code readable.

mkdir stream-fc && cd stream-fc
npm init -y
npm pkg set type=module
npm install openai zod
npm install -D tsx typescript

We pull in openai for the client and zod for runtime argument validation. You can swap openai for a bare fetch if you prefer zero deps; the stream shape is identical. Generate a minimal tsconfig.json:

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "NodeNext",
    "moduleResolution": "NodeNext",
    "strict": true
  }
}

Step 2: Configure an OpenAI-compatible client

Instantiate the client with a base URL and key. If you point it at n4n.ai, you get one OpenAI-compatible endpoint that addresses 240+ models and automatic fallback when a provider is rate-limited, but the code below works against any server that speaks the protocol.

import OpenAI from 'openai';

const client = new OpenAI({
  baseURL: process.env.LLM_BASE_URL ?? 'https://api.openai.com/v1',
  apiKey: process.env.LLM_API_KEY!,
});

const MODEL = process.env.LLM_MODEL ?? 'gpt-4o-mini';

Set those env vars in a .env or shell. The stream: true flag is what unlocks token and tool-call deltas. The client returns an async iterable; you do not need to manually pump a ReadableStream unless you are bypassing the SDK.

Step 3: Declare function schemas and a local executor

Define a couple of tools. Keep schemas strict; the model will emit arguments as a JSON string fragment by fragment.

const tools = [{
  type: 'function' as const,
  function: {
    name: 'get_weather',
    parameters: {
      type: 'object',
      properties: { city: { type: 'string' } },
      required: ['city'],
    },
  },
}];

const executors = {
  async get_weather({ city }: { city: string }) {
    // Stand-in for a real HTTP call or DB lookup
    return { temp_c: 21, city };
  },
};

This trivial executor stands in for a DB query or external API. In production, wrap it with timeouts, retries, and auth. Never trust the model to validate types—that is what step 5 handles.

Step 4: Consume the stream with an async iterator

The heart of streaming function calls nodejs async iterators is the for await loop over the completion stream. Each chunk may contain a delta.tool_calls array. We accumulate by index because the model can emit multiple parallel calls in a single response.

import type { ChatCompletionChunk } from 'openai/resources/chat/completions';

type Accum = { id?: string; name?: string; args: string };

async function streamToolCalls(messages: any[]) {
  const accum = new Map<number, Accum>();
  const stream = await client.chat.completions.create({
    model: MODEL,
    messages,
    tools,
    stream: true,
  });

  for await (const chunk of stream) {
    const tcDelta = chunk.choices[0]?.delta?.tool_calls;
    if (!tcDelta) continue;
    for (const call of tcDelta) {
      const idx = call.index;
      const cur = accum.get(idx) ?? { args: '' };
      if (call.id) cur.id = call.id;
      if (call.function?.name) cur.name = call.function.name;
      if (call.function?.arguments) cur.args += call.function.arguments;
      accum.set(idx, cur);
      // Forward partial args to a UI or buffer here if needed
    }
  }
  return [...accum.entries()].map(([idx, v]) => ({
    index: idx,
    id: v.id!,
    name: v.name!,
    arguments: JSON.parse(v.args) as unknown,
  }));
}

Note we only JSON.parse after the stream closes. Partial JSON will throw, so don’t parse mid-stream unless you ship a streaming JSON parser such as partial-json. The async iterator yields control back to the event loop between chunks, which is what makes cancellation and concurrency feasible.

Step 5: Validate and run the functions

Once the stream ends, we have complete argument strings. Validate with zod before execution to avoid injecting garbage into your backend.

import { z } from 'zod';

const WeatherSchema = z.object({ city: z.string() });

async function executeCalls(calls: { name: string; arguments: unknown }[]) {
  const results = [];
  for (const call of calls) {
    if (call.name === 'get_weather') {
      const parsed = WeatherSchema.parse(call.arguments);
      const data = await executors.get_weather(parsed);
      results.push({ name: call.name, data });
    } else {
      results.push({ name: call.name, error: 'unknown tool' });
    }
  }
  return results;
}

If you need to run tools concurrently, Promise.all the executors. Keep side effects idempotent; the model may retry the same call if you loop. For long-running tools, consider emitting progress by writing to a separate channel rather than blocking the iterator.

Step 6: Feed results back and close the loop

OpenAI expects a role: 'tool' message per call, referencing the tool_call_id. This continues the conversation so the model can synthesize a final answer. Streaming function calls nodejs async iterators becomes a cycle: stream → accumulate → execute → append → stream again.

async function runAgent(userPrompt: string) {
  const messages = [{ role: 'user', content: userPrompt }];
  const calls = await streamToolCalls(messages);
  if (calls.length === 0) return messages;

  const results = await executeCalls(calls);
  for (let i = 0; i < calls.length; i++) {
    messages.push({
      role: 'assistant',
      content: null,
      tool_calls: [{
        id: calls[i].id,
        type: 'function',
        function: { name: calls[i].name, arguments: JSON.stringify(calls[i].arguments) },
      }],
    });
    messages.push({
      role: 'tool',
      tool_call_id: calls[i].id,
      content: JSON.stringify(results[i].data ?? results[i].error),
    });
  }

  const final = await client.chat.completions.create({ model: MODEL, messages, stream: false });
  return final.choices[0].message.content;
}

The second call uses stream: false for brevity. In a real app you’d loop the streaming path until finish_reason === 'stop', handling multiple tool rounds. If a tool fails, return the error string in the tool message and let the model recover.

Step 7: Verify the pipeline end to end

Create index.ts:

runAgent('What is the weather in Oslo?').then(console.log).catch(console.error);

Run with:

npx tsx index.ts

Success looks like: the process prints a natural language answer that includes the mocked temperature (21°C) for Oslo, and no JSON.parse errors appear. Add console.log inside the for await loop to confirm deltas arrive incrementally—you should see args growing chunk by chunk. If you kill the process mid-stream, partial state is discarded, proving the async iterator yields control back to the event loop.

For production, attach an AbortController to the stream consumption and reject on abort. The provider will meter per-token usage; capture chunk.usage on the final delta if your endpoint exposes it. If you need to honor client routing directives or provider cache-control hints, forward them in the request options—they pass through transparently on compliant gateways.

That is the entire flow. You now have a minimal, composable base for streaming function calls nodejs async iterators that you can extend with parallel tools, retries, and partial-argument UIs without rewiring your core loop.