OpenAI API Guide: Python Examples for Production

Introduction

The OpenAI API looks simple until you run it in production: streaming stalls, rate limits hit at the worst moment, a “working” integration silently burns tokens, and the API surface itself moves under your feet. The biggest move recently is that OpenAI consolidated everything around the Responses API, which absorbs chat, tool calling and structured output into one interface — the Assistants API it replaced is now deprecated.

This guide covers both surfaces honestly: Responses API as the default for new work, chat completions because most production code (and most model providers) still speaks it, plus streaming, embeddings, fine-tuning, batch jobs and the error handling you’ll actually need. Examples are Python-first, with JavaScript variants where they differ meaningfully. Runnable versions live in the companion repo.

Responses API

The Responses API is OpenAI’s current recommended interface. One responses.create call handles plain chat, built-in tools, function calling and structured output, and it can keep server-side state so you don’t resend the whole conversation on every turn.

Basic Response

from openai import OpenAI

client = OpenAI()

response = client.responses.create(
    model="gpt-4o",
    input="Explain Python decorators in 2 sentences."
)

print(response.output_text)
print(f"Tokens used: {response.usage.total_tokens}")

output_text flattens the response for you — no digging through choices[0].message.content. For multi-turn conversations, chain responses with previous_response_id instead of managing the message list yourself:

first = client.responses.create(
    model="gpt-4o",
    input="You are a Python expert. What is a decorator?"
)

follow_up = client.responses.create(
    model="gpt-4o",
    previous_response_id=first.id,
    input="Show me an example."
)

print(follow_up.output_text)

Structured Output with responses.parse

For typed output, responses.parse accepts a Pydantic model directly and validates the result:

from pydantic import BaseModel

class Person(BaseModel):
    name: str
    age: int
    email: str
    skills: list[str]

response = client.responses.parse(
    model="gpt-4o",
    input=[
        {"role": "system", "content": "Extract person information."},
        {"role": "user", "content": "Jane Smith, 28, jane@test.com. Skills: Python, SQL, Docker."}
    ],
    text_format=Person,
)

person = response.output_parsed
print(f"Name: {person.name}, Age: {person.age}")

This is the cleanest path when you need guaranteed fields — no JSON parsing, no schema validation by hand.

Function Calling on Responses API

Function tools have a flattened shape here (no nested function key), and calls come back as items in response.output:

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

response = client.responses.create(
    model="gpt-4o",
    input="What's the weather in Madrid?",
    tools=tools
)

for item in response.output:
    if item.type == "function_call":
        result = get_weather(**json.loads(item.arguments))
        # Send the tool result back as function_call_output
        follow = client.responses.create(
            model="gpt-4o",
            previous_response_id=response.id,
            input=[{
                "type": "function_call_output",
                "call_id": item.call_id,
                "output": result
            }]
        )
        print(follow.output_text)

Chat Completions

Chat completions isn’t going anywhere: it’s what most existing codebases use, what OpenAI-compatible providers implement, and it’s still perfectly supported. If you’re maintaining an existing integration, there’s no urgency to migrate — for new projects, prefer the Responses API above.

Basic Completion

from openai import OpenAI

client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain Python decorators in 2 sentences."}
    ],
    temperature=0.7,
    max_tokens=200
)

print(response.choices[0].message.content)
print(f"Tokens used: {response.usage.total_tokens}")

Multi-Turn Conversation

Unlike the Responses API, chat completions is stateless — you send the full history every call. Keeping the conversation in a list and appending both sides is the standard pattern:

conversation = [
    {"role": "system", "content": "You are a Python expert."},
    {"role": "user", "content": "What is a decorator?"}
]

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

assistant_msg = response.choices[0].message.content
conversation.append({"role": "assistant", "content": assistant_msg})

conversation.append({"role": "user", "content": "Show me an example."})
response = client.chat.completions.create(model="gpt-4o", messages=conversation)
print(response.choices[0].message.content)

Watch the token cost here: long conversations resend the entire history on every call, so history truncation or summarization becomes mandatory past a few dozen turns.

Model Selection

Pricing moves, so treat the numbers as directional and check the pricing page before budgeting:

