@flowget/ai

Request-first, stateless AI workflow-authoring primitives for Flowget.

Send a natural-language command → get a schema-valid workflow graph back. author(config, request) owns the catalog, the LLM, toolsets, guardrail, authorizers, and validation. This package ships primitives, not an HTTP server — build a config with resolveConfig and call author, then wrap it in whatever transport you need (a customer who wants an API writes it themselves).

Stateless by design. The workflow graph IS the state, so there is no session store — every request is self-contained. Optional continuity is carried by the client via context. A chat is a thin, optional layer on the same primitive.

command + currentGraph? + actor?  →  { kind: "proposal", graph, summary } | { kind: "message", text }

• proposal — a workflow graph to preview / apply. Node positions are optional (the builder lays it out).
• message — a text reply, including a clarifying question when the command is ambiguous.

⚠️ A proposal is model-authored — untrusted until approved. Validation gates runnability (schema, known node types, resolvable refs), NOT field values: a malicious or mistaken value (e.g. an exfiltrating URL in an http_request node) passes. Never auto-apply a proposal without human review and/or a downstream field-value policy check. Treat summary (and any message.text) as untrusted model output — output-encode before rendering.

Install

npm install @flowget/ai
# or: pnpm add @flowget/ai · yarn add @flowget/ai · bun add @flowget/ai

Developed with Bun internally; consumers can use any package manager and any Node-compatible runtime.

Quickstart

import { resolveConfig, author, openaiAdapter } from "@flowget/ai";

const config = await resolveConfig({
  adapter: openaiAdapter({ apiKey: process.env.OPENROUTER_API_KEY! }),
  catalog: { registryDir: process.env.REGISTRY_DIR! }, // or a pre-loaded NodeMetadata[]
});

const event = await author(config, {
  command: "When a webhook fires, email ops with the order payload",
});

if (event.kind === "proposal") {
  console.log(event.summary, event.graph); // a workflow graph authored from YOUR registry
} else {
  console.log(event.text); // a clarifying question or a plain reply
}

A proposal graph looks like:

{
  "kind": "proposal",
  "summary": "Emails ops whenever the order webhook fires.",
  "graph": {
    "nodes": [
      { "id": "trigger_1", "type": "webhook_trigger", "data": { "path": "/orders" } },
      { "id": "email_1", "type": "send_email",
        "data": { "to": "[email protected]", "subject": "New order",
                  "body": "{{ steps.trigger_1.output.payload }}" } }
    ],
    "edges": [{ "id": "e1", "source": "trigger_1", "target": "email_1" }]
  }
}

Point catalog at your own directory of node .json files ({ registryDir }) to author against your real catalog. resolveConfig loads the catalog; finalizeConfig is the sync variant when you already hold a NodeMetadata[].

Configuration

Every piece is a named, typed, swappable hook with a batteries-included default. Only adapter and catalog are required.

| Option | Default | What it does | | ------------------- | -------------------------------------- | ----------------------------------------------------------------------- | | adapter | — (required) | The LLM. openaiAdapter({ apiKey }) → OpenRouter, or BYO. | | catalog | — (required) | NodeMetadata[], { registryDir }, or { paths }. | | guardrail | none | Pre-LLM policy gate → allow / block(message). | | authorizers | [] | Verify the trusted actor before guardrail/toolsets. | | toolsets | [] | Actor-scoped, model-invoked tools. | | hooks | no-op | onProposal / onBlocked audit sinks. | | validator | AJV + catalog cross-ref + template-ref | Bounds what can become a proposal. | | persona | workflow-only persona | System-prompt override. | | model | adapter default | Model slug forwarded to the adapter. | | maxToolIterations | 8 | Max adapter turns per request. | | maxRepairAttempts | 2 | Max validate/repair retries before giving up with a message. |

BYO LLM adapter

The adapter is the only LLM-specific seam — one method:

import type { AiAdapter } from "@flowget/ai";

const myAdapter: AiAdapter = {
  async generate({ messages, tools }) {
    // Call any LLM/gateway with the messages + tool definitions.
    // Return the assistant's turn: free text and/or parsed tool calls.
    return { text: null, toolCalls: [] };
  },
};

Actor-scoped toolset

Model-invoked tools run against your data, scoped by the actor. A domain miss is a return value; an infra fault throws. See examples/db-toolset.ts (in the repo — the examples/ dir is not in the published tarball):

import type { Toolset } from "@flowget/ai";

