SDK reference

Every function and option in the TypeScript and Python SDKs, generated from their source.

The TypeScript and Python SDKs send model calls, traces and tool spans from your app to a coding.sh server. This page is generated from the SDK source.

Both SDKs send to a server you run with coding.sh serve (production guide, serverless apps). Without CODING_SH_URL they do nothing. They never change return values, errors or stream behavior, and never throw into your app. The SDKs do not price calls; the server does.

Environment variables

Both SDKs read the same variables. Options passed to init() win over them. Every other variable coding.sh reads is in the configuration reference.

TypeScript: @codingdotsh/sdk

Install

npm install @codingdotsh/sdk
export CODING_SH_URL=http://127.0.0.1:4318   # where coding.sh serve listens
export CODING_SH_SERVICE=support-agent

Node >=18. No runtime dependencies. Add CODING_SH_INGEST_KEY when your server requires ingest keys.

Minimal use

import OpenAI from "openai";
import { observe } from "@codingdotsh/sdk";

const openai = observe(new OpenAI());
const reply = await openai.chat.completions.create({ model: "gpt-4o-mini", messages: [{ role: "user", content: "Where is my order?" }] });

observe() also takes an Anthropic client. Each model call becomes one session with its tokens, model, latency, time to first token, stop reason and tool calls. The server prices it.

Traces, tools and conversations

import { span, trace } from "@codingdotsh/sdk";

const answer = await trace(
  "support-reply",
  async () => {
    const order = await span("tool", "lookup_order", () => orders.get(42), { input: { id: 42 } });
    const res = await openai.chat.completions.create({
      model: "gpt-4o-mini",
      messages: [{ role: "user", content: `${question} Order status: ${order.status}` }],
    });
    return res.choices[0].message.content;
  },
  { sessionId: conversationId, userId: "user-17", input: question },
);

Each top-level trace() is its own session unless you pass sessionId, which makes each trace one turn of that conversation. A string returned from the traced function becomes the turn's output. Context follows AsyncLocalStorage, so it crosses await.

Short-lived processes

Events are sent in the background in batches. The SDK flushes on beforeExit, which does not fire on process.exit(), and a serverless function can freeze as soon as it responds:

import { flush, shutdown } from "@codingdotsh/sdk";

await flush(); // serverless: waitUntil(flush())
await shutdown(); // before process.exit()

AI SDK

import { openai } from "@ai-sdk/openai";
import { OpenTelemetry } from "@ai-sdk/otel";
import { generateText } from "ai";
import { aiSdkTracer } from "@codingdotsh/sdk";

const { text } = await generateText({
  model: openai("gpt-4o-mini"),
  prompt: "Summarize this ticket.",
  telemetry: { integrations: new OpenTelemetry({ tracer: aiSdkTracer() }) },
});

That is AI SDK 7. On AI SDK 6 and older, pass experimental_telemetry: { isEnabled: true, tracer: aiSdkTracer() } instead.

Exports

init(opts?)

Configures the SDK. Called implicitly (from env) on first use; call it to override env.

function init(opts?: InitOptions): void

flush(timeoutMs?)

Waits until everything recorded so far is sent, or timeoutMs passes. Serverless: waitUntil(flush()). Resolves true when the queue drained.

function flush(timeoutMs = 5000): Promise<boolean>

shutdown(timeoutMs?)

Final flush, then stop sending. Apps that call process.exit should await shutdown() first.

function shutdown(timeoutMs = 5000): Promise<boolean>

stats()

Counts of events queued, sent, dropped and failed in this process. Counts from a transport replaced by a later init() are kept.

function stats(): TransportStats

observe(client)

Wraps an OpenAI or Anthropic client so its model calls are recorded. Unknown objects are returned untouched; wrapping twice is a no-op.

function observe<T>(client: T): T

observeFetch(f?)

Returns a fetch that records LLM API calls. Pass the fetch to wrap (default: the global fetch at the time of the call to observeFetch).

function observeFetch(f?: typeof fetch): typeof fetch

trace(name, fn, opts?)

Runs fn as one traced turn. Model calls, tools and steps inside it join the trace. A trace nested inside another trace is recorded as a step span.

function trace<T>(name: string, fn: (t: TraceHandle) => T, opts: TraceOptions = {}): T

span(kind, name, fn, opts?)

Runs fn as a tool call (tool.start/tool.end) or a step (span) inside the current trace. Outside a trace it becomes its own one-span trace.

