Defining Nodes

In Casacaide, a node is composed of 3 async functions

prep : select required data from the state and prepare it (formatting, history pruning, compression, etc.)
exec : execute your business logic using prep output
post : postprocess if needed, write back to the state and spawn the next nodes using exec output

We will learn how to write a node by implementing a node that calls an LLM (part of ReAct agent).

Since this is an LLM node, we have to consider streaming. We use the Canonical shapes prescribed in this tutorial. However, it is not required, which is why you see any types for all message like objects. Please read the capabilities/streaming section for an in depth discussion. The helper functions used are purely for convenience.

There ate two types of server nodes

streaming
non-streaming

We will discuss both and types.

Types


export interface ServerNodeDefinition<TPrep = any, TExec = any>
  extends NodeDefinition {
  isStreaming: false;
  prep: (cascadeContext: WorkflowContext, initialContext: any) => Promise<TPrep>;
  exec: (prepOutput: TPrep, controller?: CascadeController) => Promise<TExec>;
  post: (execOutput: TExec) => Promise<PostResult>;
}
 
export interface StreamingServerNodeDefinition<TPrep = any>
  extends NodeDefinition {
  isStreaming: true;
  prep: (cascadeContext: WorkflowContext, initialContext: any) => Promise<TPrep>;
  exec: (prepOutput: TPrep, controller?: CascadeController) => Promise<StreamConfig>;
  post: (execOutput: StreamingExecOutput) => Promise<PostResult>;
}
 
export type StreamingExecOutput = {
  assistantMessage: any;
  uiAssistantMessage?: any;
  cascadeId?: string;
  history: any[];
  userId?: string;
}
 
 
export type Updates = {
  [cascadeId: string] : WorkflowStep
}
 
export type SpawnContext<T = Record<string, any>> = {
  cascadeId?: string;
  history: any;
  userId?: string;
} & T;
 
export type Spawns = Record<string, SpawnContext>;
 
export type PostResult = {
  updates: Updates;
  uiUpdates? : Updates;
  spawns?: Record<string, SpawnContext>; 
}
 
export type WorkflowStep = {
  history: any[];
  status: string;
  [extraKeys: string]: any; // this should give some freedom for custom keys
};
 
export type WorkflowContext = {
  [cascadeId: string]: WorkflowStep[];
};

As you can see, the differences between streaming and non streaming nodes happen in the return ofexec and the input of post.

Prep

🎯

Goal: read from cascade state and initialContext input and prepare input for exec step.

Here, we assume that this node is part of an agent that uses Canonical types.


import { WorkflowContext} from "@cascaide-ts/core";
import { toProviderHistory } from "@cascaide-ts/helpers";
 
type SearchAgentPrepOut = {
    history: any[]; //type should be geminiMessage[], as we convert before passing
    cascadeId: string;
  };
 
 
export async prep(cascadeContext: WorkflowContext, initialContext: any): Promise<SearchAgentPrepOut> {
        const cascadeId = initialContext.cascadeId;
        const dataArray = cascadeContext[cascadeId];
        const canonicalHistory = dataArray.flatMap((item: any) => item.history || []);
        const history = toProviderHistory('gemini-genai', canonicalHistory);    
 
      
        return { history, cascadeId };
}

Two points to note here:

We barely used initialContext. initialContext contains the cascadeId, userId, history and whatever data you pass in when you spawn the node. any type is used to give developers some freedom, but we recommend you enforce type safety at the node declaration level as per your requirement. The reason we did not use history from initialContext is because by the time prep is run, context[cascadeId] will contain the output from the history from the previous node.
context[cascadeId] will give you an array of WorkflowStep

Cascaide enforces the presence history and status, but you may add custom keys. Then, you can map out the values you need in the prep step.

Exec

This is where streaming and node streaming nodes first diverge. We will see how llm calls work.

Streaming Node


 
import { callLLM, ToolParam, buildTools } from '@cascaide-ts/helpers';
import { StreamConfig } from '@cascaide-ts/core';
 
export async exec(prepOutput: SearchAgentPrepOut, controller?: CascadeController): Promise<StreamConfig> {
      const { history } = prepOutput;
 
      /*
          We are not using the controller in this example. Every `exec` gets one, and it allows you
          to control the graph execution from within nodes (delegation, recursion, pausing the node for some condition, etc.)
 
          See: recursiveReactAgent tutorial
      */
 
      const systemPrompt = `
            You are an expert technical AI assistant equipped with web search capabilities.
            You have access to the following tools:
            1. search_tool
            Use for general web searches, finding documentation, or retrieving facts.
            Pass a clear, concise query.
 
      `.trim();
 
      /*
             We will construct the tools here. We will write the tool definitions in `ToolParam` shape and 
             convert it into provider specific shape using `buildTools` helper.
      */
 
        const searchToolParams: ToolParam[] = [
                {
                  name:        'search_tool',
                  description: 'Searches the web using the input query.',
                  parameters: {
                    type: 'object' as const,
                    properties: {
                      query: { type: 'string', description: 'The natural language query' },
                    },
                    required: ['query'],
                  },
                },
              ];
 
      const geminiTools = buildTools('gemini-genai', searchToolParams);
      const {stream, provider} = await callLLM('gemini-genai', 'gemini-3-flash-preview', systemPrompt, history, geminiTools,  true);
      return { stream, provider };
},

This is a streaming node. For cascaide to be able to correctly assemble the stream, the return must be of type


