Code examples

import asyncio

from agents import Agent, Runner, function_tool

# --- OTel imports ---
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.openai_agents import OpenAIAgentsInstrumentor
from opentelemetry.instrumentation.openai_v2 import OpenAIInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor


# --- OTel setup: configure tracer provider and OTLP exporter ---
def configure_otel() -> TracerProvider:
    resource = Resource.create()
    provider = TracerProvider(resource=resource)
    exporter = OTLPSpanExporter()  # reads OTEL_EXPORTER_OTLP_ENDPOINT from env
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return provider


# --- Your app logic (replace with your own) ---
@function_tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is 22C and sunny."


async def main():
    # OTel: initialize tracing and auto-instrument the OpenAI Agents SDK
    provider = configure_otel()
    OpenAIAgentsInstrumentor().instrument(tracer_provider=provider)
    OpenAIInstrumentor().instrument(tracer_provider=provider)

    # Your app logic
    agent = Agent(
        name="Travel Assistant",
        instructions="You are a concise travel assistant. Answer in one sentence.",
        tools=[get_weather],
    )

    result = await Runner.run(agent, input="What's the weather in Paris?")
    print(f"Agent response: {result.final_output}")

    # OTel: flush and shut down the tracer provider
    provider.force_flush()
    provider.shutdown()


if __name__ == "__main__":
    asyncio.run(main())

import anthropic

# --- OTel imports ---
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor


# --- OTel setup: configure tracer provider and OTLP exporter ---
def configure_otel() -> TracerProvider:
    resource = Resource.create()
    provider = TracerProvider(resource=resource)
    exporter = OTLPSpanExporter()  # reads OTEL_EXPORTER_OTLP_ENDPOINT from env
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return provider


def main():
    # OTel: initialize tracing and auto-instrument the Anthropic SDK
    provider = configure_otel()
    AnthropicInstrumentor().instrument(tracer_provider=provider)

    # Your app logic
    client = anthropic.Anthropic()
    message = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=256,
        messages=[
            {"role": "user", "content": "What is OpenTelemetry in one sentence?"}
        ],
    )
    print(f"Claude response: {message.content[0].text}")

    # OTel: flush and shut down the tracer provider
    provider.force_flush()
    provider.shutdown()


if __name__ == "__main__":
    main()

import boto3

# --- OTel imports ---
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.botocore import BotocoreInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor


# --- OTel setup: configure tracer provider and OTLP exporter ---
def configure_otel() -> TracerProvider:
    resource = Resource.create()
    provider = TracerProvider(resource=resource)
    exporter = OTLPSpanExporter()  # reads OTEL_EXPORTER_OTLP_ENDPOINT from env
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return provider


def main():
    # OTel: initialize tracing and auto-instrument botocore (Bedrock)
    provider = configure_otel()
    BotocoreInstrumentor().instrument(tracer_provider=provider)

    # Your app logic
    client = boto3.client("bedrock-runtime", region_name="us-east-1")
    response = client.converse(
        modelId="anthropic.claude-3-haiku-20240307-v1:0",
        messages=[
            {
                "role": "user",
                "content": [{"text": "What is OpenTelemetry in one sentence?"}],
            }
        ],
        inferenceConfig={"maxTokens": 256, "temperature": 0.7},
    )

    text = response["output"]["message"]["content"][0]["text"]
    usage = response["usage"]
    print(f"Bedrock response: {text}")
    print(f"Tokens - input: {usage['inputTokens']}, output: {usage['outputTokens']}")

    # OTel: flush and shut down the tracer provider
    provider.force_flush()
    provider.shutdown()


if __name__ == "__main__":
    main()

from openai import OpenAI
from traceloop.sdk import Traceloop  # OTel: Traceloop auto-instruments OpenAI

Traceloop.init()  # OTel: reads TRACELOOP_BASE_URL, TRACELOOP_HEADERS from env

# Your app logic
client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "You are a concise assistant."},
        {"role": "user", "content": "What is OpenTelemetry in one sentence?"},
    ],
    max_tokens=100,
)
print(response.choices[0].message.content)

import json, os
import httpx

# --- OTel imports ---
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import SpanKind, StatusCode

# --- OTel setup: configure tracer provider and OTLP exporter ---
CX_ENDPOINT = os.environ.get("CX_OTEL_ENDPOINT", "https://ingress.:443")
CX_API_KEY = os.environ.get("CX_API_KEY", "")
OPENAI_KEY = os.environ.get("OPENAI_API_KEY", "")

resource = Resource.create({"service.name": "manual-genai-demo",
    "cx.application.name": "my-genai-app", "cx.subsystem.name": "my-service"})
provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(endpoint=CX_ENDPOINT,
                     headers=[("Authorization", f"Bearer {CX_API_KEY}")])))
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("genai-manual", "1.0.0")


def to_semconv(messages):
    """Convert OpenAI messages to new semconv JSON parts format."""
    result = []
    for m in messages:
        parts = []
        if m.get("content"): parts.append({"type": "text", "content": m["content"]})
        if m.get("tool_calls"):
            for tc in m["tool_calls"]:
                parts.append({"type": "tool_call", "id": tc["id"],
                    "name": tc["function"]["name"], "arguments": tc["function"]["arguments"]})
        entry = {"role": m["role"], "parts": parts}
        if m.get("tool_call_id"): entry["tool_call_id"] = m["tool_call_id"]
        result.append(entry)
    return json.dumps(result)


# --- Your app logic with manual OTel GenAI spans ---
def chat(messages, model="gpt-4o-mini", max_tokens=200, user=None):
    # OTel: create a GenAI span with required attributes
    with tracer.start_as_current_span(f"chat {model}", kind=SpanKind.CLIENT,
        attributes={"gen_ai.operation.name": "chat", "gen_ai.provider.name": "openai",
            "gen_ai.request.model": model, "gen_ai.request.max_tokens": max_tokens,
            "gen_ai.input.messages": to_semconv(messages),
            "server.address": "api.openai.com", "server.port": 443}) as span:
        if user: span.set_attribute("gen_ai.request.user", user)

        body = {"model": model, "messages": messages, "max_tokens": max_tokens}
        if user: body["user"] = user

        resp = httpx.post("https://api.openai.com/v1/chat/completions",
            headers={"Authorization": f"Bearer {OPENAI_KEY}",
                     "Content-Type": "application/json"}, json=body, timeout=60.0)
        resp.raise_for_status()
        data = resp.json()

        usage = data.get("usage", {})
        choices = data.get("choices", [])
        span.set_attribute("gen_ai.response.model", data.get("model", model))
        span.set_attribute("gen_ai.response.id", data.get("id", ""))
        span.set_attribute("gen_ai.response.finish_reasons",
            json.dumps([c.get("finish_reason") for c in choices]))
        span.set_attribute("gen_ai.usage.input_tokens", usage.get("prompt_tokens", 0))
        span.set_attribute("gen_ai.usage.output_tokens", usage.get("completion_tokens", 0))

        out = []
        for c in choices:
            m = c.get("message", {})
            parts = []
            if m.get("content"): parts.append({"type": "text", "content": m["content"]})
            if m.get("tool_calls"):
                for tc in m["tool_calls"]:
                    parts.append({"type": "tool_call", "id": tc["id"],
                        "name": tc["function"]["name"], "arguments": tc["function"]["arguments"]})
            out.append({"role": m.get("role", "assistant"), "parts": parts})
        span.set_attribute("gen_ai.output.messages", json.dumps(out))
        return data


result = chat(
    messages=[{"role": "system", "content": "You are a concise assistant."},
              {"role": "user", "content": "What is OpenTelemetry in one sentence?"}],
    user="user-42")
print(f"Response: {result['choices'][0]['message']['content']}")

provider.force_flush()
provider.shutdown()

import os
from google import genai

# --- OTel imports ---
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.google_genai import GoogleGenAiSdkInstrumentor
from opentelemetry.sdk.resources import SERVICE_NAME, Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor


# --- OTel setup: configure tracer provider and OTLP exporter ---
def configure_otel() -> TracerProvider:
    resource = Resource.create({
        SERVICE_NAME: "my-genai-service",
        "cx.application.name": "my-genai-app",
        "cx.subsystem.name": "my-service",
    })
    provider = TracerProvider(resource=resource)
    endpoint = os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317")
    insecure = os.getenv("OTEL_EXPORTER_OTLP_INSECURE", "true").lower() == "true"
    provider.add_span_processor(
        BatchSpanProcessor(OTLPSpanExporter(endpoint=endpoint, insecure=insecure))
    )
    trace.set_tracer_provider(provider)
    return provider


def main():
    # OTel: initialize tracing, then auto-instrument the Google GenAI SDK
    provider = configure_otel()
    GoogleGenAiSdkInstrumentor().instrument()

    # Your app logic
    client = genai.Client()
    response = client.models.generate_content(
        model="gemini-2.0-flash",
        contents="What is OpenTelemetry in one sentence?",
    )
    print(f"Gemini response: {response.text}")

    # OTel: flush and shut down the tracer provider
    provider.force_flush()
    provider.shutdown()


if __name__ == "__main__":
    main()