function span<T>(kind: "tool" | "step", name: string, fn: () => T, opts: SpanOptions = {}): T

withSession(id, fn)

Runs fn with every trace and untraced call inside it recorded into session id.

function withSession<T>(id: string, fn: () => T): T

aiSdkTracer(opts?)

An OpenTelemetry Tracer-shaped object that sends AI SDK spans to coding.sh.

function aiSdkTracer(opts: AiSdkTracerOptions = {}): AiSdkTracer

InitOptions

interface InitOptions {
  url?: string;
  key?: string;
  service?: string;
  capture?: CaptureMode;
  flushIntervalMs?: number;
  maxBatch?: number;
  maxQueue?: number;
  disabled?: boolean;
  debug?: boolean;
  fetch?: typeof fetch;
}

CaptureMode

type CaptureMode = "full" | "metadata";

TraceOptions

interface TraceOptions {
  sessionId?: string;
  userId?: string;
  metadata?: Record<string, unknown>;
  input?: unknown;
}

TraceHandle

Passed to the function given to trace(). The ids of the trace and the session it records into.

interface TraceHandle {
  traceId: string;
  spanId: string;
  sessionId: string;
}

SpanOptions

interface SpanOptions {
  input?: unknown;
}

AiSdkTracer

What aiSdkTracer() returns: the subset of the OpenTelemetry Tracer interface the AI SDK calls.

interface AiSdkTracer {
  startSpan(name: string, options?: SpanOptionsLike, context?: unknown): SdkSpan;
  startActiveSpan<F extends (span: SdkSpan) => unknown>(name: string, fn: F): ReturnType<F>;
  startActiveSpan<F extends (span: SdkSpan) => unknown>(name: string, options: SpanOptionsLike, fn: F): ReturnType<F>;
  startActiveSpan<F extends (span: SdkSpan) => unknown>(name: string, options: SpanOptionsLike, context: unknown, fn: F): ReturnType<F>;
}

AiSdkTracerOptions

interface AiSdkTracerOptions {
  sessionId?: string;
}

TransportStats

What stats() returns.

interface TransportStats {
  queued: number;
  sent: number;
  dropped: number;
  failed: number;
}

Python: coding-sh

Import it as coding_sh.

Install

pip install coding-sh
export CODING_SH_URL=http://127.0.0.1:4318   # where coding.sh serve listens
export CODING_SH_SERVICE=billing-agent

Python >=3.9. Standard library only. Extras: coding-sh[openai], coding-sh[anthropic], coding-sh[otel]. Add CODING_SH_INGEST_KEY when your server requires ingest keys.

Minimal use

import coding_sh
from openai import OpenAI

client = coding_sh.observe(OpenAI())
reply = client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": "Where is my order?"}])

observe() also takes AsyncOpenAI, Anthropic and AsyncAnthropic clients, sync, async and streaming.

Traces, tools and conversations

with coding_sh.trace("billing-reply", session_id=conversation_id, user_id="user-17", input=question) as t:
    with coding_sh.span("tool", "lookup_invoice", input={"id": 42}) as s:
        s.set_output(invoices.get(42))
    reply = client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": question}])
    t.set_output(reply.choices[0].message.content)

@coding_sh.span("tool", "refund")
def refund(invoice_id: int) -> dict:
    return invoices.refund(invoice_id)

coding_sh.flush(timeout=5)  # end of a script, a job or a serverless handler

trace and span work as context managers (with and async with) and as decorators. A decorated tool span records the call's arguments as input and the return value as output. Context follows contextvars, so it crosses await and asyncio tasks. Work handed to a thread pool starts untraced unless you run it with contextvars.copy_context().run.

Public names

Everything in coding_sh.__all__:

coding_sh.__version__

The package version, a string.

coding_sh.init(...)

Configure the SDK. Optional: the first event configures it from env vars.

Env: CODING_SH_URL, CODING_SH_INGEST_KEY, CODING_SH_SERVICE (or OTEL_SERVICE_NAME), CODING_SH_CAPTURE_MODE (full|metadata), CODING_SH_DISABLED, CODING_SH_DEBUG. Without a URL the SDK does nothing.

def init(
    url: Optional[str] = None,
    key: Optional[str] = None,
    service: Optional[str] = None,
    capture: Optional[str] = None,
    flush_interval_ms: Optional[float] = None,
    max_batch: Optional[int] = None,
    max_queue: Optional[int] = None,
    disabled: Optional[bool] = None,
    debug: Optional[bool] = None,
) -> None