export interface StreamConfig {
  stream:       AsyncIterable<any>;
  provider:     string;  
  mapper?:      ChunkMapper | (() => ChunkMapper);
  filter?:      StreamFilter;
}
 
where provider can be 'gemini-genai' | 'openai' | 'openai-responses' | 'anthropic' | 'custom'

For a deep dive into streaming, please refer to capabilities/streaming.

Non Streaming Node

This is more straigtforward. Perform the logic you need (an LLM api call in this case), and pass the results and other required data to post


import { callLLM, ToolParam, buildTools, parseCompletedResponse  } from '@cascaide-ts/helpers';
 
type SearchAgentExecOut = {
    assistantMessage : CanonicalMessage;
    cascadeId: string;
}
  
 
export async exec(prepOutput: SearchAgentPrepOut, controller?: CascadeController): Promise<SearchAgentExecOut> {
      const { history, cascadeId } = prepOutput;
      const systemPrompt = `
            You are an expert technical AI assistant equipped with web search capabilities.
            You have access to the following tools:
            1. search_tool
            Use for general web searches, finding documentation, or retrieving facts.
            Pass a clear, concise query.
 
      `.trim();
      const searchToolParams: ToolParam[] = [
                {
                  name:        'search_tool',
                  description: 'Searches the web using the input query.',
                  parameters: {
                    type: 'object' as const,
                    properties: {
                      query: { type: 'string', description: 'The natural language query' },
                    },
                    required: ['query'],
                  },
                },
              ];
 
      const geminiTools = buildTools('gemini-genai', searchToolParams);
      //non streaming callLLM
      const {response, provider} = await callLLM('gemini-genai', 'gemini-3-flash-preview', systemPrompt, history, geminiTools,  false);
      const assistantMessage = parseCompletedResponse(provider,response);
 
      return { assistantMessage, cascadeId };
}

Post

In the post step we

post process the exec step result if needed (we create pending tool calls in this example)
write updates to the state using Updates in PostResult
optionally write uiUpdates as well in PostResult
spawn the next set of nodes using Spawns in PostResult

The input to post changes basis if the exec step was streaming or not.

Streaming Node

If exec was streaming, the input will of type


export type StreamingExecOutput = {
  assistantMessage: any;
  uiAssistantMessage?: any;
  cascadeId?: string;
  history: any[];
  userId?: string;
}


export async post(execOutput: StreamingExecOutput): Promise<PostResult> {
      const { assistantMessage, uiAssistantMessage, history, cascadeId, userId } = execOutput;
 
      const pendingToolCalls = assistantMessage.tool_calls ?? [] as CanonicalToolCall[]
 
      /*
        if you had a filter set up in `StreamConfig`, the two `assistantMessage`s would differ.
        By passing an optional `uiUpdate`, you can censor what the client side sees even when hydrating 
        the cascade in new user sessions. Otherwise, the client would be censored only during the stream,
        and upon hydrating the cascade later, the full sensitive `Update` will hit the client.
 
      */
 
      const isDifferent = uiAssistantMessage !== undefined &&
      JSON.stringify(assistantMessage) !== JSON.stringify(uiAssistantMessage);
  
      return {
        updates: {
          [cascadeId as string]: {
            history:    [assistantMessage as CanonicalMessage],
            status:     pendingToolCalls.length > 0 ? 'calling_tool' : 'complete',
            lastUpdate: Date.now(),
            //you can add more fields if you want, it will be written to the cascade state
            // what's mandatory is history and status
          },
        } as Updates,
        ...(isDifferent ? {
          uiUpdates: {
            [cascadeId as string]: { history: [uiAssistantMessage as CanonicalMessage], status: pendingToolCalls.length > 0 ? 'calling_tool' : 'complete' },
          },
        } : {}),
        spawns: pendingToolCalls.length > 0
          ? {  // we spawn the hardcoded searchToolNode if there are tool calls
              ['searchToolNode']: {
                history:            [assistantMessage],
                toolCallsToExecute: pendingToolCalls,
                cascadeId,
              },
            } as Spawns
          : undefined,

Non Streaming Node

If exec was non streaming, the input will be whatever the exec output was.


 
export async post(execOutput: SearchAgentExecOut): Promise<PostResult> {
      const { assistantMessage, cascadeId} = execOutput;
      const pendingToolCalls = assistantMessage.tool_calls ?? [] as CanonicalToolCall[];
      /*
       Skipping the uiUpdate part here, since we only returned `assistantMessage`.
      */
   
      return {
        updates: {
          [cascadeId as string]: {
            history:    [assistantMessage as CanonicalMessage],
            status:     pendingToolCalls.length > 0 ? 'calling_tool' : 'complete',
            lastUpdate: Date.now(),
            //you can add more fields if you want, it will be written to the cascade state
            // what's mandatory is history and status
          },
        } as Updates,
        
        spawns: pendingToolCalls.length > 0
          ? {  
              ['searchToolNode']: {
                history:            [assistantMessage],
                toolCallsToExecute: pendingToolCalls,
                cascadeId,
              },
            } as Spawns
          : undefined,

Some behaviours worth noting when spawning nodes

no cascadeId => this spawn is not part of a cascade, ephemeral, no persistence. Useful when you need one of fire and forget nodes, that do not need to be persisted. They are ghosts -> cannot write to state, but could spawn new nodes. This is mostly used to spawn observer UI nodes to watch cascades and/or initiate new ones.
no userId => safe to omit. the new node will inherit from the parent. Then why is it available in initialContext and post? userIds enter the nodes because you might need it for retrieving memories, or other scoped operations.