# Environment variable (recommended)
export OPENAI_API_KEY="sk-proj-..."

# Or load from file
source ~/.secrets/openai  # Loads OPENAI_API_KEY

curl https://api.openai.com/v1/chat/completions \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -H "OpenAI-Organization: org-xxxxx" \  # Optional
  -H "OpenAI-Project: proj-xxxxx"         # Optional

from openai import OpenAI

client = OpenAI()  # Uses OPENAI_API_KEY env var

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain SDD methodology in 3 sentences."}
    ]
)

print(response.choices[0].message.content)

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

    # Generation
    temperature=0.7,        # 0-2, higher = more creative
    top_p=1.0,              # Nucleus sampling (alternative to temp)
    max_tokens=4096,        # Max response length
    n=1,                    # Number of completions

    # Control
    stop=["\n\n", "END"],   # Stop sequences
    presence_penalty=0.0,   # -2 to 2, penalize repeated topics
    frequency_penalty=0.0,  # -2 to 2, penalize repeated tokens

    # Format
    response_format={"type": "json_object"},  # Force JSON output
    seed=42,                # Deterministic outputs

    # Advanced
    logprobs=True,          # Return token probabilities
    top_logprobs=5,         # Number of logprobs per position
    user="user-123"         # Track abuse
)

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

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

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get current weather for a location",
            "parameters": {
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name, e.g., 'Kyiv'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "What's the weather in Kyiv?"}],
    tools=tools,
    tool_choice="auto"  # "auto" | "none" | "required" | {"type": "function", "function": {"name": "get_weather"}}
)

# Check if tool call was made
message = response.choices[0].message
if message.tool_calls:
    for tool_call in message.tool_calls:
        print(f"Function: {tool_call.function.name}")
        print(f"Arguments: {tool_call.function.arguments}")

import json

def get_weather(location: str, unit: str = "celsius") -> dict:
    # Your implementation
    return {"temperature": 15, "condition": "cloudy", "unit": unit}

# Initial request
messages = [{"role": "user", "content": "What's the weather in Kyiv?"}]
response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools
)

message = response.choices[0].message
messages.append(message)  # Add assistant message

# Process tool calls
if message.tool_calls:
    for tool_call in message.tool_calls:
        args = json.loads(tool_call.function.arguments)
        result = get_weather(**args)

        messages.append({
            "role": "tool",
            "tool_call_id": tool_call.id,
            "content": json.dumps(result)
        })

    # Get final response
    final_response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools
    )
    print(final_response.choices[0].message.content)

# Response with multiple tool calls
{
    "tool_calls": [
        {"id": "call_1", "function": {"name": "get_weather", "arguments": "{\"location\": \"Kyiv\"}"}},
        {"id": "call_2", "function": {"name": "get_weather", "arguments": "{\"location\": \"London\"}"}}
    ]
}

from pydantic import BaseModel
from typing import List, Optional

class Address(BaseModel):
    street: str
    city: str
    country: str

class Person(BaseModel):
    name: str
    age: int
    email: Optional[str]
    addresses: List[Address]

response = client.beta.chat.completions.parse(
    model="gpt-4o",
    messages=[
        {"role": "user", "content": "Extract: John Doe, 30, [email protected], lives at 123 Main St, Kyiv, Ukraine"}
    ],
    response_format=Person
)

person = response.choices[0].message.parsed
print(person.name)  # "John Doe"
print(person.addresses[0].city)  # "Kyiv"

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "What's in this image?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "https://example.com/image.jpg",
                        "detail": "high"  # "low" | "high" | "auto"
                    }
                }
            ]
        }
    ]
)

import base64

def encode_image(image_path: str) -> str:
    with open(image_path, "rb") as f:
        return base64.standard_b64encode(f.read()).decode("utf-8")

