OpenAI Assistants API — Build Stateful AI Agents 2026
Advertisement
Introduction
Why This Matters
The standard OpenAI Chat Completions API is stateless — you manage conversation history yourself. The Assistants API handles state for you through persistent Threads, stores and retrieves files via Vector Stores, runs code in a sandboxed Code Interpreter, and orchestrates multi-step tool calls in a Run. For production agents, this infrastructure eliminates weeks of custom state management code.
Core Concepts
The Assistants API introduces four objects:
Assistant — A configured AI instance with a model, instructions, and enabled tools. Think of it as a persistent system prompt configuration.
Thread — A conversation session. Messages accumulate here and OpenAI manages context window limits automatically.
Message — A single user or assistant contribution within a Thread.
Run — The execution of an Assistant on a Thread. Runs can call tools, retrieve files, and loop through multiple reasoning steps.
Setup and First Assistant
from openai import OpenAI
client = OpenAI()
# Create an Assistant (do this once; save the ID)
assistant = client.beta.assistants.create(
name="Code Reviewer",
instructions=(
"You are a senior software engineer specializing in code review. "
"Review code for: security vulnerabilities, performance issues, "
"missing error handling, and style violations. "
"Format findings as a numbered list with severity (critical/medium/low)."
),
model="gpt-4o",
tools=[{"type": "code_interpreter"}], # Enable sandbox code execution
)
print(f"Assistant ID: {assistant.id}")
# Save assistant.id — reuse it across sessionsRunning a Conversation
import time
from openai import OpenAI
client = OpenAI()
ASSISTANT_ID = "asst_your_id_here"
def review_code(code: str) -> str:
"""Submit code for review and wait for the result."""
# 1. Create a Thread (one per conversation)
thread = client.beta.threads.create()
# 2. Add a Message to the Thread
client.beta.threads.messages.create(
thread_id=thread.id,
role="user",
content=f"Review this code:\n\n```python\n{code}\n```",
)
# 3. Create a Run (triggers the Assistant to respond)
run = client.beta.threads.runs.create(
thread_id=thread.id,
assistant_id=ASSISTANT_ID,
)
# 4. Poll until the Run completes
while run.status in ("queued", "in_progress"):
time.sleep(1)
run = client.beta.threads.runs.retrieve(
thread_id=thread.id, run_id=run.id
)
if run.status != "completed":
raise RuntimeError(f"Run failed with status: {run.status}")
# 5. Retrieve the Assistant's response
messages = client.beta.threads.messages.list(thread_id=thread.id)
return messages.data[0].content[0].text.value
# Usage
code = """
def get_user(user_id):
query = "SELECT * FROM users WHERE id = " + user_id
return db.execute(query)
"""
print(review_code(code))File Search with Vector Stores
File Search lets the Assistant retrieve information from uploaded documents. Build a documentation assistant that answers questions from your internal docs:
from openai import OpenAI
import time
client = OpenAI()
def build_docs_assistant(doc_paths: list[str]) -> str:
"""Upload documents and create an assistant that searches them."""
# 1. Create a Vector Store
vector_store = client.beta.vector_stores.create(name="Project Docs")
# 2. Upload files
file_streams = [open(path, "rb") for path in doc_paths]
batch = client.beta.vector_stores.file_batches.upload_and_poll(
vector_store_id=vector_store.id,
files=file_streams,
)
for f in file_streams:
f.close()
# 3. Create Assistant with file_search tool
assistant = client.beta.assistants.create(
name="Docs Assistant",
instructions=(
"Answer questions using the provided documentation. "
"Quote relevant passages and cite file names. "
"If you cannot find an answer, say so clearly."
),
model="gpt-4o",
tools=[{"type": "file_search"}],
tool_resources={"file_search": {"vector_store_ids": [vector_store.id]}},
)
return assistant.id
assistant_id = build_docs_assistant(["api_docs.pdf", "architecture.md"])Function Calling in Assistants
Define tools the Assistant can invoke during a Run. The Run enters requires_action status, and you execute the function and submit results.
import json
import time
from openai import OpenAI
client = OpenAI()
# Assistant with a database lookup tool
assistant = client.beta.assistants.create(
name="Database Assistant",
instructions="Help users query the database. Use the run_query tool for data.",
model="gpt-4o",
tools=[
{
"type": "function",
"function": {
"name": "run_query",
"description": "Execute a read-only SQL query",
"parameters": {
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SQL SELECT statement"},
},
"required": ["sql"],
},
},
}
],
)
def handle_run(thread_id: str, run_id: str) -> str:
"""Poll a run, handling tool calls until completion."""
while True:
run = client.beta.threads.runs.retrieve(
thread_id=thread_id, run_id=run_id
)
if run.status == "completed":
messages = client.beta.threads.messages.list(thread_id=thread_id)
return messages.data[0].content[0].text.value
elif run.status == "requires_action":
tool_outputs = []
for tc in run.required_action.submit_tool_outputs.tool_calls:
args = json.loads(tc.function.arguments)
if tc.function.name == "run_query":
# Execute real query here
result = f"Query result for: {args['sql']}"
tool_outputs.append({
"tool_call_id": tc.id,
"output": result,
})
client.beta.threads.runs.submit_tool_outputs(
thread_id=thread_id,
run_id=run_id,
tool_outputs=tool_outputs,
)
elif run.status in ("failed", "cancelled", "expired"):
raise RuntimeError(f"Run ended with status: {run.status}")
time.sleep(1)Common Mistakes
- Creating a new Assistant for every request — Assistants are persistent configurations, create them once
- Not saving Thread IDs — losing a thread ID loses the conversation history
- Polling in a tight loop without sleep — causes unnecessary API calls and rate limit hits
- Not handling
requires_actionstatus — Run hangs indefinitely without submitting tool outputs - Uploading files repeatedly — Vector Stores persist; upload once and reuse the store ID
Best Practices
- Store Assistant IDs and Vector Store IDs in your database or environment variables
- Use streaming Runs (
client.beta.threads.runs.stream()) for responsive UIs instead of polling - Set
max_prompt_tokensandmax_completion_tokenson Runs to control cost - Monitor run status transitions — log
failedruns with theirlast_errorfor debugging - Clean up Threads after conversations complete to avoid accumulating storage costs
Key Takeaways
- The Assistants API provides persistent Threads, automatic context management, and multi-step tool execution that you would otherwise build yourself
- Runs move through status transitions:
queued→in_progress→completed(orrequires_actionfor tool calls) - File Search with Vector Stores enables RAG (retrieval-augmented generation) without building a vector database
- Function calling in Assistants requires polling for
requires_actionstatus and submitting tool outputs to continue the Run - Assistants, Threads, and Vector Stores all persist — create them once and reuse their IDs
- Streaming Runs reduce perceived latency for user-facing applications compared to polling
- Code Interpreter runs Python in an OpenAI-managed sandbox — useful for data analysis and computation tasks
- Track Run costs by logging
usage.total_tokensfrom completed Run objects
Advertisement