Model	Best for	Input ($/1M)	Output ($/1M)
gpt-4o	All-around, multimodal, complex tasks	~2.50	~10.00
gpt-4o-mini	Simple tasks, high volume	~0.15	~0.60
o1	Multi-step reasoning	~15.00	~60.00
o1-mini	Math/code reasoning, cheaper	~3.00	~12.00
text-embedding-3-small	Cheapest embeddings	~0.02	—
text-embedding-3-large	Highest-quality embeddings, 3072 dims	~0.13	—

A practical default: route everything to gpt-4o-mini and upgrade to gpt-4o only where evals show it matters. For a deeper treatment, see the LLM cost optimization guide.

Streaming

Streaming doesn’t make the model faster — total latency is the same — but time-to-first-token drops to a few hundred milliseconds, which is what users actually perceive. Stream whenever a human is waiting on the output; skip it for background jobs where it only adds complexity.

Server-Sent Events Streaming

from openai import OpenAI

client = OpenAI()

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

full_text = ""
for chunk in stream:
    if chunk.choices[0].delta.content is not None:
        token = chunk.choices[0].delta.content
        full_text += token
        print(token, end="", flush=True)

print(f"\n\nFull text length: {len(full_text)}")

JavaScript Streaming

The Node SDK exposes the same pattern as an async iterator:

import OpenAI from "openai";

const client = new OpenAI();

const stream = await client.chat.completions.create({
  model: "gpt-4o",
  messages: [{ role: "user", content: "Explain async programming." }],
  stream: true,
});

for await (const chunk of stream) {
  process.stdout.write(chunk.choices[0]?.delta?.content ?? "");
}

Streaming with FastAPI

The production pattern: wrap the SDK stream in a generator and return it as a StreamingResponse. Note the endpoint yields SSE-formatted chunks, so the client must parse data: lines — the OpenAI SDK events aren’t plain text.

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
import json

app = FastAPI()
client = AsyncOpenAI()

@app.post("/chat/stream")
async def chat_stream(request: dict):
    async def generate():
        stream = await client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": request["message"]}
            ],
            stream=True
        )

        async for chunk in stream:
            if chunk.choices[0].delta.content:
                data = json.dumps({"token": chunk.choices[0].delta.content})
                yield f"data: {data}\n\n"

        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

For a complete SSE implementation including client consumption, see the recipe on streaming LLM output with SSE.

Function Calling

Function calling turns the model into a router: you declare what functions exist, the model decides when to call them and with what arguments, and your code executes the real work. The model never runs your code — it only proposes calls. That loop looks like this:

Defining Functions on Chat Completions

from openai import OpenAI
import json

client = OpenAI()

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. 'Madrid, Spain'"
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature unit"
                    }
                },
                "required": ["location"]
            }
        }
    }
]

def get_weather(location: str, unit: str = "celsius") -> str:
    # Simulated API call
    return f"Weather in {location}: 22°{unit[0].upper()}, sunny"

function_map = {"get_weather": get_weather}

messages = [{"role": "user", "content": "What's the weather in Madrid?"}]

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

tool_calls = response.choices[0].message.tool_calls

if tool_calls:
    messages.append(response.choices[0].message)

    for call in tool_calls:
        func_args = json.loads(call.function.arguments)
        result = function_map[call.function.name](**func_args)

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

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

Two production details people skip: always validate the arguments the model returns (it’s a suggestion, not a guarantee), and bound the loop — a model that keeps proposing tool calls can burn tokens indefinitely. The recipe on function calling with structured output covers multi-tool setups in depth.

Structured Outputs

Two levels of guarantee exist. JSON mode forces syntactically valid JSON but says nothing about the shape. Structured outputs with json_schema enforces your exact schema — field names, types, required keys — which is what you want when downstream code consumes the result.

JSON Mode

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a data extractor. Return valid JSON only."},
        {"role": "user", "content": "Extract name, age, and email from: John Doe, 30, john@example.com"}
    ],
    response_format={"type": "json_object"}
)

import json
data = json.loads(response.choices[0].message.content)
print(data)  # {"name": "John Doe", "age": 30, "email": "john@example.com"}

Structured Outputs with Schema

from pydantic import BaseModel