image_data = encode_image("screenshot.png")

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe this screenshot"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:image/png;base64,{image_data}"
                    }
                }
            ]
        }
    ]
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Compare these two designs"},
                {"type": "image_url", "image_url": {"url": "https://example.com/design1.png"}},
                {"type": "image_url", "image_url": {"url": "https://example.com/design2.png"}}
            ]
        }
    ]
)

response = client.images.generate(
    model="dall-e-3",
    prompt="A serene Japanese garden with a red bridge over a koi pond, photorealistic",
    size="1024x1024",      # "1024x1024" | "1792x1024" | "1024x1792"
    quality="hd",          # "standard" | "hd"
    style="vivid",         # "vivid" | "natural"
    n=1                    # DALL-E 3 only supports n=1
)

image_url = response.data[0].url
revised_prompt = response.data[0].revised_prompt  # DALL-E 3 rewrites prompts

# URL (default, expires in 1 hour)
response = client.images.generate(
    model="dall-e-3",
    prompt="...",
    response_format="url"
)

# Base64 (for immediate use)
response = client.images.generate(
    model="dall-e-3",
    prompt="...",
    response_format="b64_json"
)
image_b64 = response.data[0].b64_json

response = client.images.edit(
    model="dall-e-2",
    image=open("original.png", "rb"),
    mask=open("mask.png", "rb"),  # Transparent areas will be regenerated
    prompt="Add a red sports car",
    size="1024x1024",
    n=1
)

response = client.images.create_variation(
    model="dall-e-2",
    image=open("original.png", "rb"),
    size="1024x1024",
    n=3
)

Style keywords: "photorealistic", "digital art", "oil painting", "3D render", "anime style"
Lighting: "soft lighting", "golden hour", "neon glow", "dramatic shadows"
Composition: "centered", "rule of thirds", "wide angle", "close-up", "aerial view"
Quality: "highly detailed", "8K resolution", "professional photography"
Mood: "serene", "dramatic", "playful", "mysterious", "minimalist"

audio_file = open("speech.mp3", "rb")

transcript = client.audio.transcriptions.create(
    model="whisper-1",
    file=audio_file,
    language="uk",           # ISO 639-1 code (optional)
    response_format="text",  # "json" | "text" | "srt" | "vtt" | "verbose_json"
    temperature=0.0,         # 0-1, lower = more deterministic
    prompt="SDD, Faion Network"  # Hints for proper nouns
)

print(transcript)  # Plain text for response_format="text"

transcript = client.audio.transcriptions.create(
    model="whisper-1",
    file=audio_file,
    response_format="verbose_json",
    timestamp_granularities=["word", "segment"]
)

for segment in transcript.segments:
    print(f"[{segment.start:.2f}s - {segment.end:.2f}s] {segment.text}")

# Translates any language to English
translation = client.audio.translations.create(
    model="whisper-1",
    file=open("ukrainian_speech.mp3", "rb"),
    response_format="text"
)

response = client.audio.speech.create(
    model="tts-1-hd",        # "tts-1" (fast) | "tts-1-hd" (quality)
    voice="nova",            # alloy | echo | fable | onyx | nova | shimmer
    input="Hello, welcome to Faion Network!",
    speed=1.0,               # 0.25 - 4.0
    response_format="mp3"    # mp3 | opus | aac | flac | wav | pcm
)

response.stream_to_file("output.mp3")

with client.audio.speech.with_streaming_response.create(
    model="tts-1",
    voice="alloy",
    input="Long text to speak..."
) as response:
    response.stream_to_file("output.mp3")

response = client.embeddings.create(
    model="text-embedding-3-large",
    input="Faion Network provides AI tools for solopreneurs",
    encoding_format="float",  # "float" | "base64"
    dimensions=1536           # Optional, reduce for cost savings
)

embedding = response.data[0].embedding
print(f"Dimensions: {len(embedding)}")

texts = [
    "What is SDD methodology?",
    "How to create AI agents?",
    "Faion Network pricing plans"
]

