Build an AI Customer Support Agent with OpenAI Assistants API 2026

The OpenAI Assistants API provides managed state, built-in tools (file search, code interpreter), and persistent threads—removing most of the infrastructure complexity from building AI agents. This tutorial builds a complete customer support system.

What We're Building

A customer support AI that:

Answers questions from your knowledge base (FAQ, docs)

Looks up order status via a custom function

Generates support tickets

Maintains conversation context across sessions

Setup

python
from openai import OpenAI
import json
import time
client = OpenAI()

Step 1: Create the Knowledge Base

Upload your documentation as files:

python
Upload FAQ and documentation files
files = []
for filename in ["faq.pdf", "product-docs.pdf", "return-policy.txt"]:
    with open(filename, "rb") as f:
        uploaded_file = client.files.create(
            file=f,
            purpose="assistants"
        )
        files.append(uploaded_file.id)
        print(f"Uploaded {filename}: {uploaded_file.id}")
Create a vector store from the files
vector_store = client.beta.vector_stores.create(
    name="Support Knowledge Base",
    file_ids=files
)
Wait for processing
while vector_store.status == "in_progress":
    time.sleep(2)
    vector_store = client.beta.vector_stores.retrieve(vector_store.id)
print(f"Vector store ready: {vector_store.id}")

Step 2: Create the Assistant

python
assistant = client.beta.assistants.create(
    name="TechCorp Support Agent",
    instructions="""You are a helpful customer support agent for TechCorp.
    
    Guidelines:
    - Be friendly, professional, and empathetic
    - Always search the knowledge base before answering
    - For order inquiries, use the get_order_status function
    - If you cannot resolve an issue, offer to create a support ticket
    - Never make up information you don't have
    - Keep responses concise but complete""",
    model="gpt-4o",
    tools=[
        {"type": "file_search"},  # RAG over uploaded files
        {
            "type": "function",
            "function": {
                "name": "get_order_status",
                "description": "Retrieve order status and tracking information",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "order_id": {
                            "type": "string",
                            "description": "The order ID (e.g., ORD-12345)"
                        }
                    },
                    "required": ["order_id"]
                }
            }
        },
        {
            "type": "function",
            "function": {
                "name": "create_support_ticket",
                "description": "Create a support ticket for unresolved issues",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "subject": {"type": "string"},
                        "description": {"type": "string"},
                        "priority": {
                            "type": "string",
                            "enum": ["low", "medium", "high", "urgent"]
                        }
                    },
                    "required": ["subject", "description", "priority"]
                }
            }
        }
    ],
    tool_resources={
        "file_search": {
            "vector_store_ids": [vector_store.id]
        }
    }
)
print(f"Assistant created: {assistant.id}")

Step 3: Implement Tool Handlers

python
def handle_tool_call(tool_name: str, args: dict) -> str:
    if tool_name == "get_order_status":
        # Connect to your actual order management system
        order_id = args["order_id"]
        # Simulated response:
        orders_db = {
            "ORD-12345": {"status": "Shipped", "tracking": "1Z999AA10123456784", "eta": "2026-01-15"},
            "ORD-67890": {"status": "Processing", "tracking": None, "eta": "2026-01-18"}
        }
        
        if order_id in orders_db:
            order = orders_db[order_id]
            return json.dumps(order)
        else:
            return json.dumps({"error": f"Order {order_id} not found"})
    
    elif tool_name == "create_support_ticket":
        ticket_id = f"TICKET-{int(time.time())}"
        # Save to your ticketing system (Zendesk, Jira, etc.)
        print(f"Creating ticket: {args['subject']} | Priority: {args['priority']}")
        return json.dumps({"ticket_id": ticket_id, "status": "created"})
    
    return json.dumps({"error": "Unknown tool"})

Step 4: Build the Conversation Loop

python
def run_support_conversation(user_id: str, user_message: str, thread_id: str = None):
    # Create or retrieve thread (threads persist conversation history)
    if thread_id is None:
        thread = client.beta.threads.create()
        thread_id = thread.id
        print(f"New conversation: {thread_id}")
    
    # Add user message to thread
    client.beta.threads.messages.create(
        thread_id=thread_id,
        role="user",
        content=user_message
    )
    
    # Run the assistant
    run = client.beta.threads.runs.create(
        thread_id=thread_id,
        assistant_id=assistant.id
    )
    
    # Poll for completion
    while True:
        run = client.beta.threads.runs.retrieve(
            thread_id=thread_id,
            run_id=run.id
        )
        
        if run.status == "completed":
            break
        
        elif run.status == "requires_action":
            # Handle tool calls
            tool_outputs = []
            for tool_call in run.required_action.submit_tool_outputs.tool_calls:
                args = json.loads(tool_call.function.arguments)
                output = handle_tool_call(tool_call.function.name, args)
                tool_outputs.append({
                    "tool_call_id": tool_call.id,
                    "output": output
                })
            
            run = 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 Exception(f"Run failed with status: {run.status}")
        
        time.sleep(0.5)
    
    # Get the assistant's response
    messages = client.beta.threads.messages.list(thread_id=thread_id, order="desc", limit=1)
    response_text = messages.data[0].content[0].text.value
    
    return {"response": response_text, "thread_id": thread_id}
Example usage
result = run_support_conversation(
    user_id="user-123",
    user_message="What's the status of my order ORD-12345?"
)
print(result["response"])
Continue conversation in same thread
result2 = run_support_conversation(
    user_id="user-123",
    user_message="When will it arrive?",
    thread_id=result["thread_id"]  # Continue same conversation!
)
print(result2["response"])

Step 5: Deploy as a REST API

python
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional
import uvicorn
app = FastAPI()
class ChatRequest(BaseModel):
    user_id: str
    message: str
    thread_id: Optional[str] = None
class ChatResponse(BaseModel):
    response: str
    thread_id: str
@app.post("/support/chat", response_model=ChatResponse)
async def support_chat(request: ChatRequest):
    try:
        result = run_support_conversation(
            user_id=request.user_id,
            user_message=request.message,
            thread_id=request.thread_id
        )
        return ChatResponse(**result)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Production Considerations

Thread Management

python
Store thread IDs per user session
from redis import Redis
redis = Redis(host="localhost", port=6379, decode_responses=True)
def get_or_create_thread(user_id: str, session_id: str) -> str:
    key = f"thread:{user_id}:{session_id}"
    thread_id = redis.get(key)
    
    if not thread_id:
        thread = client.beta.threads.create()
        redis.setex(key, 3600, thread.id)  # Expire after 1 hour
        return thread.id
    
    return thread_id

Cost Estimation

For a 1000-message/day support system:

gpt-4o: ~$15-30/day

gpt-4o-mini: ~$1.50-3/day

Consider gpt-4o-mini for simple FAQ queries, routing to gpt-4o for complex issues.

Results and Metrics

Companies using OpenAI Assistants API for customer support report:

65% reduction in tickets escalated to humans

<3 second average response time

87% customer satisfaction score

70% cost reduction vs. traditional support staffing

Conclusion

The OpenAI Assistants API makes it surprisingly straightforward to build production-quality AI support agents. With built-in file search, persistent threads, and tool calling, you get capabilities that would take weeks to build from scratch. The key is uploading comprehensive, well-organized documentation and designing clear, specific tool functions.