Async OpenAI Function Stream

Utility helpers for stream-based, asynchronous handling of OpenAI “function-call” messages (or any JSON RPC-style exchange):

• Wraps your local TypeScript/JavaScript functions so they can be called through a streamed JSON protocol.
• Validates incoming function-call payloads against parameter schemas.
• Sends back success or error responses in the OpenAI function-call output format.
• Seamlessly joins multiple function collections into one.
• Works with @mojsoski/async-stream, @silyze/async-pipe, @silyze/async-duplex, or any compatible stream implementation.

Install

npm i @silyze/async-openai-function-stream

Quick usage

import {
  createJsonStreamFunctions,
  joinFunctions,
  JsonFunctionData,
} from "@silyze/async-openai-function-stream";
import { AsyncDuplex } from "@silyze/async-duplex";

// 1.  Create a duplex “wire” over which JSON function-call frames
//     will travel (could be a WebSocket, TCP socket, etc.).
const wire = new AsyncDuplex<JsonFunctionData>();

// 2.  Declare functions and their JSON parameter schemas.
const tools = [
  {
    type: "function",
    name: "test",
    description: "A demo function",
    parameters: {
      type: "object",
      properties: { in: { type: "string" } },
      required: ["in"],
    },
  },
  {
    type: "function",
    name: "test2",
    description: "Another demo function",
    parameters: {
      type: "object",
      properties: { in: { type: "string" } },
      required: ["in"],
    },
  },
];

// 3.  Wrap the declarations into runnable calls.
const streamFunctions = joinFunctions(
  createJsonStreamFunctions(tools, wire.reverse())
);

// 4.  Echo everything written back onto the wire (a simple loop-back
//     in this demo). In real use you would forward to OpenAI, etc.
wire.transform().pipe({
  write(frame) {
    // echo straight back
    return wire.write(frame);
  },
});

// 5.  Invoke the remote functions exactly like local async functions.
async function demo() {
  const out1 = await streamFunctions.collection["test2"]({ in: "OK 2" });
  console.log("result:", out1); // -> { ... }

  const out2 = await streamFunctions.collection["test"]({ in: "OK 1" });
  console.log("result:", out2); // -> { ... }
}

demo();

API Reference

createFunctionCalls(...functions: FunctionWithSchema[]): FunctionCalls

Builds a function-call bundle from an array of { tool, call } objects.

createJsonStreamFunctions(functionTools, stream, signal?)

Creates FunctionCalls backed by an async JSON frame stream.
Each tool becomes an async function that:

1. Generates a unique ID.
2. Writes { id, name, value } into stream.
3. Waits for a frame with the same id to arrive back, then resolves with value.

new FunctionStream(collection)

Low-level AsyncStream<JsonValue, JsonValue> that listens for OpenAI function-call frames and pushes the corresponding result/error frames.

joinFunctions(...calls: FunctionCalls[]): FunctionCalls

Merges any number of FunctionCalls bundles, deduplicating tools and merging collection.

Types

| Name | Purpose | | ------------------ | ---------------------------------------------------------------------------- | | FunctionTool | OpenAI-compatible metadata (type, name, description, parameters). | | CallableFunction | (input: JsonValue) => Promise<JsonValue> — implementation signature. | | FunctionCalls | { tools: FunctionTool[]; collection: Record<string, CallableFunction>; } | | JsonFunctionData | { id: string; name: string; value: JsonValue; } — frame shape on the wire. |