openai/openai-python

OpenAI Python API library

)

The OpenAI Python library provides convenient access to the OpenAI REST API from any Python 3.10+ application. The library includes type definitions for all request params and response fields, and offers both synchronous and asynchronous clients powered by HTTPX2.

It is generated from our OpenAPI specification.

Documentation

The REST API documentation can be found on platform.openai.com. The full API of this library can be found in api.md.

Installation

# install from PyPI
pip install openai

Usage

The full API of this library can be found in api.md.

The primary API for interacting with OpenAI models is the Responses API. You can generate text from the model with the code below.

import os
from openai import OpenAI
client = OpenAI(
    # This is the default and can be omitted
    api_key=os.environ.get("OPENAI_API_KEY"),
)
response = client.responses.create(
    model="gpt-5.5",
    instructions="You are a coding assistant that talks like a pirate.",
    input="How do I check if a Python object is an instance of a class?",
)
print(response.output_text)

The previous standard (supported indefinitely) for generating text is the Chat Completions API. You can use that API to generate text from the model with the code below.

from openai import OpenAI
client = OpenAI()
completion = client.chat.completions.create(
    model="gpt-5.5",
    messages=[
        {"role": "developer", "content": "Talk like a pirate."},
        {
            "role": "user",
            "content": "How do I check if a Python object is an instance of a class?",
        },
    ],
)
print(completion.choices[0].message.content)

While you can provide an api_key keyword argument, we recommend using python-dotenv to add OPENAI_API_KEY="My API Key" to your .env file so that your API key is not stored in source control. Get an API key here.

Workload Identity Authentication

For secure, automated environments like cloud-managed Kubernetes, Azure, and Google Cloud Platform, you can use workload identity authentication with short-lived tokens from cloud identity providers instead of long-lived API keys.

Kubernetes (service account tokens)

from openai import OpenAI
from openai.auth import k8s_service_account_token_provider
client = OpenAI(
    workload_identity={
        "identity_provider_id": "idp-123",
        "service_account_id": "sa-456",
        "provider": k8s_service_account_token_provider(
            "/var/run/secrets/kubernetes.io/serviceaccount/token"
        ),
    },
)
response = client.chat.completions.create(
    model="gpt-5.5",
    messages=[{"role": "user", "content": "Hello!"}],
)

Azure (managed identity)

from openai import OpenAI
from openai.auth import azure_managed_identity_token_provider
client = OpenAI(
    workload_identity={
        "identity_provider_id": "idp-123",
        "service_account_id": "sa-456",
        "provider": azure_managed_identity_token_provider(
            resource="https://management.azure.com/",
        ),
    },
)

Google Cloud Platform (compute engine metadata)

from openai import OpenAI
from openai.auth import gcp_id_token_provider
client = OpenAI(
    workload_identity={
        "identity_provider_id": "idp-123",
        "service_account_id": "sa-456",
        "provider": gcp_id_token_provider(audience="https://api.openai.com/v1"),
    },
)

Custom subject token provider

from openai import OpenAI
def get_custom_token() -> str:
    return "your-jwt-token"
client = OpenAI(
    workload_identity={
        "identity_provider_id": "idp-123",
        "service_account_id": "sa-456",
        "provider": {
            "token_type": "jwt",
            "get_token": get_custom_token,
        },
    }
)

You can also customize the token refresh buffer (default is 1200 seconds (20 minutes) before expiration):

from openai import OpenAI
from openai.auth import k8s_service_account_token_provider
client = OpenAI(
    workload_identity={
        "identity_provider_id": "idp-123",
        "service_account_id": "sa-456",
        "provider": k8s_service_account_token_provider("/var/token"),
        "refresh_buffer_seconds": 120.0,
    }
)

X.509 workload identity (mutual TLS)

For X.509 workload identity federation, configure the client certificate and server trust on an HTTPX2 client, then pass only the identity-provider and service-account IDs to the SDK:

import os
import ssl
from openai import OpenAI, DefaultHttpx2Client
from openai.auth import x509_workload_identity
tls_context = ssl.create_default_context(
    cafile=os.getenv("OPENAI_MTLS_CA_BUNDLE"),
)
tls_context.load_cert_chain(
    certfile=os.environ["OPENAI_MTLS_CERTIFICATE_CHAIN"],
    keyfile=os.environ["OPENAI_MTLS_PRIVATE_KEY"],
    password=os.getenv("OPENAI_MTLS_PRIVATE_KEY_PASSWORD"),
)
client = OpenAI(
    workload_identity=x509_workload_identity(
        identity_provider_id=os.environ["OPENAI_IDENTITY_PROVIDER_ID"],
        service_account_id=os.environ["OPENAI_SERVICE_ACCOUNT_ID"],
        # refresh_buffer_seconds=120.0,
    ),
    http_client=DefaultHttpx2Client(
        verify=tls_context,
        follow_redirects=False,
    ),
)

