OpenAI Assistants API — Build Stateful AI Agents 2026

Sanjeev SharmaSanjeev Sharma
5 min read

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 sessions

Running 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_action status — 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_tokens and max_completion_tokens on Runs to control cost
  • Monitor run status transitions — log failed runs with their last_error for 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: queuedin_progresscompleted (or requires_action for 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_action status 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_tokens from completed Run objects

Advertisement

Sanjeev Sharma

Written by

Sanjeev Sharma

Full Stack Engineer · E-mopro

Related reading