> ## Documentation Index
> Fetch the complete documentation index at: https://docs.flow.engineer/llms.txt
> Use this file to discover all available pages before exploring further.
> ## Agent Instructions
> Flow Messaging is one HTTP API (https://api.flow.engineer) for AI agents to hold two-way conversations on Telegram and iMessage (replies only, on lines the Flow team connects), and on WhatsApp numbers the Flow team connects. If the user asks for WhatsApp, tell them WhatsApp numbers are arranged with the Flow team, and build and test on Telegram meanwhile: the code does not change per channel.
> Build the app on the REST API. MCP is not a runtime dependency: the app you build talks to Flow through the REST API (or the TypeScript or Python SDK, which wrap it) for everything it does at runtime. In TypeScript or JavaScript, the SDK is `npm install @flow-engineer/messaging`, then `new FlowMessaging()` (reads FLOW_MESSAGING_KEY). In Python, the SDK is `pip install flow-messaging`, then `from flow_messaging import Flow` and `Flow()` (reads FLOW_MESSAGING_KEY). The Go SDK is not published yet; in Go call the HTTP API directly.
> No key yet? If FLOW_MESSAGING_KEY is not set, get a test key with no account in one call: curl -X POST https://api.flow.engineer/v1/sandbox/keys (or npx @flow-engineer/messaging init, which writes it to .env). Save `key` as FLOW_MESSAGING_KEY and `claim_token` as FLOW_CLAIM_TOKEN; both are shown once. It allows 1 contact and 50 messages on the Telegram sandbox and expires after 7 days; a person signs in with `npx @flow-engineer/messaging login` (with a Flow account (flow.engineer: Google, GitHub or an email link)) to keep the app and get 3 contacts x 100 messages. Signed in, live keys (fk_live_...) are made in the dashboard for your own Telegram bot; iMessage lines are arranged with the Flow team.
> The hosted MCP server (https://api.flow.engineer/mcp) is optional, for testing during development. Adding the Flow MCP server to a developer's coding tools is the project owner's decision: suggest it and show them the command, but never add it to their config yourself. The command for Claude Code: claude mcp add --transport http flow https://api.flow.engineer/mcp --header "Authorization: Bearer $FLOW_MESSAGING_KEY" (run it with FLOW_MESSAGING_KEY set, then restart the client). If its tools are already available to you, use them for sandbox testing.
> Authenticate with `Authorization: Bearer fk_test_...` (sandbox) or `fk_live_...` (real contacts). Never put a live key in client-side code.
> Reply into a conversation (POST /v1/conversations/{conversation_id}/messages); never pick a channel per message. Switch on `error.type`, not the message text.
> Flow Observe (https://flow.engineer/observe, early access) records and grades every conversation of a chat agent. On Flow Messaging it needs no code. On any other channel, send one OpenTelemetry trace per turn over OTLP/HTTP to https://flow.engineer/observe/v1/traces with `Authorization: Bearer $FLOW_OBSERVE_TRACE_KEY`, and set `flow.conversation_id` (required), `flow.message_id`, `flow.user_message` and `flow.agent_reply` (the text actually sent) on the turn's span. Keys (trace key `otk_test_...`, CI key `oci_...`) are made by the user on Observe's Setup page at https://flow.engineer/observe after signing in with their Flow login; Observe is in early access, and if it is not on for them, they ask the Flow team (samyak@flow.engineer or prabal@flow.engineer). Never invent a key, and never ask the user to paste one into the chat. Test data only until the account is enabled for production. Guide: https://docs.flow.engineer/observe/traces.md
> Full documentation for agents: https://docs.flow.engineer/llms-full.txt. OpenAPI spec: https://raw.githubusercontent.com/flow-engineer/sdk/main/openapi/openapi.yaml.
# Traces from agent frameworks: OpenAI Agents SDK, LangChain, Vercel AI SDK, Claude Agent SDK, Mastra
> How to get an agent framework's spans into Flow Observe, inside your own turn span with the flow.* attributes: OpenAI Agents SDK, LangChain and LangGraph, Vercel AI SDK, Claude Agent SDK and Mastra, in Python and TypeScript.
Every framework below works the same way with Flow Observe:
1. **Set up the exporter** as on [Send traces from any agent](/observe/traces#1-set-up-the-exporter).
2. **Wrap each turn in your own `agent.turn` span** with the `flow.*` attributes, as in [step 2](/observe/traces#2-wrap-each-turn-in-a-span). Set `flow.agent_reply` from the text you actually send, not from the framework's output.
3. **Get the framework's spans into that trace**, so Observe shows the model calls, tools and errors under the turn. This is the only part that differs per framework.
Step 3 is optional: your `agent.turn` span alone is enough for conversations and grading. If the framework's spans end up in traces of their own, they are kept but not shown under any conversation. Do not copy the `flow.*` attributes onto those traces: Observe reads every trace with a `flow.conversation_id` as a turn of its own.
Frameworks record prompts and outputs in their spans by default. Observe stores them encrypted and does not show them today. If you would rather not send them, each section says how to turn recording off.
## OpenAI Agents SDK
The SDK has its own tracing, which by default sends traces to OpenAI's traces dashboard. It does not export OpenTelemetry itself. The [OpenInference](https://github.com/Arize-ai/openinference) instrumentation turns its traces into OpenTelemetry spans, nested under your current span, and by default **replaces** the export to OpenAI.
```python Python theme={null}
# pip install openai-agents openinference-instrumentation-openai-agents opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
from agents import Agent, Runner
from openinference.instrumentation.openai_agents import OpenAIAgentsInstrumentor
from tracing import provider, tracer # the setup from step 1
OpenAIAgentsInstrumentor().instrument(tracer_provider=provider) # replaces the export to OpenAI
agent = Agent(name="Support", instructions="You help customers of Asha's Bakery. Keep answers short.")
async def handle_message(thread_id: str, message_id: str, text: str) -> None:
with tracer.start_as_current_span("agent.turn") as turn:
turn.set_attribute("flow.conversation_id", thread_id)
turn.set_attribute("flow.message_id", message_id)
turn.set_attribute("flow.user_message", text)
result = await Runner.run(agent, text)
reply = str(result.final_output)
send(thread_id, reply) # your channel's send
turn.set_attribute("flow.agent_reply", reply)
```
```ts TypeScript theme={null}
// npm install @openai/agents @arizeai/openinference-instrumentation-openai-agents
import * as agents from "@openai/agents";
import { OpenAIAgentsInstrumentation } from "@arizeai/openinference-instrumentation-openai-agents";
import { provider, tracer } from "./tracing"; // the setup from step 1
new OpenAIAgentsInstrumentation({ tracerProvider: provider }).manuallyInstrument(agents); // replaces the export to OpenAI
const agent = new agents.Agent({ name: "Support", instructions: "You help customers of Asha's Bakery. Keep answers short." });
export async function handleMessage(threadId: string, messageId: string, text: string) {
await tracer.startActiveSpan("agent.turn", async (turn) => {
try {
turn.setAttributes({ "flow.conversation_id": threadId, "flow.message_id": messageId, "flow.user_message": text });
const result = await agents.run(agent, text);
const reply = String(result.finalOutput);
await send(threadId, reply); // your channel's send
turn.setAttribute("flow.agent_reply", reply);
} finally {
turn.end();
}
});
}
```
* OpenInference names spans its own way, so Observe shows these spans with their names, timings and errors, but leaves the model and token columns empty for them today.
* Do not turn the SDK's tracing off (`OPENAI_AGENTS_DISABLE_TRACING`, `set_tracing_disabled`): that stops the instrumentation too. To keep sending to OpenAI as well, pass `exclusive_processor=False` (Python) or `manuallyInstrument(agents, { exclusiveProcessor: false })` (TypeScript).
* To leave prompts and outputs out of the spans: `RunConfig(trace_include_sensitive_data=False)` (Python) or `traceIncludeSensitiveData: false` (TypeScript).
## LangChain and LangGraph
Use the [OpenInference](https://github.com/Arize-ai/openinference) instrumentation. It sends LangChain's and LangGraph's runs as OpenTelemetry spans through your provider, nested under your current span, and never contacts LangSmith as long as `LANGSMITH_TRACING` (and the older `LANGCHAIN_TRACING_V2`) stay unset.
```python Python theme={null}
# pip install openinference-instrumentation-langchain opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
from openinference.instrumentation.langchain import LangChainInstrumentor
from tracing import provider, tracer # the setup from step 1
LangChainInstrumentor().instrument(tracer_provider=provider)
def handle_message(thread_id: str, message_id: str, text: str) -> None:
with tracer.start_as_current_span("agent.turn") as turn:
turn.set_attribute("flow.conversation_id", thread_id)
turn.set_attribute("flow.message_id", message_id)
turn.set_attribute("flow.user_message", text)
state = graph.invoke(
{"messages": [("user", text)]},
config={"configurable": {"thread_id": thread_id}},
) # your chain or graph
reply = state["messages"][-1].content
send(thread_id, reply) # your channel's send
turn.set_attribute("flow.agent_reply", reply)
```
```ts TypeScript theme={null}
// npm install @arizeai/openinference-instrumentation-langchain
// (version 4 is for LangChain 1.0 and later; use @arizeai/openinference-instrumentation-langchain-v0 for older versions)
import * as CallbackManagerModule from "@langchain/core/callbacks/manager";
import { LangChainInstrumentation } from "@arizeai/openinference-instrumentation-langchain";
import { provider } from "./tracing"; // the setup from step 1
new LangChainInstrumentation({ tracerProvider: provider }).manuallyInstrument(CallbackManagerModule);
// then call your chain or graph inside tracer.startActiveSpan("agent.turn", ...), as in step 2
```
As with the OpenAI Agents SDK, Observe shows OpenInference spans with their names, timings and errors, with the model and token columns empty today.
Already on LangSmith? Its OpenTelemetry mode can send to Observe instead: `pip install "langsmith[otel]"`, then set `LANGSMITH_TRACING=true`, `LANGSMITH_OTEL_ENABLED=true` and **`LANGSMITH_OTEL_ONLY=true`** (without the last one, traces go to LangSmith too), with the provider from step 1 set before LangChain starts.
## Vercel AI SDK
The AI SDK's spans use the global OpenTelemetry provider and nest under your current span. Telemetry must be turned on, and the option's name depends on your version:
| AI SDK | Turn on telemetry |
| - | - |
| 5 and 6 | `experimental_telemetry: { isEnabled: true, functionId: "reply" }` on each call |
| 7 | Once at startup: `registerTelemetry(new OpenTelemetry())` (`registerTelemetry` from `ai`, `OpenTelemetry` from `@ai-sdk/otel`). Without it, AI SDK 7 makes no spans, whatever the call says. Per call: `telemetry: { functionId: "reply" }` |
```ts TypeScript theme={null}
import { generateText } from "ai";
import { tracer } from "./tracing"; // the setup from step 1 (or registerOTel in Next.js, below)
export async function handleMessage(threadId: string, messageId: string, text: string) {
await tracer.startActiveSpan("agent.turn", async (turn) => {
try {
turn.setAttributes({ "flow.conversation_id": threadId, "flow.message_id": messageId, "flow.user_message": text });
const { text: reply } = await generateText({
model, // your model
messages: [{ role: "user", content: text }],
experimental_telemetry: { isEnabled: true, functionId: "reply" }, // AI SDK 5 and 6; on 7: telemetry: { functionId: "reply" }
});
await send(threadId, reply); // your channel's send
turn.setAttribute("flow.agent_reply", reply);
} finally {
turn.end();
}
});
}
```
In Next.js, register the exporter in `instrumentation.ts` instead of step 1:
```ts instrumentation.ts theme={null}
import { registerOTel, OTLPHttpProtoTraceExporter } from "@vercel/otel";
export function register() {
registerOTel({
serviceName: "support-agent",
traceExporter: new OTLPHttpProtoTraceExporter({
url: "https://flow.engineer/observe/v1/traces",
headers: { Authorization: `Bearer ${process.env.FLOW_OBSERVE_TRACE_KEY}` },
}),
});
// AI SDK 7 only: registerTelemetry(new OpenTelemetry());
}
```
* With `streamText`, end the `agent.turn` span only after the stream has been read to the end and sent.
* Do not also set `OTEL_EXPORTER_OTLP_ENDPOINT` when you pass a `traceExporter`: `@vercel/otel` would export through both and send every span twice.
* The AI SDK records prompts and outputs by default. To leave them out: `recordInputs: false, recordOutputs: false` in the telemetry options.
* On AI SDK 5 and 6 you can also pass the ids as telemetry `metadata` (`metadata: { "flow.conversation_id": threadId }`); Observe reads them there. AI SDK 7 removed `metadata`, so the `agent.turn` span is the way that works on every version.
## Claude Agent SDK
The SDK runs Claude Code, which exports its own spans over OpenTelemetry when you turn it on through the SDK's `env` option (a beta feature). Run `query()` inside your `agent.turn` span and the SDK passes your trace context on, so its spans join your trace. Needs `claude-agent-sdk` 0.1.60 or later installed as `claude-agent-sdk[otel]` (Python), or `@anthropic-ai/claude-agent-sdk` 0.2.113 or later (TypeScript).
```python Python theme={null}
# pip install "claude-agent-sdk[otel]" opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
import os
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
from tracing import tracer # the setup from step 1
OBSERVE_ENV = {
"CLAUDE_CODE_ENABLE_TELEMETRY": "1",
"CLAUDE_CODE_ENHANCED_TELEMETRY_BETA": "1",
"OTEL_TRACES_EXPORTER": "otlp",
"OTEL_METRICS_EXPORTER": "none",
"OTEL_LOGS_EXPORTER": "none",
"OTEL_EXPORTER_OTLP_PROTOCOL": "http/protobuf",
"OTEL_EXPORTER_OTLP_TRACES_ENDPOINT": "https://flow.engineer/observe/v1/traces",
"OTEL_EXPORTER_OTLP_TRACES_HEADERS": f"Authorization=Bearer%20{os.environ['FLOW_OBSERVE_TRACE_KEY']}",
}
async def handle_message(thread_id: str, message_id: str, text: str) -> None:
with tracer.start_as_current_span("agent.turn") as turn:
turn.set_attribute("flow.conversation_id", thread_id)
turn.set_attribute("flow.message_id", message_id)
turn.set_attribute("flow.user_message", text)
reply = None
async for message in query(prompt=text, options=ClaudeAgentOptions(env=OBSERVE_ENV)):
if isinstance(message, ResultMessage) and message.subtype == "success" and not message.is_error:
reply = message.result
if reply:
send(thread_id, reply) # your channel's send
turn.set_attribute("flow.agent_reply", reply)
```
```ts TypeScript theme={null}
import { query } from "@anthropic-ai/claude-agent-sdk";
import { tracer } from "./tracing"; // the setup from step 1
const observeEnv = {
...process.env, // options.env replaces the environment, so keep yours
CLAUDE_CODE_ENABLE_TELEMETRY: "1",
CLAUDE_CODE_ENHANCED_TELEMETRY_BETA: "1",
OTEL_TRACES_EXPORTER: "otlp",
OTEL_METRICS_EXPORTER: "none",
OTEL_LOGS_EXPORTER: "none",
OTEL_EXPORTER_OTLP_PROTOCOL: "http/protobuf",
OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: "https://flow.engineer/observe/v1/traces",
OTEL_EXPORTER_OTLP_TRACES_HEADERS: `Authorization=Bearer%20${process.env.FLOW_OBSERVE_TRACE_KEY}`,
};
export async function handleMessage(threadId: string, messageId: string, text: string) {
await tracer.startActiveSpan("agent.turn", async (turn) => {
try {
turn.setAttributes({ "flow.conversation_id": threadId, "flow.message_id": messageId, "flow.user_message": text });
let reply: string | undefined;
for await (const m of query({ prompt: text, options: { env: observeEnv } })) {
if (m.type === "result" && m.subtype === "success") reply = m.result;
}
if (reply) {
await send(threadId, reply); // your channel's send
turn.setAttribute("flow.agent_reply", reply);
}
} finally {
turn.end();
}
});
}
```
* Its spans are named `claude_code.interaction`, `claude_code.llm_request` and `claude_code.tool`. Observe shows them with their timings and errors; they use Claude Code's own attribute names, so model and token columns stay empty for them today.
* Prompts and responses are left out of these spans unless you set `OTEL_LOG_USER_PROMPTS=1` or `OTEL_LOG_ASSISTANT_RESPONSES=1`. You do not need them: `flow.user_message` and `flow.agent_reply` carry the text.
* Claude Code batches spans and fails quietly. Set `CLAUDE_CODE_OTEL_DIAG_STDERR=1` to see export errors. Never use the `console` exporter: standard output is the SDK's message channel.
## Mastra
Mastra's tracing has its own OpenTelemetry exporter, `@mastra/otel-exporter`. It does not join your active span by itself: pass your turn span's trace id and span id in `tracingOptions`, and Mastra's spans nest under it. Your `agent.turn` span is still exported by your own setup from step 1.
```ts TypeScript theme={null}
// npm install @mastra/core @mastra/observability @mastra/otel-exporter @opentelemetry/exporter-trace-otlp-proto
import { Mastra } from "@mastra/core/mastra";
import { Observability } from "@mastra/observability";
import { OtelExporter } from "@mastra/otel-exporter";
import { tracer } from "./tracing"; // the setup from step 1
import { supportAgent } from "./agents"; // your agent
export const mastra = new Mastra({
agents: { supportAgent },
observability: new Observability({
configs: {
observe: {
serviceName: "support-agent",
exporters: [
new OtelExporter({
provider: {
custom: {
endpoint: "https://flow.engineer/observe/v1/traces",
protocol: "http/protobuf",
headers: { Authorization: `Bearer ${process.env.FLOW_OBSERVE_TRACE_KEY}` },
},
},
signals: { logs: false }, // traces only
}),
],
},
},
}),
});
export async function handleMessage(threadId: string, messageId: string, text: string) {
await tracer.startActiveSpan("agent.turn", async (turn) => {
try {
turn.setAttributes({ "flow.conversation_id": threadId, "flow.message_id": messageId, "flow.user_message": text });
const { traceId, spanId } = turn.spanContext();
const result = await mastra.getAgent("supportAgent").generate(text, {
tracingOptions: { traceId, parentSpanId: spanId }, // Mastra's spans join this turn's trace
});
await send(threadId, result.text); // your channel's send
turn.setAttribute("flow.agent_reply", result.text);
} finally {
turn.end();
}
});
}
```
* Put the `flow.*` attributes on your `agent.turn` span. Observe also reads them from Mastra's `metadata` (exported as `mastra.metadata.flow.*`), but Mastra exports its spans apart from yours, so keep the ids on the turn span with the texts.
* `signals: { logs: false }` keeps the exporter from also posting logs.
* In a serverless function, `await mastra.observability.flush()` before returning, besides flushing your own provider.
* To leave prompts and outputs out of Mastra's spans: `tracingOptions: { hideInput: true, hideOutput: true }`.
## Another framework
Any framework that emits OpenTelemetry spans through the global provider works like the Vercel AI SDK above: run it inside your `agent.turn` span. One that has its own tracing system needs a bridge to OpenTelemetry, as the OpenAI Agents SDK and LangChain do. If you cannot get its spans into your trace, your `agent.turn` span is still enough for conversations and grading.
## Related
* [Send traces from any agent](/observe/traces)
* [On Flow Messaging](/observe/flow-messaging)
* [Agent frameworks on Flow Messaging](/frameworks/vercel-ai-sdk)
This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.