OpenAI API: Chat Completions, Function Calling, Structured Output Reference Guide

# Python SDK — official client
pip install openai          # ≥ 1.50 for Responses API
pip install openai[aiohttp] # async HTTP transport

# Node / TS SDK
npm install openai

# env — picked up automatically
export OPENAI_API_KEY=sk-proj-...
export OPENAI_ORG_ID=org-...        # optional, for multi-org
export OPENAI_PROJECT_ID=proj_...   # optional, scoped key

from openai import OpenAI
from pydantic import BaseModel

client = OpenAI()

class Invoice(BaseModel):
    vendor: str
    total: float
    line_items: list[str]

# Responses API — strict schema, no parsing
resp = client.responses.parse(
    model="gpt-4.1-mini",
    input="Extract from: ACME, $129.50, 2x widget, 1x cable",
    text_format=Invoice,
)

invoice: Invoice = resp.output_parsed  # typed, validated
print(invoice.vendor, invoice.total)

from openai import OpenAI
import json

client = OpenAI()

tools = [{
    "type": "function",
    "name": "get_weather",
    "description": "Current weather for a city.",
    "parameters": {
        "type": "object",
        "properties": {"city": {"type": "string"}},
        "required": ["city"],
    },
}]

resp = client.responses.create(
    model="gpt-4.1-mini",
    input=[{"role": "user", "content": "Weather in Mumbai?"}],
    tools=tools,
)

# Execute the model's tool calls and feed results back
for call in resp.output:
    if call.type == "function_call":
        result = {"temp_c": 31, "city": json.loads(call.arguments)["city"]}
        followup = client.responses.create(
            model="gpt-4.1-mini",
            previous_response_id=resp.id,
            input=[{
                "type": "function_call_output",
                "call_id": call.call_id,
                "output": json.dumps(result),
            }],
        )
        print(followup.output_text)

from openai import OpenAI

client = OpenAI()

# Responses API — SSE stream of typed events
stream = client.responses.create(
    model="gpt-4.1-mini",
    input="Write a haiku about Mumbai monsoons.",
    stream=True,
)

for event in stream:
    if event.type == "response.output_text.delta":
        print(event.delta, end="", flush=True)
    elif event.type == "response.completed":
        print(f"\n[done · {event.response.usage.total_tokens} tokens]")

from openai import OpenAI
import numpy as np

client = OpenAI()

# 1 · Embed your corpus (in real life, store these in a vector DB)
docs = [
    "Mumbai's monsoon runs June through September.",
    "The Bandra-Worli Sea Link opened in 2009.",
    "Vada pav originated in Mumbai in the 1960s.",
]
emb = client.embeddings.create(
    model="text-embedding-3-small",
    input=docs,
)
vectors = np.array([d.embedding for d in emb.data])

# 2 · Embed the query, cosine-rank
q = "When does the rainy season start?"
qv = np.array(client.embeddings.create(
    model="text-embedding-3-small", input=q,
).data[0].embedding)
top = docs[int(np.argmax(vectors @ qv))]

# 3 · Answer grounded in the top chunk
resp = client.responses.create(
    model="gpt-4.1-mini",
    instructions="Answer using only the context. Cite it verbatim.",
    input=f"Context: {top}\n\nQuestion: {q}",
)
print(resp.output_text)