class Person(BaseModel):
    name: str
    age: int
    email: str
    skills: list[str]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "Extract person information."},
        {"role": "user", "content": "Jane Smith, 28, jane@test.com. Skills: Python, SQL, Docker."}
    ],
    response_format={
        "type": "json_schema",
        "json_schema": {
            "name": "person",
            "schema": {
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "age": {"type": "integer"},
                    "email": {"type": "string"},
                    "skills": {"type": "array", "items": {"type": "string"}}
                },
                "required": ["name", "age", "email", "skills"],
                "additionalProperties": False
            },
            "strict": True
        }
    }
)

person = Person(**json.loads(response.choices[0].message.content))
print(f"Name: {person.name}, Age: {person.age}, Skills: {person.skills}")

On the Responses API, responses.parse with text_format (shown above) gives you the same guarantee with less boilerplate. Whichever you use, additionalProperties: false and strict: true are what actually enforce the schema — omit them and the model may add fields you didn’t ask for.

Embeddings

Embeddings turn text into vectors so you can compare meaning numerically — search, RAG retrieval, dedup and clustering all build on them. The cheap lesson most teams learn late: embeddings are nearly free per call, but they get expensive at corpus scale, so dimension and batch size matter.

Generating Embeddings

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=["Python is a programming language", "Java is also a programming language"]
)

embeddings = [d.embedding for d in response.data]
print(f"Embedding dimensions: {len(embeddings[0])}")  # 1536

# Cosine similarity
import numpy as np
sim = np.dot(embeddings[0], embeddings[1]) / (
    np.linalg.norm(embeddings[0]) * np.linalg.norm(embeddings[1])
)
print(f"Similarity: {sim:.4f}")

Dimensionality Reduction

The dimensions parameter trades a little recall for real storage and index-size savings — useful when you’re embedding millions of documents:

# Reduce from 3072 to 256 dimensions (for storage savings)
response = client.embeddings.create(
    model="text-embedding-3-large",
    input="Python programming language",
    dimensions=256
)

print(f"Reduced dimensions: {len(response.data[0].embedding)}")  # 256

For similarity search patterns and index building, see OpenAI embeddings with cosine similarity.

Fine-Tuning

Fine-tuning is the last tool, not the first. Prompt engineering and few-shot examples get you most of the way; fine-tune when you need consistent format/tone at scale, when you’re hitting context limits with examples, or when you want to run a smaller (cheaper, faster) model with your domain baked in. The real cost isn’t the training job — it’s curating the data.

Preparing Training Data

import json

training_data = [
    {
        "messages": [
            {"role": "system", "content": "You are a customer support agent for TechCorp."},
            {"role": "user", "content": "How do I reset my password?"},
            {"role": "assistant", "content": "To reset your password: 1) Go to techcorp.com/reset 2) Enter your email 3) Click the link in the email 4) Set a new password."}
        ]
    },
    {
        "messages": [
            {"role": "system", "content": "You are a customer support agent for TechCorp."},
            {"role": "user", "content": "What are your business hours?"},
            {"role": "assistant", "content": "Our business hours are Monday-Friday, 9 AM to 6 PM CET. We also offer 24/7 email support."}
        ]
    }
]

# Save as JSONL
with open("training_data.jsonl", "w") as f:
    for item in training_data:
        f.write(json.dumps(item) + "\n")

Fifty excellent examples beat five hundred mediocre ones — bad examples teach the model your bad habits. Plan for evaluation before you train, not after.

Creating a Fine-Tuned Model

from openai import OpenAI

client = OpenAI()

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

# Create fine-tune job
job = client.fine_tuning.jobs.create(
    training_file=file.id,
    model="gpt-4o-mini",
    hyperparameters={
        "n_epochs": 3,
        "batch_size": 4,
        "learning_rate_multiplier": 0.5
    }
)

print(f"Fine-tune job ID: {job.id}")
print(f"Status: {job.status}")

# Check status
import time
while True:
    job = client.fine_tuning.jobs.retrieve(job.id)
    print(f"Status: {job.status}")
    if job.status in ["succeeded", "failed"]:
        break
    time.sleep(30)

