> ## Documentation Index
> Fetch the complete documentation index at: https://docs.usefluency.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Fluency Workflow Telemetry SDK

# 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:

```sh theme={"system"}
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:

```sh theme={"system"}
# 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`.

```ts theme={"system"}
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

```python theme={"system"}
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:

```ts theme={"system"}
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:

```sh theme={"system"}
pip install './fluency_workflow_telemetry-0.1.0-py3-none-any.whl[langchain]'
```

```python theme={"system"}
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:

1. an explicit valid process ID, when supplied by a Fluency-managed integration;
2. a unique normalized workflow-name match;
3. tenant-local LLM matching using the run name, steps, artifacts, and candidate process descriptions; and
4. 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:

```ts theme={"system"}
run.artifact('Invoice decision', {
  kind: 'invoice',
  content: decision,
  businessArtifact: { source: 'erp:production:invoices', external_id: invoice.id },
});
```

```python theme={"system"}
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](/fluency-workflow-telemetry-api).