X.509 mode defaults to https://mtls.api.openai.com/v1 when neither base_url nor OPENAI_BASE_URL is set. The same configured HTTP client presents its certificate to the fixed mTLS token-exchange endpoint and to the API. Tokens are exchanged lazily, cached, and refreshed automatically. Certificate files, private keys, passwords, server trust, proxies, and rotation remain application and transport concerns.

X.509 API requests require HTTPS and must stay on the configured API origin. The effective HTTP Host authority must match that origin. Provider API-key and proxy-only headers cannot be sent to the API alongside X.509 authentication. Token exchanges do not inherit API request hooks, authentication, or cookies. Identity settings are captured when the client is constructed; create a new client to change the identity. Azure clients do not support X.509 workload identity.

For asynchronous requests, use AsyncOpenAI with DefaultAsyncHttpx2Client. See the complete sync rollout-toggle example and async rollout-toggle example, which select API-key or X.509 authentication with the application-owned OPENAI_AUTH_MODE environment variable. X.509 workload identity currently supports HTTP APIs; Realtime and WebSockets are not included.

Vision

With an image URL:

prompt = "What is in this image?"
img_url = "https://api.nga.gov/iiif/a2e6da57-3cd1-4235-b20e-95dcaefed6c8/full/!800,800/0/default.jpg"
response = client.responses.create(
    model="gpt-5.5",
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": prompt},
                {"type": "input_image", "image_url": f"{img_url}"},
            ],
        }
    ],
)

With the image as a base64 encoded string:

import base64
from openai import OpenAI
client = OpenAI()
prompt = "What is in this image?"
with open("path/to/image.png", "rb") as image_file:
    b64_image = base64.b64encode(image_file.read()).decode("utf-8")
response = client.responses.create(
    model="gpt-5.5",
    input=[
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": prompt},
                {"type": "input_image", "image_url": f"data:image/png;base64,{b64_image}"},
            ],
        }
    ],
)

Async usage

Simply import AsyncOpenAI instead of OpenAI and use await with each API call:

import os
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI(
    # This is the default and can be omitted
    api_key=os.environ.get("OPENAI_API_KEY"),
)
async def main() -> None:
    response = await client.responses.create(
        model="gpt-5.5", input="Explain disestablishmentarianism to a smart five year old."
    )
    print(response.output_text)
asyncio.run(main())

Functionality between the synchronous and asynchronous clients is otherwise identical.

With aiohttp

By default, the async client uses HTTPX2. For improved concurrency performance, you may also use aiohttp as the HTTPX2 transport.

You can enable this by installing aiohttp:

# install from PyPI
pip install openai[aiohttp]

Then you can enable it by instantiating the client with http_client=DefaultAioHttpClient():

import os
import asyncio
from openai import DefaultAioHttpClient
from openai import AsyncOpenAI
async def main() -> None:
    async with AsyncOpenAI(
        api_key=os.environ.get("OPENAI_API_KEY"),  # This is the default and can be omitted
        http_client=DefaultAioHttpClient(),
    ) as client:
        chat_completion = await client.chat.completions.create(
            messages=[
                {
                    "role": "user",
                    "content": "Say this is a test",
                }
            ],
            model="gpt-5.5",
        )
asyncio.run(main())

HTTPX2 migration

HTTPX2 is the default HTTP client. If you configure a custom HTTP client, transport, timeout, authentication handler, event hook, or request mock, see the HTTPX2 migration guide.

Streaming responses

We provide support for streaming responses using Server-Sent Events (SSE).

from openai import OpenAI
client = OpenAI()
stream = client.responses.create(
    model="gpt-5.5",
    input="Write a one-sentence bedtime story about a unicorn.",
    stream=True,
)
for event in stream:
    print(event)

The async client uses the exact same interface.

import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main():
    stream = await client.responses.create(
        model="gpt-5.5",
        input="Write a one-sentence bedtime story about a unicorn.",
        stream=True,
    )
async for event in stream:
        print(event)
asyncio.run(main())

Realtime API

The Realtime API enables you to build low-latency, multi-modal conversational experiences. It currently supports text and audio as both input and output, as well as function calling through a WebSocket connection.

Under the hood the SDK uses the websockets library to manage connections.

The Realtime API works through a combination of client-sent events and server-sent events. Clients can send events to do things like update session configuration or send text and audio inputs. Server events confirm when audio responses have completed, or when a text response from the model has been received. A full event reference can be found here and a guide can be found here.

Basic text based example:

import asyncio
from openai import AsyncOpenAI
async def main():
    client = AsyncOpenAI()