response = client.embeddings.create(
    model="text-embedding-3-large",
    input=texts
)

for i, item in enumerate(response.data):
    print(f"Text {i}: {len(item.embedding)} dimensions")

# Reduce dimensions for cost/performance tradeoff
response = client.embeddings.create(
    model="text-embedding-3-large",
    input="Your text",
    dimensions=256  # Reduced from 3072
)

import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

query_embedding = client.embeddings.create(
    model="text-embedding-3-small",
    input="How to use SDD?"
).data[0].embedding

doc_embedding = client.embeddings.create(
    model="text-embedding-3-small",
    input="SDD is a specification-driven methodology..."
).data[0].embedding

similarity = cosine_similarity(query_embedding, doc_embedding)
print(f"Similarity: {similarity:.4f}")

assistant = client.beta.assistants.create(
    model="gpt-4o",
    name="SDD Expert",
    instructions="You are an expert on Specification-Driven Development. Help users understand and apply SDD methodology.",
    tools=[
        {"type": "code_interpreter"},
        {"type": "file_search"}
    ],
    temperature=0.7,
    metadata={"version": "1.0"}
)

thread = client.beta.threads.create(
    messages=[
        {
            "role": "user",
            "content": "What are the key phases of SDD?"
        }
    ]
)

run = client.beta.threads.runs.create_and_poll(
    thread_id=thread.id,
    assistant_id=assistant.id
)

if run.status == "completed":
    messages = client.beta.threads.messages.list(thread_id=thread.id)
    for message in messages.data:
        if message.role == "assistant":
            print(message.content[0].text.value)

from openai import AssistantEventHandler

class EventHandler(AssistantEventHandler):
    def on_text_delta(self, delta, snapshot):
        print(delta.value, end="", flush=True)

    def on_tool_call_created(self, tool_call):
        print(f"\n[Using tool: {tool_call.type}]")

with client.beta.threads.runs.stream(
    thread_id=thread.id,
    assistant_id=assistant.id,
    event_handler=EventHandler()
) as stream:
    stream.until_done()

# Create vector store
vector_store = client.beta.vector_stores.create(name="SDD Documentation")

# Upload files
file = client.files.create(
    file=open("sdd-guide.pdf", "rb"),
    purpose="assistants"
)

client.beta.vector_stores.files.create(
    vector_store_id=vector_store.id,
    file_id=file.id
)

# Attach to assistant
assistant = client.beta.assistants.update(
    assistant_id=assistant.id,
    tool_resources={"file_search": {"vector_store_ids": [vector_store.id]}}
)

# Upload data file
file = client.files.create(
    file=open("data.csv", "rb"),
    purpose="assistants"
)

# Create thread with attachment
thread = client.beta.threads.create(
    messages=[
        {
            "role": "user",
            "content": "Analyze this data and create a chart",
            "attachments": [
                {"file_id": file.id, "tools": [{"type": "code_interpreter"}]}
            ]
        }
    ]
)

import json

# Create JSONL file with requests
requests = [
    {
        "custom_id": "request-1",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": "Hello!"}],
            "max_tokens": 100
        }
    },
    {
        "custom_id": "request-2",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": "World!"}],
            "max_tokens": 100
        }
    }
]

with open("batch_requests.jsonl", "w") as f:
    for req in requests:
        f.write(json.dumps(req) + "\n")

# Upload file
batch_file = client.files.create(
    file=open("batch_requests.jsonl", "rb"),
    purpose="batch"
)

# Create batch
batch = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
    metadata={"description": "Daily summaries"}
)

print(f"Batch ID: {batch.id}")
print(f"Status: {batch.status}")

batch = client.batches.retrieve(batch.id)
print(f"Status: {batch.status}")
print(f"Completed: {batch.request_counts.completed}/{batch.request_counts.total}")

