Anthropic Claude SDK Integration

Quickstart

Install both SDKs, then wrap your Anthropic calls inside a Nexus trace. Record input/output tokens in metadata for cost tracking.

npm install @anthropic-ai/sdk @keylightdigital/nexus

pip install anthropic keylightdigital-nexus

TypeScript example

import Anthropic from '@anthropic-ai/sdk'
import { NexusClient } from '@keylightdigital/nexus'

const anthropic = new Anthropic({ apiKey: process.env.ANTHROPIC_API_KEY })
const nexus = new NexusClient({
  apiKey: process.env.NEXUS_API_KEY!,
  agentId: 'my-claude-agent',
})

async function runAgent(prompt: string) {
  const trace = await nexus.startTrace({ name: 'claude-completion' })

  try {
    const span = await trace.startSpan({ name: 'anthropic.messages.create' })

    const message = await anthropic.messages.create({
      model: 'claude-sonnet-4-6',
      max_tokens: 1024,
      messages: [{ role: 'user', content: prompt }],
    })

    await span.end({
      status: 'success',
      metadata: {
        model: message.model,
        input_tokens: message.usage.input_tokens,
        output_tokens: message.usage.output_tokens,
        stop_reason: message.stop_reason,
      },
    })

    await trace.end({ status: 'success' })
    return message.content[0].type === 'text' ? message.content[0].text : ''
  } catch (err) {
    await trace.end({ status: 'error', metadata: { error: String(err) } })
    throw err
  }
}

Python example

import os
import anthropic
from keylightdigital_nexus import NexusClient

client = anthropic.Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
nexus = NexusClient(
    api_key=os.environ["NEXUS_API_KEY"],
    agent_id="my-claude-agent",
)

def run_agent(prompt: str) -> str:
    trace = nexus.start_trace(name="claude-completion")
    try:
        span = trace.start_span(name="anthropic.messages.create")

        message = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=1024,
            messages=[{"role": "user", "content": prompt}],
        )

        span.end(
            status="success",
            metadata={
                "model": message.model,
                "input_tokens": message.usage.input_tokens,
                "output_tokens": message.usage.output_tokens,
                "stop_reason": message.stop_reason,
            },
        )

        trace.end(status="success")
        return message.content[0].text if message.content else ""
    except Exception as err:
        trace.end(status="error", metadata={"error": str(err)})
        raise

Multi-turn conversation tracing

For multi-turn conversations, create one parent trace per session and one child span per API call. This gives you a waterfall view of the entire conversation in the Nexus dashboard.

// One trace = one conversation session
const trace = await nexus.startTrace({ name: 'conversation-session' })

for (const userMessage of conversationHistory) {
  const span = await trace.startSpan({ name: `turn-${turn++}` })
  const response = await anthropic.messages.create({ ... })
  await span.end({ status: 'success', metadata: { tokens: response.usage } })
}

await trace.end({ status: 'success' })

Each span appears as a row in the waterfall timeline on the trace detail page, making it easy to see which turns took longest.