Fluency Workflow Telemetry SDK

Send automated workflow executions to Fluency so your team can compare them with human executions of the same business process. Workflow Telemetry records the duration and outcome of each execution, its meaningful steps, and the artifacts it reads or creates. It works for AI agents, deterministic automations, and long-running services. Fluency attempts to associate each run with a process already observed in your organization and creates an automated workflow process when no safe match exists.

What Fluency provides

Fluency provides two server-side secrets during onboarding:

FLUENCY_WORKFLOW_ENDPOINT=https://YOUR-REGIONAL-ENDPOINT
FLUENCY_WORKFLOW_API_KEY=YOUR-ORGANIZATION-API-KEY

The API key is owned by one WorkOS organization and carries only workflow-telemetry:write. The receiving service derives the organization from the verified key. Your application does not send a Fluency organization ID, user login, process ID, or WorkOS administrator key. Keep the key in a server-side secret manager. Do not expose it in browser code, prompts, telemetry content, logs, or source control.

Install

Fluency currently supplies versioned release files during onboarding:

# TypeScript or Node.js
npm install ./fluency-workflow-telemetry-0.1.0.tgz

# Python
pip install ./fluency_workflow_telemetry-0.1.0-py3-none-any.whl

The TypeScript package requires Node.js 24 or later. The Python package requires Python 3.12 or later.

Choose the executor type

Every integration declares the system that performed the run:

Type	Use it for
`agent`	An AI agent that plans, calls models or tools, or makes decisions dynamically.
`automation`	A deterministic job, RPA flow, scheduler, queue worker, or orchestration workflow.
`service`	A backend service performing a business operation as part of its normal request processing.

Use a stable executor ID across deployments. Change the optional version when the implementation, model, or workflow definition materially changes.

TypeScript quick start

Initialize once when the service starts. Wrap each complete business execution in run and each meaningful operation in step.

import { init } from '@fluency/workflow-telemetry';

const fluency = init({
  endpoint: process.env.FLUENCY_WORKFLOW_ENDPOINT!,
  executor: {
    type: 'automation',
    id: 'accounts-payable-workflow',
    name: 'Accounts payable workflow',
    version: '2026.09',
  },
  onError: (error) => logger.warn({ error }, 'Fluency telemetry delivery failed'),
});

export async function processInvoice(invoiceId: string, executionId: string) {
  return fluency.run(
    'Process supplier invoice',
    async (run) => {
      const invoice = await run.step('Read invoice', () => readInvoice(invoiceId), {
        kind: 'tool',
        input: { invoiceId },
      });
      const decision = await run.step('Check approval policy', () => approveInvoice(invoice), {
        kind: 'step',
        input: { invoice },
      });
      run.artifact('Invoice decision', {
        kind: 'approval-decision',
        content: decision,
        businessArtifact: { source: 'erp:production:invoices', external_id: invoiceId },
      });
      return decision;
    },
    { runId: executionId }
  );
}

// Call after active workflows finish during graceful shutdown.
await fluency.shutdown();

The SDK reads FLUENCY_WORKFLOW_API_KEY automatically. You can pass token directly or provide a token callback when your deployment manages short-lived credentials.

Python quick start

import logging
import os

from fluency_workflow_telemetry import init

logger = logging.getLogger(__name__)
fluency = init(
  endpoint=os.environ["FLUENCY_WORKFLOW_ENDPOINT"],
  executor={
    "type": "automation",
    "id": "accounts-payable-workflow",
    "name": "Accounts payable workflow",
    "version": "2026.09",
  },
  on_error=lambda error: logger.warning("Fluency telemetry delivery failed: %s", error),
)

def process_invoice(invoice_id: str, execution_id: str):
  with fluency.run("Process supplier invoice", run_id=execution_id) as run:
    with run.step("Read invoice", kind="tool", input={"invoice_id": invoice_id}) as step:
      invoice = read_invoice(invoice_id)
      step.output = invoice
    with run.step("Check approval policy", kind="step", input={"invoice": invoice}) as step:
      decision = approve_invoice(invoice)
      step.output = decision
    run.artifact(
      "Invoice decision",
      kind="approval-decision",
      content=decision,
      business_artifact={"source": "erp:production:invoices", "external_id": invoice_id},
    )
    return decision

# Call after active workflows finish during graceful shutdown.
fluency.shutdown()

Python supports both with and async with for runs and steps.

What `run` captures automatically

The generic run wrapper always captures the execution name, executor, start and end time, duration, and completed or failed status. It cannot identify arbitrary function calls inside your code as tools because normal JavaScript and Python calls do not carry that meaning. You do not need to wrap every tool when the workflow framework exposes its execution tree:

LangChain: attach one Fluency callback to the outermost runnable. The callback captures nested chains, model calls, tools, and retrievers.
OpenTelemetry: submit one completed root trace. The adapter turns descendant spans into timed steps.
Custom code without framework callbacks or spans: wrap only the operations you want Fluency to display with step.

Fluency does not monkey patch global model, network, or tool functions. That keeps concurrent runs isolated and preserves accurate parent-child timing.

LangChain

Use executor type agent and attach one Fluency callback to the outermost runnable. LangChain chains, model calls, tools, and retrievers become timed steps automatically. The root output becomes an artifact. TypeScript:

import { createLangChainHandler } from '@fluency/workflow-telemetry/langchain';

const telemetry = createLangChainHandler({
  endpoint: process.env.FLUENCY_WORKFLOW_ENDPOINT!,
  executor: { type: 'agent', id: 'support-agent', name: 'Customer support agent' },
  onError: (error) => logger.warn({ error }, 'Fluency telemetry delivery failed'),
});
const result = await chain.invoke(input, { callbacks: [telemetry] });
await telemetry.shutdown();

