Last Updated: December 24, 2025
Quick Start
pip install openai
from openai import OpenAI
client = OpenAI(api_key="your-api-key")
# Create assistant
assistant = client.beta.assistants.create(
name="Math Tutor",
instructions="You are a helpful math tutor.",
model="gpt-4-turbo",
tools=[{"type": "code_interpreter"}]
)
# Create thread
thread = client.beta.threads.create()
# Add message
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="Solve: 3x + 11 = 14"
)
# Run assistant
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id
)
# Wait for completion and get response
# (see polling section below)
Create an Assistant
assistant = client.beta.assistants.create(
name="Customer Support Bot",
description="Helps with customer inquiries",
instructions="""You are a helpful customer support assistant.
Be friendly, concise, and professional. Always ask clarifying
questions when needed.""",
model="gpt-4-turbo",
tools=[
{"type": "code_interpreter"},
{"type": "file_search"},
{
"type": "function",
"function": {
"name": "get_order_status",
"description": "Get current order status",
"parameters": {
"type": "object",
"properties": {
"order_id": {
"type": "string",
"description": "The order ID"
}
},
"required": ["order_id"]
}
}
}
],
metadata={"version": "1.0"}
)
Available Tools
| Tool | Description | Use Case |
|---|---|---|
code_interpreter |
Execute Python code | Math, data analysis, visualizations |
file_search |
Search through uploaded files | RAG, document Q&A, knowledge bases |
function |
Call custom functions | API calls, database queries, actions |
Threads & Messages
# Create thread
thread = client.beta.threads.create(
metadata={"user_id": "user123"}
)
# Add user message
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="What's the weather like?",
metadata={"timestamp": "2025-12-24"}
)
# Add message with file
message_with_file = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="Analyze this data",
attachments=[{
"file_id": file.id,
"tools": [{"type": "code_interpreter"}]
}]
)
# List messages
messages = client.beta.threads.messages.list(
thread_id=thread.id
)
for msg in messages.data:
print(f"{msg.role}: {msg.content[0].text.value}")
Running Assistants
# Create and run
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=assistant.id,
additional_instructions="Be extra concise",
additional_messages=[
{"role": "user", "content": "Additional context"}
]
)
# Create with streaming
with client.beta.threads.runs.stream(
thread_id=thread.id,
assistant_id=assistant.id
) as stream:
for text in stream.text_deltas:
print(text, end="", flush=True)
# Poll for completion
import time
while run.status in ["queued", "in_progress"]:
run = client.beta.threads.runs.retrieve(
thread_id=thread.id,
run_id=run.id
)
time.sleep(1)
if run.status == "completed":
messages = client.beta.threads.messages.list(
thread_id=thread.id
)
print(messages.data[0].content[0].text.value)
File Search (RAG)
# Upload files
file = client.files.create(
file=open("knowledge.pdf", "rb"),
purpose="assistants"
)
# Create vector store
vector_store = client.beta.vector_stores.create(
name="Product Documentation",
file_ids=[file.id]
)
# Create assistant with file search
assistant = client.beta.assistants.create(
name="Documentation Assistant",
instructions="Answer questions using the provided docs",
model="gpt-4-turbo",
tools=[{"type": "file_search"}],
tool_resources={
"file_search": {
"vector_store_ids": [vector_store.id]
}
}
)
# Or attach to thread
thread = client.beta.threads.create(
messages=[{
"role": "user",
"content": "What does the documentation say about API limits?",
"attachments": [{
"file_id": file.id,
"tools": [{"type": "file_search"}]
}]
}]
)
Code Interpreter
# Create assistant with code interpreter
assistant = client.beta.assistants.create(
name="Data Analyst",
instructions="Analyze data and create visualizations",
model="gpt-4-turbo",
tools=[{"type": "code_interpreter"}]
)
# Upload data file
file = client.files.create(
file=open("sales_data.csv", "rb"),
purpose="assistants"
)
# Create thread with file
thread = client.beta.threads.create()
message = client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content="Create a bar chart of monthly sales",
attachments=[{
"file_id": file.id,
"tools": [{"type": "code_interpreter"}]
}]
)
# Run and get outputs
run = client.beta.threads.runs.create_and_poll(
thread_id=thread.id,
assistant_id=assistant.id
)
# Download generated images
if run.status == "completed":
messages = client.beta.threads.messages.list(thread_id=thread.id)
for msg in messages.data[0].content:
if msg.type == "image_file":
image_data = client.files.content(msg.image_file.file_id)
with open(f"chart_{msg.image_file.file_id}.png", "wb") as f:
f.write(image_data.content)
Function Calling
# When run requires action
if run.status == "requires_action":
tool_calls = run.required_action.submit_tool_outputs.tool_calls
tool_outputs = []
for tool_call in tool_calls:
if tool_call.function.name == "get_weather":
# Parse arguments
args = json.loads(tool_call.function.arguments)
# Execute function
result = get_weather(args["location"])
# Add to outputs
tool_outputs.append({
"tool_call_id": tool_call.id,
"output": json.dumps(result)
})
# Submit tool outputs
run = client.beta.threads.runs.submit_tool_outputs(
thread_id=thread.id,
run_id=run.id,
tool_outputs=tool_outputs
)
# Helper function for automatic handling
def handle_tool_calls(run, thread_id):
if run.status == "requires_action":
tool_outputs = []
for tool_call in run.required_action.submit_tool_outputs.tool_calls:
func_name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
# Call your function
result = globals()[func_name](**args)
tool_outputs.append({
"tool_call_id": tool_call.id,
"output": json.dumps(result)
})
return client.beta.threads.runs.submit_tool_outputs(
thread_id=thread_id,
run_id=run.id,
tool_outputs=tool_outputs
)
return run
Streaming Responses
from openai import AssistantEventHandler
class EventHandler(AssistantEventHandler):
def on_text_created(self, text):
print("\nassistant > ", end="", flush=True)
def on_text_delta(self, delta, snapshot):
print(delta.value, end="", flush=True)
def on_tool_call_created(self, tool_call):
print(f"\nassistant > {tool_call.type}\n", flush=True)
def on_message_done(self, message):
print("\n")
# Use event handler
with client.beta.threads.runs.stream(
thread_id=thread.id,
assistant_id=assistant.id,
event_handler=EventHandler()
) as stream:
stream.until_done()
Managing Assistants
# List assistants
assistants = client.beta.assistants.list(
order="desc",
limit=20
)
# Update assistant
assistant = client.beta.assistants.update(
assistant_id=assistant.id,
instructions="Updated instructions",
model="gpt-4-turbo",
temperature=0.7
)
# Delete assistant
client.beta.assistants.delete(assistant.id)
# Retrieve assistant
assistant = client.beta.assistants.retrieve(assistant.id)
Run Configuration
| Parameter | Description | Default |
|---|---|---|
temperature |
Randomness (0-2) | 1 |
top_p |
Nucleus sampling (0-1) | 1 |
max_prompt_tokens |
Limit input tokens | Model max |
max_completion_tokens |
Limit output tokens | Model max |
truncation_strategy |
How to handle long threads | auto |
Error Handling
from openai import OpenAIError
try:
run = client.beta.threads.runs.create_and_poll(
thread_id=thread.id,
assistant_id=assistant.id
)
if run.status == "failed":
print(f"Run failed: {run.last_error}")
elif run.status == "expired":
print("Run expired, retry needed")
except OpenAIError as e:
print(f"API error: {e}")
Best Practices
Use threads for conversations
One thread per user conversation
Clear instructions
Be specific about assistant behavior
Implement streaming
Better UX for long responses
Handle tool calls properly
Always check for requires_action status
Clean up resources
Delete unused threads and files
Use metadata
Track user IDs, sessions, etc.
💡 Pro Tip: Use create_and_poll() for simpler code - it handles the polling loop automatically. For production, implement streaming with event handlers for better user experience. Use vector stores for file search instead of attaching files directly to threads.