async with client.realtime.connect(model="gpt-realtime-2") as connection:
        await connection.session.update(
            session={"type": "realtime", "output_modalities": ["text"]}
        )
await connection.conversation.item.create(
            item={
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "Say hello!"}],
            }
        )
        await connection.response.create()
async for event in connection:
            if event.type == "response.output_text.delta":
                print(event.delta, flush=True, end="")
elif event.type == "response.output_text.done":
                print()
elif event.type == "response.done":
                break
asyncio.run(main())

However the real magic of the Realtime API is handling audio inputs / outputs, see this example TUI script for a fully fledged example.

Realtime error handling

Whenever an error occurs, the Realtime API will send an error event and the connection will stay open and remain usable. This means you need to handle it yourself, as _no errors are raised directly_ by the SDK when an error event comes in.

client = AsyncOpenAI()
async with client.realtime.connect(model="gpt-realtime-2") as connection:
    ...
    async for event in connection:
        if event.type == 'error':
            print(event.error.type)
            print(event.error.code)
            print(event.error.event_id)
            print(event.error.message)

Using types

Nested request parameters are TypedDicts. Responses are Pydantic models which also provide helper methods for things like:

• Serializing back into JSON, model.to_json()
• Converting to a dictionary, model.to_dict()

Typed requests and responses provide autocomplete and documentation within your editor. If you would like to see type errors in VS Code to help catch bugs earlier, set python.analysis.typeCheckingMode to basic.

Pagination

List methods in the OpenAI API are paginated.

This library provides auto-paginating iterators with each list response, so you do not have to request successive pages manually:

from openai import OpenAI
client = OpenAI()
all_jobs = []
Automatically fetches more pages as needed.
for job in client.fine_tuning.jobs.list(
    limit=20,
):
    # Do something with job here
    all_jobs.append(job)
print(all_jobs)

Or, asynchronously:

import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def main() -> None:
    all_jobs = []
    # Iterate through items across all pages, issuing requests as needed.
    async for job in client.fine_tuning.jobs.list(
        limit=20,
    ):
        all_jobs.append(job)
    print(all_jobs)
asyncio.run(main())

Alternatively, you can use the .has_next_page(), .next_page_info(), or .get_next_page() methods for more granular control working with pages:

first_page = await client.fine_tuning.jobs.list(
    limit=20,
)
if first_page.has_next_page():
    print(f"will fetch next page using these details: {first_page.next_page_info()}")
    next_page = await first_page.get_next_page()
    print(f"number of items we just fetched: {len(next_page.data)}")
Remove await for non-async usage.

Or just work directly with the returned data:

first_page = await client.fine_tuning.jobs.list(
    limit=20,
)
print(f"next page cursor: {first_page.after}")  # => "next page cursor: ..."
for job in first_page.data:
    print(job.id)
Remove await for non-async usage.

Nested params

Nested parameters are dictionaries, typed using TypedDict, for example:

from openai import OpenAI
client = OpenAI()
response = client.responses.create(
    input=[
        {
            "role": "user",
            "content": "How much ?",
        }
    ],
    model="gpt-5.5",
    text={"format": {"type": "json_object"}},
)

File uploads

Request parameters that correspond to file uploads can be passed as bytes, a file-like object, a PathLike instance, or a tuple of (filename, contents, media type).

from pathlib import Path
from openai import OpenAI
client = OpenAI()
client.files.create(
    file=Path("input.jsonl"),
    purpose="fine-tune",
)

When uploading an in-memory file-like object such as io.BytesIO, include a filename when the API needs the file extension to determine its format. Passing a tuple is the most explicit option:

import io
audio = io.BytesIO(audio_bytes)
transcription = client.audio.transcriptions.create(
    model="gpt-4o-transcribe",
    file=("audio.wav", audio, "audio/wav"),
)

You can also set a .name attribute such as audio.name = "audio.wav" on a mutable file-like object before passing it directly. Without a filename, multipart transports may use a generic name such as upload, which does not provide an audio extension for format detection.

The async client uses the exact same interface. If you pass a PathLike instance, the file contents will be read asynchronously automatically.

Webhook Verification

Verifying webhook signatures is _optional but encouraged_.

For more information about webhooks, see the API docs.

Parsing webhook payloads

For most use cases, you will likely want to verify the webhook and parse the payload at the same time. To achieve this, we provide the method client.webhooks.unwrap(), which parses a webhook request and verifies that it was sent by OpenAI. This method will raise an error if the signature is invalid.

Note that the body parameter must be the raw JSON string sent from the server (do not parse it first). The .unwrap() method will parse this JSON for you into an event object after verifying the webhook was sent from OpenAI.