if job.status == "succeeded":
    print(f"Fine-tuned model: {job.fine_tuned_model}")

    # Use the fine-tuned model
    response = client.chat.completions.create(
        model=job.fine_tuned_model,
        messages=[
            {"role": "system", "content": "You are a customer support agent for TechCorp."},
            {"role": "user", "content": "How do I cancel my subscription?"}
        ]
    )
    print(response.choices[0].message.content)

For the full workflow including validation sets and evals, see the LLM fine-tuning recipe.

Batch API

The Batch API trades latency for price: jobs run within a 24-hour window at roughly half the standard cost. It fits anything that doesn’t need a response right now — bulk classification, offline evals, nightly enrichment. The awkward part is the file-based workflow: you write JSONL, upload it, poll, then download results.

Submitting a Batch

import json

# Prepare batch requests
requests = [
    {
        "custom_id": f"request-{i}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [
                {"role": "system", "content": "Classify the sentiment as positive, negative, or neutral."},
                {"role": "user", "content": f"Review: {review}"}
            ]
        }
    }
    for i, review in enumerate(reviews)
]

# Save as JSONL
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"
)

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

# Check status (batch takes minutes to hours)
batch = client.batches.retrieve(batch.id)
if batch.status == "completed":
    # Download results
    result_file = client.files.content(batch.output_file_id)
    results = result_file.text.strip().split("\n")

    for line in results:
        result = json.loads(line)
        if result.get("error"):
            print(f"Error: {result['error']}")
        else:
            content = result["response"]["body"]["choices"][0]["message"]["content"]
            print(f"{result['custom_id']}: {content}")

custom_id is the only thing linking results back to your input — put your database keys there, not just sequence numbers.

Assistants API (Deprecated)

The Assistants API — threads, runs, and client.beta.assistants — is deprecated and scheduled for shutdown; OpenAI directs new integrations to the Responses API, which covers the same use cases (persistent state, tools, file search) with a simpler model. If you maintain an existing Assistant, the migration path is: threads become previous_response_id chains or server-side state, tool wiring becomes function tools, and code interpreter becomes a Responses API built-in tool.

For legacy code you’re not migrating yet, the pattern below still works — just don’t build anything new on it:

# Legacy pattern — migrate to Responses API for new work
assistant = client.beta.assistants.create(
    name="Code Reviewer",
    model="gpt-4o",
    instructions="You are a senior code reviewer.",
    tools=[{"type": "code_interpreter"}]
)
thread = client.beta.threads.create()
client.beta.threads.messages.create(
    thread_id=thread.id, role="user", content="Review: def add(a,b): return a+b"
)
run = client.beta.threads.runs.create(thread_id=thread.id, assistant_id=assistant.id)

Rate Limits and Error Handling

Rate limits are per-model and split into requests-per-minute and tokens-per-minute — you can blow the token bucket while staying under the request cap. Every production integration needs two things: a client-side limiter that stays under your tier, and backoff with jitter for when you inevitably exceed it anyway.

Rate Limit Handling

import asyncio
import time
from openai import AsyncOpenAI, RateLimitError

client = AsyncOpenAI()

async def call_with_backoff(messages, max_retries=5):
    for attempt in range(max_retries):
        try:
            response = await client.chat.completions.create(
                model="gpt-4o",
                messages=messages
            )
            return response

        except RateLimitError as e:
            # Parse retry-after header
            retry_after = getattr(e, "retry_after", 2 ** attempt)
            print(f"Rate limited. Retrying in {retry_after}s...")
            await asyncio.sleep(retry_after)

    raise RuntimeError("Max retries exceeded")

# Token bucket rate limiter
class RateLimiter:
    def __init__(self, requests_per_minute: int):
        self.interval = 60.0 / requests_per_minute
        self.last_request = 0.0

    async def wait(self):
        now = time.time()
        elapsed = now - self.last_request
        if elapsed < self.interval:
            await asyncio.sleep(self.interval - elapsed)
        self.last_request = time.time()

limiter = RateLimiter(requests_per_minute=50)  # RPM limit

async def rate_limited_call(messages):
    await limiter.wait()
    return await client.chat.completions.create(model="gpt-4o", messages=messages)

Error Types