coding_sh.observe(client)

Record every model call made through client. Returns the same client.

Supports openai.OpenAI / AsyncOpenAI (and Azure variants) and anthropic.Anthropic / AsyncAnthropic (and Bedrock/Vertex variants). Anything else is returned untouched.

def observe(client: Any) -> Any

coding_sh.trace(...)

A unit of work: with coding_sh.trace("checkout", session_id=conv_id, input=prompt) as t:.

Also a decorator (sync or async). The yielded object has set_input(), set_output(), session_id and trace_id. A decorated function's return value becomes the output.

def trace(
    name: str = "trace",
    session_id: Optional[str] = None,
    user_id: Optional[str] = None,
    metadata: Optional[Dict[str, Any]] = None,
    input: Any = None,
) -> _ScopeCM

coding_sh.span(kind, name, ...)

A tool call (span("tool", "lookup_order")) or a step (span("step", "rerank")).

Context manager or decorator; as a decorator, the call's arguments become the input and its return value the output.

def span(kind: str, name: str, input: Any = None) -> _ScopeCM

coding_sh.with_session(session_id)

Group traces and untraced calls into one conversation session (context manager or decorator).

def with_session(session_id: str) -> _SessionCM

coding_sh.current_session_id()

The session the next event would go to, if one is set.

def current_session_id() -> Optional[str]

coding_sh.flush(...)

Send everything queued so far. Returns True when the queue drained in time.

def flush(timeout: Optional[float] = None) -> bool

coding_sh.shutdown(...)

Flush, then stop accepting events.

def shutdown(timeout: Optional[float] = 2.0) -> bool

coding_sh.stats()

{queued, sent, dropped, failed} event counts for this process.

def stats() -> Dict[str, int]

coding_sh.otel

For apps that already emit OpenTelemetry gen_ai spans. Needs pip install coding-sh[otel]. configure() picks the endpoint from its url argument, then OTEL_EXPORTER_OTLP_TRACES_ENDPOINT or OTEL_EXPORTER_OTLP_ENDPOINT, then CODING_SH_URL with CODING_SH_INGEST_KEY. See plain OpenTelemetry.

coding_sh.otel.configure(...)

Build (and by default install) a TracerProvider that exports to coding.sh.

Returns the provider, or None when OpenTelemetry is not installed or no endpoint is known. Call provider.shutdown() (or rely on the SDK's atexit hook) to flush on exit.

def configure(
    service: Optional[str] = None,
    url: Optional[str] = None,
    key: Optional[str] = None,
    set_global: bool = True,
    batch: bool = True,
) -> Any

coding_sh.otel.otel_env(...)

OTEL_* exporter env vars that point any OTel SDK at a coding.sh server.

Defaults come from CODING_SH_URL, CODING_SH_INGEST_KEY and CODING_SH_SERVICE. The result holds the ingest key: pass it to a child process, never log it.

def otel_env(
    url: Optional[str] = None,
    key: Optional[str] = None,
    service: Optional[str] = None,
) -> Dict[str, str]

coding_sh.otel.ConversationSpanProcessor

Sets gen_ai.conversation.id from the current coding.sh session on every new span.

class ConversationSpanProcessor:
    def on_start(self, span: Any, parent_context: Any = None) -> None: ...
    def on_end(self, span: Any) -> None: ...
    def shutdown(self) -> None: ...
    def force_flush(self, timeout_millis: int = 30000) -> bool: ...

coding_sh.otel.CONVERSATION_ATTR

A constant: "gen_ai.conversation.id".

Differences between the SDKs

• flush() waits at most 5000 ms by default in TypeScript. In Python, flush() with timeout=None waits until the queue is sent or the sender gives up on it.
• shutdown() waits at most 5000 ms in TypeScript and 2.0 s in Python. Without a call, Python flushes at exit for at most 2.0 s, and TypeScript flushes on beforeExit for at most 5000 ms.
• TypeScript times are in milliseconds (timeoutMs, flushIntervalMs). Python timeouts are in seconds; only flush_interval_ms is in milliseconds.
• In TypeScript, trace(), span() and withSession() take the function to run. In Python, they are context managers and decorators.
• aiSdkTracer() and observeFetch() are TypeScript only. coding_sh.otel is Python only.