Copy as Markdown[Open in ChatGPT](https://chatgpt.com/?q=Read%20https%3A%2F%2Fdocs-docusaurus.kinsta.page%2Fuser-guides%2Fai%2Fotel-integration%2Fcode-examples%2Fpython-manual.md%20and%20help%20me%20with%20my%20question%20about%20this%20Coralogix%20documentation%20page.)[Open in Claude](https://claude.ai/new?q=Read%20https%3A%2F%2Fdocs-docusaurus.kinsta.page%2Fuser-guides%2Fai%2Fotel-integration%2Fcode-examples%2Fpython-manual.md%20and%20help%20me%20with%20my%20question%20about%20this%20Coralogix%20documentation%20page.)

# Python — manual instrumentation

Pure manual spans with no instrumentation library — just OTel SDK + raw HTTP. **Universal template for any provider/language.**

Before you start

These examples export to a local OpenTelemetry Collector over OTLP/gRPC. Deploy the collector first — see [Code examples → Deploy an OpenTelemetry Collector](https://docs-docusaurus.kinsta.page/user-guides/ai/otel-integration/code-examples/.md#prerequisites-deploy-an-opentelemetry-collector).

## Install[​](#install "Direct link to Install")

```
pip install opentelemetry-sdk opentelemetry-exporter-otlp-proto-grpc httpx
```

## Environment variables[​](#environment-variables "Direct link to Environment variables")

```
export OPENAI_API_KEY="sk-..."

export CX_OTEL_ENDPOINT="http://<COLLECTOR_HOST>:4317"
```

## Script[​](#script "Direct link to Script")

```
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", "http://<COLLECTOR_HOST>:4317")

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, insecure=True)))

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()
```

Universal template

To adapt for Anthropic, Gemini, or any other provider, change the HTTP endpoint, request body format, and response parsing. The OTel span attributes remain identical.

## Next steps[​](#next-steps "Direct link to Next steps")

Look up which open-source library to use for your provider in [Compatibility matrix](https://docs-docusaurus.kinsta.page/user-guides/ai/otel-integration/providers/.md).