OpenAI Integration Guide

from openai import OpenAI

client = OpenAI(
    base_url="https://gateway.muxx.dev/v1",
    default_headers={
        "X-Muxx-Api-Key": "muxx_sk_live_xxxxxxxxxxxx"
    }
)

# Use normally - all requests are logged
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

from muxx import Muxx
from openai import OpenAI

muxx = Muxx()
client = muxx.wrap(OpenAI())

# Use normally - all requests are traced
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

from muxx import Muxx
from openai import OpenAI

muxx = Muxx()

client = muxx.wrap(OpenAI(
    base_url="https://gateway.muxx.dev/v1",
    default_headers={"X-Muxx-Api-Key": muxx.api_key}
))

from muxx import Muxx, trace
from openai import OpenAI

muxx = Muxx()
client = muxx.wrap(OpenAI())

@trace("chat-turn")
def handle_message(user_id: str, message: str, history: list) -> str:
    messages = history + [{"role": "user", "content": message}]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages
    )

    return response.choices[0].message.content

from muxx import Muxx, trace, span
from openai import OpenAI

muxx = Muxx()
client = muxx.wrap(OpenAI())

@trace("rag-query")
def answer_question(question: str) -> str:
    # Get embedding
    with muxx.span("embed-query"):
        embedding = client.embeddings.create(
            model="text-embedding-3-small",
            input=question
        )

    # Search (your vector DB)
    with muxx.span("search"):
        documents = search_similar(embedding.data[0].embedding)

    # Generate answer
    with muxx.span("generate"):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": f"Context: {documents}"},
                {"role": "user", "content": question}
            ]
        )

    return response.choices[0].message.content

@trace("function-call")
def handle_with_tools(message: str) -> str:
    tools = [
        {
            "type": "function",
            "function": {
                "name": "get_weather",
                "parameters": {"type": "object", "properties": {...}}
            }
        }
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": message}],
        tools=tools
    )

    # Tool calls are logged automatically
    if response.choices[0].message.tool_calls:
        # Handle tool calls
        pass

    return response.choices[0].message.content

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Write a story"}],
    stream=True
)

for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

# Complete response is logged when stream ends

from openai import OpenAIError, RateLimitError

@trace("api-call")
def safe_call(message: str) -> str:
    try:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": message}]
        )
        return response.choices[0].message.content
    except RateLimitError:
        # Logged as error in trace
        raise
    except OpenAIError as e:
        # Logged with error details
        raise