Code examples

import asyncio

from agents import Agent, Runner, function_tool
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


def configure_otel() -> TracerProvider:
    resource = Resource.create()
    provider = TracerProvider(resource=resource)
    exporter = OTLPSpanExporter()
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return provider


@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():
    provider = configure_otel()
    OpenAIAgentsInstrumentor().instrument(tracer_provider=provider)
    OpenAIInstrumentor().instrument(tracer_provider=provider)

    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}")

    provider.force_flush()
    provider.shutdown()


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

import anthropic
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


def configure_otel() -> TracerProvider:
    resource = Resource.create()
    provider = TracerProvider(resource=resource)
    exporter = OTLPSpanExporter()
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return provider


def main():
    provider = configure_otel()
    AnthropicInstrumentor().instrument(tracer_provider=provider)

    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}")

    provider.force_flush()
    provider.shutdown()


if __name__ == "__main__":
    main()

import boto3
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


def configure_otel() -> TracerProvider:
    resource = Resource.create()
    provider = TracerProvider(resource=resource)
    exporter = OTLPSpanExporter()
    provider.add_span_processor(BatchSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return provider


def main():
    provider = configure_otel()
    BotocoreInstrumentor().instrument(tracer_provider=provider)

    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']}")

    provider.force_flush()
    provider.shutdown()


if __name__ == "__main__":
    main()

from openai import OpenAI
from traceloop.sdk import Traceloop

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

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
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

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)


def chat(messages, model="gpt-4o-mini", max_tokens=200, user=None):
    with tracer.start_as_current_span(f"chat {model}", kind=SpanKind.CLIENT,
        attributes={"gen_ai.operation.name": "chat", "gen_ai.system": "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()