Calling Server Nodes from Workflows

The Bridge Pattern

From the workflow engine's perspective, a call to a server node looks like any other node execution. The difference is that the executor resolves a server node from the registry rather than performing the operation itself. This keeps the workflow engine unaware of whether a node is in-process or an external service.

Workflow Node Configuration

{
  "nodeType": "ServerNodeCall",
  "name": "embedDocumentText",
  "config": {
    "serverNodeId": "local-embedding",
    "payload": {
      "text":       "$output.extractText.content",
      "dimensions": 384,
      "normalize":  true
    },
    "timeoutSeconds": 30
  }
}

ServerNodeCallExecutor Implementation

public class ServerNodeCallExecutor : BaseNodeExecutor<ServerNodeCallSettings>
{
    private readonly IServerNodeRegistry _registry;
    private readonly IExpressionEvaluator _expr;

    public ServerNodeCallExecutor(
        IServerNodeRegistry registry,
        IExpressionEvaluator expr)
    {
        _registry = registry;
        _expr     = expr;
    }

    protected override async Task<NodeExecutionResult> ExecuteAsync(
        NodeExecutionContext ctx,
        ServerNodeCallSettings settings,
        CancellationToken ct)
    {
        // 1. Resolve target server node
        if (!_registry.TryGetNode(settings.ServerNodeId, out var serverNode) || serverNode is null)
            return NodeExecutionResult.Failure(
                $"Server node '{settings.ServerNodeId}' is not registered or not ready.");

        // 2. Evaluate payload expressions
        var evaluatedPayload = await _expr.EvaluateObjectAsync(settings.Payload, ctx, ct);

        // 3. Build request envelope
        var request = new ServerNodeRequest
        {
            ExecutionId = ctx.ExecutionId,
            TenantId    = ctx.TenantId,
            NodeId      = settings.ServerNodeId,
            Payload     = JsonSerializer.SerializeToDocument(evaluatedPayload)
        };

        // 4. Dispatch with timeout
        using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        cts.CancelAfter(TimeSpan.FromSeconds(settings.TimeoutSeconds ?? 30));

        var response = await serverNode.HandleRequestAsync(request, cts.Token);

        // 5. Map response to node output
        if (!response.Success)
            return NodeExecutionResult.Failure(
                response.ErrorMessage ?? "Server node returned an error.");

        return NodeExecutionResult.Success(response.Data);
    }
}

Error Paths