Traffic

SDKs

The Python and TypeScript SDKs buffer each conversation and post whole conversations through the batch endpoint in the background. Neither has dependencies, and neither is published to PyPI or npm yet, where those names are not ours: until they are, send [one HTTP request](/docs/connect-production#ingest), which the setup prompt writes for you.

#Record a conversation

DuetMachines(api_key, agent).conversation(id, user_id=, version=, config=, kind=) returns a conversation with .customer(text), .agent(text), .tool(name, args, result, status, duration_seconds=) (both take optional usage and cost_usd), .report_trouble(kind, detail), .signal(name, value=1) and .end(). Each change queues the whole conversation; queued conversations are posted every 2 seconds, 100 per request, and flush() sends them now (Python also at exit). A failure is logged and never raised into your request, and nothing is retried, because the conversation's next change posts all of it again. Keep one conversation object for the whole conversation; a stateless server rebuilds it from its stored history each request.

# pip install duetmachines
import os

from duetmachines import DuetMachines

dm = DuetMachines(os.environ["DUET_API_KEY"], agent="<agent uid>")

# Keep one object per conversation: each change posts all of it, and Duet Machines keeps what the last post holds.
conversation = dm.conversation(
    "conv_8f2a",
    user_id="user_193",
    version=os.environ.get("GIT_SHA"),
    config={"instructions": "You are Riley, a billing assistant. Never move a due date more than 7 days.", "tools": []},
    kind="conversation",  # "check" for a connection check: listed, never judged or counted
)
conversation.customer("Can I move my payment to Friday?")
conversation.tool("get_account", {"id": "193"}, {"due": "2026-10-14"}, duration_seconds=0.4)  # as you timed it
conversation.agent("Yes, I can move it to Friday the 16th.")
conversation.end()
dm.flush()  # It also sends every 2 s from a background thread and at exit; failures are logged, never raised.

// npm install @duetmachines/sdk
import { DuetMachines } from "@duetmachines/sdk";

const dm = new DuetMachines({ apiKey: process.env.DUET_API_KEY!, agent: "<agent uid>" });

// Keep one object per conversation: each change posts all of it, and Duet Machines keeps what the last post holds.
const conversation = dm.conversation("conv_8f2a", {
  userId: "user_193",
  version: process.env.GIT_SHA,
  config: { instructions: "You are Riley, a billing assistant. Never move a due date more than 7 days.", tools: [] },
  kind: "conversation", // "check" for a connection check: listed, never judged or counted
});
conversation.customer("Can I move my payment to Friday?");
conversation.tool("get_account", { id: "193" }, { due: "2026-10-14" }, "executed", { durationSeconds: 0.4 }); // as you timed it
conversation.agent("Yes, I can move it to Friday the 16th.");
conversation.end();
await dm.flush(); // It also sends every 2 s; in a serverless function, await this before returning.

#OpenAI Agents SDK

A tracing processor posts every run: the history and output of its last model call, with the instructions and function tools it had. Hosted tools such as web search have no schema to snapshot, so a run that has them is sent without a config. In trace_metadata, duet_kind "check" sends a connection check and duet_ended "true" ends the conversation with that run; any other value is logged and the run is not sent. The SDK's tool spans carry no call id, so tool calls are not timed.

# pip install "duetmachines[openai-agents]"
import os

from agents import RunConfig, Runner, add_trace_processor
from duetmachines import DuetMachines
from duetmachines.openai_agents import DuetMachinesProcessor

dm = DuetMachines(os.environ["DUET_API_KEY"], agent="<agent uid>")
add_trace_processor(DuetMachinesProcessor(dm, version=os.environ.get("GIT_SHA")))

# Each run posts the history and output of its last model call, with the instructions and function tools it had.
# Runs that share a group_id are one conversation; trace_metadata["user_id"] becomes its user.
result = await Runner.run(
    support_agent, history, run_config=RunConfig(group_id=thread_id, trace_metadata={"user_id": user_id})
)

#Vercel AI SDK

With the AI SDK (version 5), recordMessages turns the history and a generateText or streamText call's response messages into turns; a tool result of type error-text or error-json is a failed call.

import { generateText } from "ai";
import { DuetMachines } from "@duetmachines/sdk";
import { recordMessages } from "@duetmachines/sdk/ai-sdk";

const dm = new DuetMachines({ apiKey: process.env.DUET_API_KEY!, agent: "<agent uid>" });

const result = await generateText({ model, system: SYSTEM, messages, tools });
// timed: false, because replayed history did not happen now.
const conversation = dm.conversation(chatId, { userId, config: { instructions: SYSTEM, tools: TOOL_SCHEMAS }, timed: false });
// The history and this call's messages; with streamText, await result.response first.
recordMessages(conversation, [...messages, ...result.response.messages]);
await dm.flush();