if batch.status == "completed":
    # Download results
    result_file = client.files.content(batch.output_file_id)
    results = result_file.text

    for line in results.strip().split("\n"):
        result = json.loads(line)
        print(f"{result['custom_id']}: {result['response']['body']['choices'][0]['message']['content']}")

import asyncio
import websockets
import json
import base64

async def realtime_conversation():
    url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17"
    headers = {
        "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
        "OpenAI-Beta": "realtime=v1"
    }

    async with websockets.connect(url, extra_headers=headers) as ws:
        # Configure session
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "voice": "alloy",
                "instructions": "You are a helpful assistant."
            }
        }))

        # Send user message
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Hello!"}]
            }
        }))

        # Request response
        await ws.send(json.dumps({"type": "response.create"}))

        # Receive response
        async for message in ws:
            event = json.loads(message)
            if event["type"] == "response.audio_transcript.delta":
                print(event["delta"], end="", flush=True)
            elif event["type"] == "response.done":
                break

asyncio.run(realtime_conversation())

{"messages": [{"role": "system", "content": "You are an SDD expert."}, {"role": "user", "content": "What is SDD?"}, {"role": "assistant", "content": "SDD (Specification-Driven Development) is a methodology..."}]}
{"messages": [{"role": "user", "content": "How to write a spec?"}, {"role": "assistant", "content": "To write a spec, start with..."}]}

# Upload training file
training_file = client.files.create(
    file=open("training_data.jsonl", "rb"),
    purpose="fine-tune"
)

# Create fine-tuning job
job = client.fine_tuning.jobs.create(
    training_file=training_file.id,
    model="gpt-4o-mini-2024-07-18",
    hyperparameters={
        "n_epochs": 3,
        "batch_size": "auto",
        "learning_rate_multiplier": "auto"
    },
    suffix="sdd-expert"
)

# Check status
job = client.fine_tuning.jobs.retrieve(job.id)
print(f"Status: {job.status}")

# List events
events = client.fine_tuning.jobs.list_events(job.id, limit=10)
for event in events.data:
    print(f"{event.created_at}: {event.message}")

response = client.chat.completions.create(
    model="ft:gpt-4o-mini-2024-07-18:org-xxx:sdd-expert:abc123",
    messages=[{"role": "user", "content": "Explain SDD phases"}]
)

import time
from openai import RateLimitError, APIError

def call_with_retry(func, max_retries=5, base_delay=1):
    for attempt in range(max_retries):
        try:
            return func()
        except RateLimitError as e:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"Rate limited. Retrying in {delay}s...")
            time.sleep(delay)
        except APIError as e:
            if e.status_code >= 500:
                if attempt == max_retries - 1:
                    raise
                delay = base_delay * (2 ** attempt)
                print(f"Server error. Retrying in {delay}s...")
                time.sleep(delay)
            else:
                raise

# Usage
response = call_with_retry(
    lambda: client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello"}]
    )
)

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import RateLimitError, APIError

@retry(
    retry=retry_if_exception_type((RateLimitError, APIError)),
    wait=wait_exponential(multiplier=1, min=1, max=60),
    stop=stop_after_attempt(5)
)
def make_request():
    return client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello"}]
    )

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

# Access headers
print(f"Requests remaining: {response.headers.get('x-ratelimit-remaining-requests')}")
print(f"Tokens remaining: {response.headers.get('x-ratelimit-remaining-tokens')}")
print(f"Reset in: {response.headers.get('x-ratelimit-reset-requests')}")

import tiktoken

def count_tokens(text: str, model: str = "gpt-4o") -> int:
    encoding = tiktoken.encoding_for_model(model)
    return len(encoding.encode(text))

# Count message tokens
messages = [
    {"role": "system", "content": "You are helpful."},
    {"role": "user", "content": "Hello world!"}
]

total = sum(count_tokens(m["content"]) for m in messages)
total += 3  # Overhead per message
print(f"Input tokens: {total}")

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