from openai import OpenAI
from flask import Flask, request
app = Flask(__name__)
client = OpenAI()  # OPENAI_WEBHOOK_SECRET environment variable is used by default
@app.route("/webhook", methods=["POST"])
def webhook():
    request_body = request.get_data(as_text=True)
try:
        event = client.webhooks.unwrap(request_body, request.headers)
if event.type == "response.completed":
            print("Response completed:", event.data)
        elif event.type == "response.failed":
            print("Response failed:", event.data)
        else:
            print("Unhandled event type:", event.type)
return "ok"
    except Exception as e:
        print("Invalid signature:", e)
        return "Invalid signature", 400
if __name__ == "__main__":
    app.run(port=8000)

Verifying webhook payloads directly

In some cases, you may want to verify the webhook separately from parsing the payload. If you prefer to handle these steps separately, we provide the method client.webhooks.verify_signature() to _only verify_ the signature of a webhook request. Like .unwrap(), this method will raise an error if the signature is invalid.

Note that the body parameter must be the raw JSON string sent from the server (do not parse it first). You will then need to parse the body after verifying the signature.

import json
from openai import OpenAI
from flask import Flask, request
app = Flask(__name__)
client = OpenAI()  # OPENAI_WEBHOOK_SECRET environment variable is used by default
@app.route("/webhook", methods=["POST"])
def webhook():
    request_body = request.get_data(as_text=True)
try:
        client.webhooks.verify_signature(request_body, request.headers)
# Parse the body after verification
        event = json.loads(request_body)
        print("Verified event:", event)
return "ok"
    except Exception as e:
        print("Invalid signature:", e)
        return "Invalid signature", 400
if __name__ == "__main__":
    app.run(port=8000)

Handling errors

When the library is unable to connect to the API (for example, due to network connection problems or a timeout), a subclass of openai.APIConnectionError is raised.

When the API returns a non-success status code (that is, 4xx or 5xx response), a subclass of openai.APIStatusError is raised, containing status_code and response properties.

All errors inherit from openai.APIError.

When consuming a Stream or AsyncStream, read timeouts raise APITimeoutError and other HTTPX request failures raise APIConnectionError. Catch these SDK exceptions instead of raw HTTPX exceptions; the original exception is available as __cause__. Stream consumption is not automatically retried, because replaying a request could duplicate output already delivered to your application. The Assistants event-handler helpers and raw with_streaming_response iterators retain their existing exception behavior.

import openai
from openai import OpenAI
client = OpenAI()
try:
    client.fine_tuning.jobs.create(
        model="gpt-4o",
        training_file="file-abc123",
    )
except openai.APIConnectionError as e:
    print("The server could not be reached")
    print(e.__cause__)  # an underlying Exception, likely raised within HTTPX2.
except openai.RateLimitError as e:
    print("A 429 status code was received; we should back off a bit.")
except openai.APIStatusError as e:
    print("Another non-200-range status code was received")
    print(e.status_code)
    print(e.response)

Error codes are as follows:

| Status Code | Error Type | | ----------- | -------------------------- | | 400 | BadRequestError | | 401 | AuthenticationError | | 403 | PermissionDeniedError | | 404 | NotFoundError | | 422 | UnprocessableEntityError | | 429 | RateLimitError | | >=500 | InternalServerError | | N/A | APIConnectionError |

Request IDs

For more information on debugging requests, see these docs

All object responses in the SDK provide a _request_id property which is added from the x-request-id response header so that you can quickly log failing requests and report them back to OpenAI.

response = await client.responses.create(
    model="gpt-5.5",
    input="Say 'this is a test'.",
)
print(response._request_id)  # req_123

Note that unlike other properties that use an _ prefix, the _request_id property _is_ public. Unless documented otherwise, _all_ other _ prefix properties, methods and modules are _private_.

[!IMPORTANT]

If you need to access request IDs for failed requests you must catch the APIStatusError exception

import openai
try:
    completion = await client.chat.completions.create(
        messages=[{"role": "user", "content": "Say this is a test"}], model="gpt-5.5"
    )
except openai.APIStatusError as exc:
    print(exc.request_id)  # req_123
    raise exc

Retries

Certain errors are automatically retried 2 times by default, with a short exponential backoff. Connection errors (for example, due to a network connectivity problem), 408 Request Timeout, 409 Conflict, 429 Rate Limit, and >=500 Internal errors are all retried by default.

You can use the max_retries option to configure or disable retry settings:

from openai import OpenAI
Configure the default for all requests:
client = OpenAI(
    # default is 2
    max_retries=0,
)
Or, configure per-request:
client.with_options(max_retries=5).chat.completions.create(
    messages=[
        {
            "role": "user",
            "content": "How can I get the name of the current day in JavaScript?",
        }
    ],
    model="gpt-5.5",
)

max_retries