Send traces to Observe

Export one OpenTelemetry trace per turn from an agent on any channel.

If your agent talks to people through anything other than Flow Messaging (your own SMS provider, a web chat, your own WhatsApp integration), your traces are all Flow Observe needs. Send one OpenTelemetry trace per turn, the work between a person’s message and your agent’s reply, with a few flow.* attributes on it. From each trace Observe builds the conversation, stores both messages and grades the reply.

Any OpenTelemetry SDK works. You need a trace key (otk_test_...), made in the dashboard (see Get access), in FLOW_OBSERVE_TRACE_KEY.


Endpoint	`POST https://flow.engineer/observe/v1/traces`
Protocol	OTLP over HTTP: protobuf (`application/x-protobuf`) or JSON (`application/json`), gzip or none. Not gRPC
Auth	`Authorization: Bearer $FLOW_OBSERVE_TRACE_KEY`

1. Set up the exporter

Once, when your process starts (the other examples import this file as tracing):

Python
TypeScript

# tracing.py
# pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
import os
from opentelemetry import trace
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

provider = TracerProvider(resource=Resource.create({
    "service.name": "support-agent",
    "service.version": os.environ.get("AGENT_VERSION", "dev"),  # shown as the agent's version
}))
provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter(
    endpoint="https://flow.engineer/observe/v1/traces",  # the full URL, path included
    headers={"Authorization": f"Bearer {os.environ['FLOW_OBSERVE_TRACE_KEY']}"},
)))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("support-agent")

// tracing.ts
// npm install @opentelemetry/api @opentelemetry/sdk-trace-node @opentelemetry/exporter-trace-otlp-proto @opentelemetry/resources
import { trace } from "@opentelemetry/api";
import { NodeTracerProvider, BatchSpanProcessor } from "@opentelemetry/sdk-trace-node";
import { OTLPTraceExporter } from "@opentelemetry/exporter-trace-otlp-proto";
import { resourceFromAttributes } from "@opentelemetry/resources";

export const provider = new NodeTracerProvider({
  resource: resourceFromAttributes({
    "service.name": "support-agent",
    "service.version": process.env.AGENT_VERSION ?? "dev", // shown as the agent's version
  }),
  spanProcessors: [
    new BatchSpanProcessor(
      new OTLPTraceExporter({
        url: "https://flow.engineer/observe/v1/traces", // the full URL, path included
        headers: { Authorization: `Bearer ${process.env.FLOW_OBSERVE_TRACE_KEY}` },
      }),
    ),
  ],
});
provider.register(); // the global provider, so framework spans use it too
export const tracer = trace.getTracer("support-agent");

The TypeScript example is for OpenTelemetry JS 2.x. Already using OpenTelemetry? Add the exporter above to your existing provider instead of making a second one. In an ES module project compiled with "module": "nodenext", write the local imports on this page with the extension: ./tracing.js.

Prefer environment variables? OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=https://flow.engineer/observe/v1/traces and OTEL_EXPORTER_OTLP_TRACES_HEADERS=Authorization=Bearer%20<key> (the space written as %20) configure an exporter created with no arguments.

2. Wrap each turn in a span

Start a span when a person’s message reaches your code, run your agent inside it, and report what you sent. Replace run_agent, split_for_sms and send_sms with your own code:

Python
TypeScript

from tracing import tracer

def handle_message(thread_id: str, from_number: str, message_id: str, text: str) -> None:
    """Call this when your channel delivers a message (your SMS provider's webhook, your web chat)."""
    with tracer.start_as_current_span("agent.turn") as turn:
        turn.set_attribute("flow.conversation_id", thread_id)
        turn.set_attribute("flow.contact_id", from_number)  # a phone number is fine: stored as a keyed hash
        turn.set_attribute("flow.message_id", message_id)   # your provider's id for this message
        turn.set_attribute("flow.channel", "sms")
        turn.set_attribute("flow.user_message", text)

        answer = run_agent(thread_id, text)  # your agent; spans it makes join this trace

        for part in split_for_sms(answer):   # what you actually send, after any post-processing
            with tracer.start_as_current_span("send") as send:
                send_sms(to=from_number, body=part)
                send.set_attribute("flow.agent_reply", part)  # set only once the send succeeded

import { tracer } from "./tracing";
import { runAgent, splitForSms, sendSms } from "./agent"; // your own code