Map each exception to a different reaction: AuthenticationError is a dead end (alert, don’t retry), RateLimitError waits, APITimeoutError/APIConnectionError retry, and BadRequestError means your request is broken — retrying just fails again slower.

from openai import (
    APIError,
    RateLimitError,
    APIConnectionError,
    APITimeoutError,
    AuthenticationError,
    BadRequestError
)

async def robust_call(messages):
    try:
        return await client.chat.completions.create(model="gpt-4o", messages=messages)

    except AuthenticationError:
        # Invalid API key
        raise RuntimeError("Invalid API key. Check OPENAI_API_KEY.")

    except BadRequestError as e:
        # Invalid request (bad model, bad params)
        raise RuntimeError(f"Bad request: {e}")

    except RateLimitError:
        # Hit rate limit
        await asyncio.sleep(60)
        return await robust_call(messages)

    except APITimeoutError:
        # Request timed out
        return await robust_call(messages)

    except APIConnectionError:
        # Network error
        await asyncio.sleep(5)
        return await robust_call(messages)

    except APIError as e:
        # Generic API error
        raise RuntimeError(f"OpenAI API error: {e}")

Cost Optimization

Most cost problems trace to three things: the wrong model for the job, resending context you don’t need, and unbounded max_tokens. Before any fancy optimization, put a tracker in front of the client so you can see which calls actually cost money — what you don’t measure gets optimized blind.

class CostTracker:
    PRICING = {
        "gpt-4o": {"input": 2.50, "output": 10.00},
        "gpt-4o-mini": {"input": 0.15, "output": 0.60},
        "text-embedding-3-small": {"input": 0.02, "output": 0.0},
    }

    def __init__(self):
        self.costs: list[dict] = []

    def track(self, model: str, input_tokens: int, output_tokens: int):
        if model not in self.PRICING:
            return

        pricing = self.PRICING[model]
        cost = (input_tokens / 1_000_000 * pricing["input"]) + \
               (output_tokens / 1_000_000 * pricing["output"])

        self.costs.append({
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": cost
        })

    def total_cost(self) -> float:
        return sum(c["cost"] for c in self.costs)

    def summary(self) -> dict:
        by_model = {}
        for c in self.costs:
            if c["model"] not in by_model:
                by_model[c["model"]] = {"calls": 0, "cost": 0.0, "tokens": 0}
            by_model[c["model"]]["calls"] += 1
            by_model[c["model"]]["cost"] += c["cost"]
            by_model[c["model"]]["tokens"] += c["input_tokens"] + c["output_tokens"]

        return {
            "total_cost": self.total_cost(),
            "by_model": by_model,
            "total_calls": len(self.costs)
        }

# Usage
tracker = CostTracker()
response = client.chat.completions.create(model="gpt-4o", messages=[...])
tracker.track("gpt-4o", response.usage.prompt_tokens, response.usage.completion_tokens)
print(f"Total cost: ${tracker.total_cost():.4f}")

Common Pitfalls

• Retrying BadRequestError in a loop — a malformed request fails forever; fix the request, not the retry policy.
• No ceiling on tool-call loops — cap iterations; a confused model proposes calls indefinitely.
• Resending full history on every turn — truncate or summarize past ~20 turns, or use previous_response_id.
• API key in client-side code — keys in shipped JS get scraped within hours; proxy calls through your backend.
• Fine-tuning before prompting — tune only after prompt engineering plateaus on evals, not on vibes.
• Counting requests, not tokens — TPM limits trigger on tokens; a few giant prompts can 429 a low-traffic app.

See Also

• Complete Guide to LLM Application Architecture
• Complete Guide to LLM Cost Optimization
• Complete Guide to LLM Evaluation
• Stream LLM Output with Server-Sent Events (SSE)
• API Rate Limiting — Design Fair and Useful Throttling

Further Reading

• OpenAI API reference — endpoints, parameters and response shapes
• OpenAI pricing — per-model and per-tier rates for budgeting
• Rate limits guide — usage tiers and limit headers
• Batch API guide — 50% discounts on async workloads
• Production best practices — key safety, spend limits and scaling
• OpenAI Cookbook — runnable examples maintained by OpenAI