Implementation:Helicone Helicone HeliconeProducer SendMessage

Knowledge Sources	Helicone
Domains	LLM Observability, Message Queuing, Asynchronous Processing
Last Updated	2026-02-14 00:00 GMT

Overview

Concrete message publishing method for enqueuing structured log data from the Helicone Cloudflare Worker proxy to a backend message queue, provided by the HeliconeProducer class. This class wraps Upstash Kafka, AWS SQS, or a direct HTTP fallback, depending on environment configuration.

Description

HeliconeProducer.sendMessage is an asynchronous method that publishes a MessageData payload to the configured message queue for downstream consumption by the Jawn backend. The HeliconeProducer constructor uses MessageProducerFactory.createProducer to select the appropriate transport based on the QUEUE_PROVIDER environment variable:

• "sqs" -- Uses SQSProducerImpl to publish to AWS SQS.
• "dual" -- Uses DualWriteProducer which writes to both Kafka and SQS simultaneously for migration safety.
• Default -- Uses KafkaProducerImpl to publish to Upstash Kafka (requires UPSTASH_KAFKA_URL, UPSTASH_KAFKA_USERNAME, UPSTASH_KAFKA_PASSWORD).

If no queue producer can be constructed (missing credentials), or if the message contains a heliconeManualAccessKey matching the environment's HELICONE_MANUAL_ACCESS_KEY, the method falls back to a synchronous HTTP POST to VALHALLA_URL/v1/log/request.

Usage

Use HeliconeProducer.sendMessage from within the async logging pipeline of the proxy worker to publish log records after the response has been returned to the client. This is typically called inside a ctx.waitUntil block.

Code Reference

Source Location

• Repository: Helicone
• File: worker/src/lib/clients/producers/HeliconeProducer.ts (lines 46-56)

Signature

async sendMessage(msg: MessageData): Promise<void>

Import

import { HeliconeProducer } from "../lib/clients/producers/HeliconeProducer";

I/O Contract

Inputs

Name	Type	Required	Description
msg	MessageData	Yes	The structured log message containing: id (string) -- Unique message identifier. authorization (string) -- The Helicone auth token for the request. heliconeMeta (HeliconeMeta) -- Metadata including model override, omit flags, webhook enabled, prompt ID/version, PostHog/Lytix keys, gateway metadata, and billing flags. log (Log) -- The request/response log data containing request ID, user ID, properties, provider, target URL, body size, timestamps, streaming flag, and response status, token counts, latency, cost, and time-to-first-token.

Outputs

Name	Type	Description
(void)	Promise<void>	The method returns void on success. If the underlying queue producer or HTTP fallback fails, errors are logged to the console but not propagated to the caller.

Usage Examples

Basic Usage

import { HeliconeProducer } from "../lib/clients/producers/HeliconeProducer";
import { MessageData } from "../lib/clients/producers/types";

const producer = new HeliconeProducer(env);

const msg: MessageData = {
  id: crypto.randomUUID(),
  authorization: "Bearer sk-helicone-...",
  heliconeMeta: {
    omitRequestLog: false,
    omitResponseLog: false,
    webhookEnabled: false,
  },
  log: {
    request: {
      id: requestId,
      userId: "user-123",
      properties: { "Helicone-Session-Id": "sess-1" },
      targetUrl: "https://api.openai.com/v1/chat/completions",
      provider: "OPENAI",
      bodySize: 1024,
      path: "/v1/chat/completions",
      requestCreatedAt: new Date(),
      isStream: false,
    },
    response: {
      id: responseId,
      status: 200,
      bodySize: 2048,
      responseCreatedAt: new Date(),
      delayMs: 350,
    },
  },
};

await producer.sendMessage(msg);

Related Pages

Implements Principle

• Principle:Helicone_Helicone_Asynchronous_Log_Queuing

Requires Environment

• Environment:Helicone_Helicone_Cloudflare_Workers_Runtime