export async function handleMessage(threadId: string, fromNumber: string, messageId: string, text: string) {
  // Call this when your channel delivers a message (your SMS provider's webhook, your web chat).
  await tracer.startActiveSpan("agent.turn", async (turn) => {
    try {
      turn.setAttributes({
        "flow.conversation_id": threadId,
        "flow.contact_id": fromNumber, // a phone number is fine: stored as a keyed hash
        "flow.message_id": messageId, // your provider's id for this message
        "flow.channel": "sms",
        "flow.user_message": text,
      });

      const answer = await runAgent(threadId, text); // your agent; spans it makes join this trace

      for (const part of splitForSms(answer)) { // what you actually send, after any post-processing
        await tracer.startActiveSpan("send", async (send) => {
          try {
            await sendSms(fromNumber, part);
            send.setAttribute("flow.agent_reply", part); // set only once the send succeeded
          } finally {
            send.end();
          }
        });
      }
    } finally {
      turn.end();
    }
  });
}

A reply sent as one message can carry flow.agent_reply on the agent.turn span itself; the send spans are for a reply sent as several messages.

In a short-lived process (a script, a serverless function), flush before it ends: provider.force_flush() in Python, await provider.forceFlush() in TypeScript. Otherwise the batch still in memory is lost.

The attributes

Put them on your turn span as in step 2, except flow.agent_reply, which goes on the turn span or on a span inside it. Observe also reads them from any other span of the trace or from the resource, and applies them to the whole trace, but the turn span is the safe place for the ids: it ends last, so when it arrives, the whole turn has.

Attribute	Value
`flow.conversation_id`	Your own id for the conversation or thread: a string (or an integer) of 1 to 256 characters. A longer one is ignored, and the turn is then not stored. Keep it the same for every turn of the thread.	Required
`flow.message_id`	Your id for the person’s message (for example your SMS provider’s). The person’s message is stored once per id: a second turn with the same id (a webhook delivered twice) adds no second copy of it, only the replies that turn sent. Default: a new id for every trace.	Recommended
`flow.user_message`	The person’s message, as received.	Recommended
`flow.agent_reply`	A message your agent sent, exactly as sent: after any post-processing, guardrails or splitting. One per span; several spans make several messages, in the order the spans end.	Recommended
`flow.contact_id`	Your own id for the person. Default: the conversation id.	Optional
`flow.channel`	`sms`, `mms`, `whatsapp`, `telegram`, `imessage`, `web`, `voice` or `other` (case does not matter; any other value counts as `other`). Default: `other`.	Optional
`flow.agent_version`	Your agent’s version (prompt, model, tools), up to 128 of `A-Z a-z 0-9 . _ : / + -`. Default: the resource’s `service.version`.	Optional

Ids are stored only as keyed hashes and never shown. A phone number is a fine conversation or contact id. Never put message text, names or email addresses in an id.
Text is stored encrypted, like every message in Observe, and deleted when your account’s retention period ends.
Times come from the spans. The person’s message is dated at the start of the span that carries flow.user_message, and each reply at the end of the span that carries its flow.agent_reply. The difference is the reply time Observe shows. So start the turn span when the message arrives, not after a queue, and end each send span when the send returns.
Conversations follow a 48-hour rule. After 48 hours with no messages, the next message starts a new conversation, even with the same flow.conversation_id. Billing counts conversations the same way.
A turn may have one side only. A reminder your agent starts has flow.agent_reply and no flow.user_message; a message your agent does not answer has flow.user_message and no flow.agent_reply. Both are stored as they are.
Without flow.user_message or flow.agent_reply, Observe reads the text from your trace’s agent span (the first span with gen_ai.operation.name invoke_agent, else your root span): the last user message of its input and the text of its output, when your framework records them. The model’s output is often not what was sent, so set both yourself.
The same names also work with the prefixes ai.settings.context., ai.telemetry.metadata. and mastra.metadata. (for example ai.telemetry.metadata.flow.conversation_id), which some frameworks add to metadata you pass them. No other names are read: gen_ai.conversation.id, enduser.id and the session ids frameworks set themselves are ignored.

3. Check it worked

Send one synthetic turn by hand. This needs nothing but curl and your key, and sends one span in OTLP/JSON:

NOW=$(date +%s); TRACE=$(openssl rand -hex 16); SPAN=$(openssl rand -hex 8)
curl -sS -w '\nHTTP %{http_code}\n' https://flow.engineer/observe/v1/traces \
  -H "Authorization: Bearer $FLOW_OBSERVE_TRACE_KEY" -H "Content-Type: application/json" \
  -d @- <<EOF
{"resourceSpans": [{
  "resource": {"attributes": [{"key": "service.name", "value": {"stringValue": "curl-check"}}]},
  "scopeSpans": [{"spans": [{
    "traceId": "$TRACE", "spanId": "$SPAN", "name": "agent.turn", "kind": 1,
    "startTimeUnixNano": "$((NOW - 2))000000000", "endTimeUnixNano": "${NOW}000000000",
    "attributes": [
      {"key": "flow.conversation_id", "value": {"stringValue": "curl-check-$TRACE"}},
      {"key": "flow.message_id", "value": {"stringValue": "curl-check-$SPAN"}},
      {"key": "flow.channel", "value": {"stringValue": "sms"}},
      {"key": "flow.user_message", "value": {"stringValue": "Do you deliver on Sundays?"}},
      {"key": "flow.agent_reply", "value": {"stringValue": "Not yet: we deliver Monday to Saturday."}}
    ]
  }]}]
}]}
EOF