Python:

pip install './fluency_workflow_telemetry-0.1.0-py3-none-any.whl[langchain]'

from fluency_workflow_telemetry.langchain import FluencyCallbackHandler

telemetry = FluencyCallbackHandler(
  endpoint=os.environ["FLUENCY_WORKFLOW_ENDPOINT"],
  executor={"type": "agent", "id": "support-agent", "name": "Customer support agent"},
)
result = chain.invoke(input, config={"callbacks": [telemetry]})
telemetry.shutdown()

Attach the callback at the root so it can determine when the complete execution ends. Fully consume or close streams before shutdown.

OpenTelemetry and other frameworks

The packages export normalizeCompletedTrace for TypeScript and normalize_completed_trace for Python. Use these adapters when your framework already produces completed OpenTelemetry spans. Select one root span and its descendants, normalize them into one run snapshot, then submit it with captureRun or capture_run. For frameworks without a native adapter, the explicit run, step, and artifact API works with any code. It does not require an LLM or an agent framework.

Process matching

Most integrations should omit processId or process_id. After accepting a run, Fluency considers:

an explicit valid process ID, when supplied by a Fluency-managed integration;
a unique normalized workflow-name match;
tenant-local LLM matching using the run name, steps, artifacts, and candidate process descriptions; and
creation or reuse of an automated workflow process when no match is safe.

The LLM must choose from candidate process IDs supplied by Fluency. Fluency validates the returned ID and confidence, and abstains on missing, conflicting, or ambiguous evidence. A matched human process keeps its existing process ID, so human and automated executions appear in the same process view with different actor labels. Process association is asynchronous. Successful delivery means Fluency stored the run; it may take a short time to appear in the product.

Artifacts shared with human work

An execution artifact is evidence from one run. A business artifact identifies the underlying invoice, ticket, order, claim, report, or other object across many human and automated executions. The most reliable identity is a source namespace plus the source system’s record ID:

run.artifact('Invoice decision', {
  kind: 'invoice',
  content: decision,
  businessArtifact: { source: 'erp:production:invoices', external_id: invoice.id },
});

run.artifact(
  "Invoice decision",
  kind="invoice",
  content=decision,
  business_artifact={"source": "erp:production:invoices", "external_id": invoice["id"]},
)

Fluency reuses the same business artifact when later executions submit the same source and external_id. Human and automated activity can therefore appear on one artifact history. When explicit identity is absent, Fluency may use exact structured identifiers such as invoice numbers, ticket IDs, or full record URLs. A bounded LLM match can consider tenant-local artifact evidence after exact checks. It must abstain when the evidence is weak or conflicting. Artifact display names alone are never treated as identity.

Stable run IDs and retries

Supply the execution ID from your queue, scheduler, or workflow engine as runId or run_id. The run ID is the idempotency key within the organization.

Retrying the same ID with identical evidence is safe.
Reusing an ID with different evidence is rejected.
A business retry that actually executes again needs a new run ID.

The SDK retries network errors, HTTP 429, and HTTP 5xx responses. A failed delivery remains in the in-memory queue for an explicit flush or shutdown retry. Delivery errors are reported through onError or on_error and never replace your workflow’s result or exception.

Failed runs

An exception escaping a recorded step marks the step and run as failed, records the end time, and rethrows the original exception. If your workflow handles a failure and returns normally, call run.fail() before returning. Execution status and Fluency processing status are separate. A failed business execution can still be successfully processed and shown in the product. If enrichment or matching fails repeatedly, the raw run remains stored and Fluency can retry its projection.

Content controls

Inputs, outputs, artifact content, URIs, and error messages are stored when supplied. This release does not redact them automatically. Set captureContent: false in TypeScript or capture_content=False in Python to keep names, IDs, status, and timing while omitting step input/output, artifact content/URI, and detailed error text. Stable business artifact identity remains available in metadata-only mode.

Runtime behavior

The SDK sends final run snapshots to POST /ext/workflow-telemetry/runs.
A run can contain at most 1,000 steps and 100 artifacts.
Default client payload limit is 1 MiB; the HTTP endpoint accepts up to 2 MiB.
Step intervals may overlap. Fluency computes automated handling time from their interval union so nested calls are not double-counted.
The queue is in memory. Call shutdown during graceful termination and inspect its boolean result.
The SDK does not fetch artifact URIs or modify workflow control flow.

For a wire-level integration, see Workflow Telemetry API.

Getting Started

IT Setup

Admin Setup

Work Explorer

Opportunities

SDKs

MCP

Product Updates

Fluency Workflow Telemetry SDK

Fluency Workflow Telemetry SDK

What Fluency provides

Install

Choose the executor type

TypeScript quick start

Python quick start

What `run` captures automatically

LangChain

OpenTelemetry and other frameworks

Process matching

Artifacts shared with human work

Stable run IDs and retries

Failed runs

Content controls

Runtime behavior

​Fluency Workflow Telemetry SDK

​What Fluency provides

​Install

​Choose the executor type

​TypeScript quick start

​Python quick start

​What run captures automatically

​LangChain

​OpenTelemetry and other frameworks

​Process matching

​Artifacts shared with human work

​Stable run IDs and retries

​Failed runs

​Content controls

​Runtime behavior

Fluency Workflow Telemetry SDK

What Fluency provides

Install

Choose the executor type

TypeScript quick start

Python quick start

What `run` captures automatically

LangChain

OpenTelemetry and other frameworks

Process matching

Artifacts shared with human work

Stable run IDs and retries

Failed runs

Content controls

Runtime behavior