usage = response.usage
print(f"Prompt tokens: {usage.prompt_tokens}")
print(f"Completion tokens: {usage.completion_tokens}")
print(f"Total tokens: {usage.total_tokens}")

# Calculate cost
input_cost = usage.prompt_tokens * 2.50 / 1_000_000
output_cost = usage.completion_tokens * 10.00 / 1_000_000
total_cost = input_cost + output_cost
print(f"Cost: ${total_cost:.6f}")

class CostTracker:
    PRICES = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "text-embedding-3-large": {"input": 0.13, "output": 0},
        "dall-e-3-hd-1024": {"per_image": 0.08},
        "whisper-1": {"per_minute": 0.006},
        "tts-1-hd": {"per_char": 0.000030}
    }

    def __init__(self):
        self.total_cost = 0.0
        self.calls = []

    def track_completion(self, model: str, prompt_tokens: int, completion_tokens: int):
        prices = self.PRICES.get(model, {"input": 0, "output": 0})
        cost = (prompt_tokens * prices["input"] + completion_tokens * prices["output"]) / 1_000_000
        self.total_cost += cost
        self.calls.append({"model": model, "cost": cost})
        return cost

    def report(self):
        print(f"Total cost: ${self.total_cost:.4f}")
        print(f"Total calls: {len(self.calls)}")

tracker = CostTracker()

# Bad
messages = [{"role": "user", "content": "Write something about AI"}]

# Good
messages = [
    {
        "role": "system",
        "content": "You are a technical writer specializing in AI. Write clear, concise content for developers."
    },
    {
        "role": "user",
        "content": """Write a 200-word introduction about AI for a developer audience.

Requirements:
- Focus on practical applications
- Include one code example
- Use simple language"""
    }
]

import hashlib
import json

class ResponseCache:
    def __init__(self):
        self.cache = {}

    def get_key(self, model: str, messages: list) -> str:
        content = json.dumps({"model": model, "messages": messages}, sort_keys=True)
        return hashlib.md5(content.encode()).hexdigest()

    def get(self, model: str, messages: list):
        key = self.get_key(model, messages)
        return self.cache.get(key)

    def set(self, model: str, messages: list, response):
        key = self.get_key(model, messages)
        self.cache[key] = response

cache = ResponseCache()

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def parallel_completions(prompts: list[str]) -> list:
    async def get_completion(prompt):
        return await async_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}]
        )

    return await asyncio.gather(*[get_completion(p) for p in prompts])

# Usage
prompts = ["Hello", "World", "!"]
responses = asyncio.run(parallel_completions(prompts))

# Never log API keys
import logging
logging.getLogger("openai").setLevel(logging.WARNING)

# Validate user input
def sanitize_input(text: str, max_length: int = 10000) -> str:
    if not isinstance(text, str):
        raise ValueError("Input must be string")
    return text[:max_length].strip()

# Use environment variables
import os
api_key = os.environ.get("OPENAI_API_KEY")
if not api_key:
    raise ValueError("OPENAI_API_KEY not set")

# Chat completion
curl https://api.openai.com/v1/chat/completions \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "gpt-4o-mini",
    "messages": [{"role": "user", "content": "Hello!"}]
  }'

# Generate image
curl https://api.openai.com/v1/images/generations \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "dall-e-3",
    "prompt": "A robot programming",
    "size": "1024x1024"
  }'

# Transcribe audio
curl https://api.openai.com/v1/audio/transcriptions \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -F model="whisper-1" \
  -F file="@speech.mp3"

# Generate speech
curl https://api.openai.com/v1/audio/speech \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model": "tts-1", "voice": "alloy", "input": "Hello world!"}' \
  --output speech.mp3

# Create embedding
curl https://api.openai.com/v1/embeddings \
  -H "Authorization: Bearer $OPENAI_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model": "text-embedding-3-small", "input": "Hello world"}'