Answer	Meaning
`200` with `{}`	Stored.
`200` with `partialSuccess`	Some spans were refused; `rejectedSpans` counts them and `errorMessage` says why (bad ids, missing or future times). The rest were stored.
`401`	No key, or a wrong or revoked one.
`403`	A live key (`otk_live_...`) on an account that takes test data only.
`400`, `413`, `415`	Not OTLP, too large, or another content type. Use an OTLP/HTTP exporter (not gRPC).
`429`, `503`	Too many requests, or Observe briefly unavailable; wait `Retry-After` (exporters do this themselves).

Then run a turn through your own code and look in the dashboard: your agent, then Conversations. The conversation appears within seconds with both messages and the reply time; open it to see the turn’s spans under the reply. The reply’s verdict appears shortly after. OpenTelemetry exporters report failures in your process’s log, not as exceptions, so check the log if nothing appears.

Common mistakes

Ids on only some traces of a turn. Some frameworks and workflow engines run one turn as several traces (each step, retry or background job starts its own). A trace without flow.conversation_id is kept but belongs to no conversation, and on a test-only account its content is deleted after 30 minutes. Keep the turn in one trace: start your framework’s work inside the agent.turn span, and pass the trace context into queued jobs. Do not copy the ids onto the other traces instead: Observe reads every trace with a flow.conversation_id as a turn of its own, so the person’s message or the reply would be stored twice.
Reporting the model’s output instead of what was sent. Post-processing, guardrails, a fallback message, truncation or splitting change the text. Set flow.agent_reply to the text your channel accepted, after all of that. A turn that sent nothing has no flow.agent_reply.
Text or personal data in ids or span names. Ids are hashed, but span names, tool names and service names are stored as plain text (long digit runs and anything with @ are masked, as a safety net). Name spans after what they do (agent.turn, send, lookup_order), never after what was said.
A new conversation id every turn. Each turn then becomes its own conversation. Use the thread’s id (for SMS, the person’s number, or the pair of your number and theirs).
Reusing a message id in tests. A second turn with the same flow.message_id is treated as a retry: its message from the person is not stored again, and its replies are added to the first turn. Make a new id per test turn.
A turn span that never reaches Observe. Observe stores the person’s message when its span arrives, and your replies once that message or your trace’s root span has arrived, so each reply gets its reply time even when your exporter sends it first. If your turn span’s parent lives in a service that does not export to Observe, the root never arrives: then a trace with no flow.user_message keeps no replies. Start the turn span in the service that receives the message, and keep the ids and both texts on it or inside it.
Not flushing. A script or serverless function that exits before the batch is exported sends nothing.
A gRPC exporter, or a URL without /v1/traces. Use the HTTP exporter, and the full URL when you pass it in code.
The trace key in a browser or mobile app. Keep it on your server.
Real people’s messages before your account is enabled for production. Send synthetic conversations only until the Flow team enables it.

Limits

Trace size and ingestion limits

At most 8 MiB per request as sent (32 MiB after gzip), 10,000 spans per request, and 256 KiB of content per span (longer content is cut).
Per key: 10 requests a second, bursts of 40, at most 8 at once. Above that, 429 with Retry-After. A batching exporter sends about one request every few seconds, far below this.
A span dated more than 5 minutes ahead of Observe’s clock is refused.
A span is stored once per trace id and span id, so an exporter’s retry changes nothing.

What Observe keeps from your spans

Stored span fields and content

Kept as metadata, and shown in the dashboard: span names, kinds (model, tool, retrieval, agent), timings, models and token counts, tool names, errors and retries, following the OpenTelemetry GenAI conventions and the common framework spellings.
Stored encrypted, not shown today: prompts, model outputs, tool inputs and results, and the other content your framework records, plus every span event (exception messages included). If you do not want your framework’s recorded prompts and outputs sent, turn recording off in the framework; flow.user_message and flow.agent_reply are enough for grading.
Dropped: every other attribute, including your own custom ones.

Was this page helpful?