const dbToolset: Toolset = {
  namespace: "db",
  tools: [
    {
      name: "find_user",
      description: "Look up a user by email within the caller's tenant.",
      parameters: { type: "object", required: ["email"],
        properties: { email: { type: "string" } } },
      run: async (input, { actor }) => {
        const user = await db.findUser({ email: String(input.email), tenantId: actor?.tenantId });
        return user ? { found: true, user } : { found: false };
      },
    },
  ],
};

⚠️ actor is client-supplied JSON — not trusted by default. A toolset that scopes data by actor (like tenantId above) is only safe when an authorizer verifies the actor first; otherwise a client can spoof tenantId and read another tenant's data. Always wire an authorizer alongside an actor-scoped toolset:

await resolveConfig({ adapter, catalog, authorizers: [verifyActor], toolsets: [dbToolset] });

The package is auth-agnostic — it never forces authentication — so enforcing this is your call. See requireVerifiedTenant in examples/db-toolset.ts.

Guardrail + audit hooks

⚠️ Bounding request inputs is a hard integration requirement. command, currentGraph, and context are all JSON.stringify'd into the prompt, and currentGraph / context can be re-sent up to maxToolIterations (default 8) times per request. The host MUST cap the size of all three before calling author — an unbounded currentGraph or context is as much a token-DoS as a huge command. A guardrail is the natural place:

import { resolveConfig, author, block, inMemoryAuditSink } from "@flowget/ai";

const audit = inMemoryAuditSink(); // default append-only sink

// Cap EVERY prompt input, not just the command.
const tooBig = (value: unknown, max: number) =>
  value !== undefined && JSON.stringify(value).length > max;

const config = await resolveConfig({
  adapter, catalog,
  guardrail: (req) =>
    req.command.length > 2_000 ||
    tooBig(req.currentGraph, 50_000) ||
    tooBig(req.context, 10_000)
      ? block("Request too large.")
      : { action: "allow" },
  hooks: audit, // audit.entries collects every proposal / block
});
const event = await author(config, { command });

Testing with a mock adapter (no LLM, no network)

Drive the author primitive with a deterministic mock and assert the message | proposal:

import { resolveConfig, author, mockAdapter, mockProposeWorkflow } from "@flowget/ai";

const config = await resolveConfig({
  catalog,
  adapter: mockAdapter([
    mockProposeWorkflow(
      { nodes: [{ id: "t1", type: "webhook_trigger", data: {} }], edges: [] },
      "A one-node workflow.",
    ),
  ]),
});

const event = await author(config, { command: "start on a webhook" });
// event: { kind: "proposal", graph, summary }

Wrapping it in a transport

author is transport-agnostic — the graph is the state, so a request is self-contained. Wrap it in whatever you already run (an HTTP route, a queue consumer, a CLI). A minimal Fetch handler:

import { resolveConfig, author, openaiAdapter } from "@flowget/ai";

const config = await resolveConfig({
  adapter: openaiAdapter({ apiKey: process.env.OPENROUTER_API_KEY! }),
  catalog: { registryDir: process.env.REGISTRY_DIR! },
});

export default {
  async fetch(req: Request): Promise<Response> {
    const { command, currentGraph, actor, context } = await req.json();
    const event = await author(config, { command, currentGraph, actor, context });
    return Response.json(event);
  },
};

Self-hosted or Flowget-hosted, the primitive is the same stateless call — any infra, any runtime, any transport.

Streaming (authorStream)

For a chat "typing" feel, authorStream yields text-deltas as the model writes, then exactly one terminal proposal / message / error. Discriminate on event.kind. Frame the events over whatever streaming transport you run — example wiring you build; the library pins no wire format. A minimal SSE handler:

import { resolveConfig, authorStream, openaiAdapter } from "@flowget/ai";

const config = await resolveConfig({
  adapter: openaiAdapter({ apiKey: process.env.OPENROUTER_API_KEY! }),
  catalog: { registryDir: process.env.REGISTRY_DIR! },
});

export default {
  async fetch(req: Request): Promise<Response> {
    const { command, currentGraph, actor, context } = await req.json();
    const encoder = new TextEncoder();
    const stream = new ReadableStream({
      async start(controller) {
        for await (const event of authorStream(config, { command, currentGraph, actor, context })) {
          // event.kind: "text-delta" | "proposal" | "message" | "error"
          controller.enqueue(encoder.encode(`event: ${event.kind}\ndata: ${JSON.stringify(event)}\n\n`));
        }
        controller.close();
      },
    });
    return new Response(stream, { headers: { "content-type": "text/event-stream" } });
  },
};

An adapter without generateStream still works — authorStream emits no deltas, just the terminal event. A mid-stream failure arrives as a terminal { kind: "error", error } rather than a throw, so your case "error" is reachable.

License

FSL-1.1-ALv2.